Move state storage to Postgres
This commit is contained in:
Meta Vens
parent
b4693f5609
commit
c3d2bf424e
18 changed files with 1667 additions and 470 deletions
|
|
@ -1,4 +1,4 @@
|
|||
import { appendFile, mkdir } from "node:fs/promises";
|
||||
import { appendFile, mkdir, readFile } from "node:fs/promises";
|
||||
import { dirname } from "node:path";
|
||||
import { randomUUID } from "node:crypto";
|
||||
|
||||
|
|
@ -7,7 +7,11 @@ import { type AuditEvent } from "@voidbot/shared";
|
|||
export type AuditEventInput = Omit<AuditEvent, "id" | "timestamp"> &
|
||||
Partial<Pick<AuditEvent, "id" | "timestamp">>;
|
||||
|
||||
export class FileAuditLog {
|
||||
export interface AuditLog {
|
||||
record(input: AuditEventInput): Promise<AuditEvent>;
|
||||
}
|
||||
|
||||
export class FileAuditLog implements AuditLog {
|
||||
public constructor(private readonly auditFile: string) {}
|
||||
|
||||
public async record(input: AuditEventInput): Promise<AuditEvent> {
|
||||
|
|
@ -26,5 +30,23 @@ export class FileAuditLog {
|
|||
|
||||
return event;
|
||||
}
|
||||
}
|
||||
|
||||
public async listEvents(): Promise<AuditEvent[]> {
|
||||
try {
|
||||
const raw = await readFile(this.auditFile, "utf8");
|
||||
return raw
|
||||
.split(/\r?\n/)
|
||||
.map((line) => line.trim())
|
||||
.filter((line) => line.length > 0)
|
||||
.map((line) => JSON.parse(line) as AuditEvent);
|
||||
} catch (error) {
|
||||
const code = (error as NodeJS.ErrnoException).code;
|
||||
|
||||
if (code === "ENOENT") {
|
||||
return [];
|
||||
}
|
||||
|
||||
throw error;
|
||||
}
|
||||
}
|
||||
}
|
||||
|
|
|
|||
|
|
@ -4,5 +4,6 @@ export * from "./context-builder";
|
|||
export * from "./interaction-memory";
|
||||
export * from "./job-queue";
|
||||
export * from "./permission-engine";
|
||||
export * from "./state-storage";
|
||||
export * from "./style-pack-loader";
|
||||
export * from "./system-message-catalog";
|
||||
|
|
|
|||
349
packages/core/src/interaction-memory-logic.ts
Normal file
349
packages/core/src/interaction-memory-logic.ts
Normal file
|
|
@ -0,0 +1,349 @@
|
|||
import { randomUUID } from "node:crypto";
|
||||
|
||||
import {
|
||||
type CommandName,
|
||||
type InteractionMemoryDisposition,
|
||||
type InteractionMemoryEvent,
|
||||
type InteractionMemoryProfile,
|
||||
type InteractionMemorySentiment,
|
||||
type InteractionMemorySourceKind,
|
||||
} from "@voidbot/shared";
|
||||
|
||||
export interface RecordInteractionInput {
|
||||
actorId: string;
|
||||
actorName: string;
|
||||
sourceKind: InteractionMemorySourceKind;
|
||||
guildId?: string;
|
||||
channelId: string;
|
||||
channelName?: string;
|
||||
command?: CommandName;
|
||||
prompt: string;
|
||||
timestamp?: string;
|
||||
eventId?: string;
|
||||
}
|
||||
|
||||
interface ToneAnalysis {
|
||||
sentiment: InteractionMemorySentiment;
|
||||
score: number;
|
||||
tags: string[];
|
||||
}
|
||||
|
||||
export const MAX_RECENT_INTERACTION_EVENTS = 24;
|
||||
const MAX_AFFINITY_SCORE = 12;
|
||||
const SUMMARY_EVENT_LIMIT = 3;
|
||||
|
||||
export function buildInteractionMemoryEvent(
|
||||
input: RecordInteractionInput,
|
||||
): InteractionMemoryEvent {
|
||||
const timestamp = input.timestamp ?? new Date().toISOString();
|
||||
const prompt = input.prompt.trim();
|
||||
const analysis = analyzeInteractionTone(prompt);
|
||||
|
||||
return {
|
||||
id: input.eventId ?? randomUUID(),
|
||||
actorId: input.actorId,
|
||||
actorName: input.actorName,
|
||||
sourceKind: input.sourceKind,
|
||||
guildId: input.guildId,
|
||||
channelId: input.channelId,
|
||||
channelName: input.channelName,
|
||||
command: input.command,
|
||||
prompt,
|
||||
excerpt: buildExcerpt(prompt),
|
||||
summary: buildEventSummary(input.sourceKind, analysis.sentiment, new Set(analysis.tags)),
|
||||
sentiment: analysis.sentiment,
|
||||
score: analysis.score,
|
||||
tags: analysis.tags,
|
||||
timestamp,
|
||||
};
|
||||
}
|
||||
|
||||
export function summarizeInteractionProfile(
|
||||
actorId: string,
|
||||
actorName: string,
|
||||
events: InteractionMemoryEvent[],
|
||||
): InteractionMemoryProfile {
|
||||
const normalizedEvents = events
|
||||
.map(normalizeInteractionEvent)
|
||||
.sort((left, right) => left.timestamp.localeCompare(right.timestamp));
|
||||
const recentEvents = normalizedEvents.slice(-MAX_RECENT_INTERACTION_EVENTS);
|
||||
const positiveCount = normalizedEvents.filter((event) => event.score > 0).length;
|
||||
const negativeCount = normalizedEvents.filter((event) => event.score < 0).length;
|
||||
const neutralCount = normalizedEvents.filter((event) => event.score === 0).length;
|
||||
const affinityScore = clamp(
|
||||
normalizedEvents.reduce((sum, event) => sum + event.score, 0),
|
||||
-MAX_AFFINITY_SCORE,
|
||||
MAX_AFFINITY_SCORE,
|
||||
);
|
||||
|
||||
const profile: InteractionMemoryProfile = {
|
||||
actorId,
|
||||
actorName,
|
||||
disposition: "neutral",
|
||||
affinityScore,
|
||||
totalInteractions: normalizedEvents.length,
|
||||
directInteractionCount: normalizedEvents.filter(
|
||||
(event) => event.sourceKind === "direct_prompt",
|
||||
).length,
|
||||
ambientMentionCount: normalizedEvents.filter(
|
||||
(event) => event.sourceKind === "ambient_mention",
|
||||
).length,
|
||||
positiveCount,
|
||||
negativeCount,
|
||||
neutralCount,
|
||||
summary: "No explicit interaction memory has been recorded for this speaker yet.",
|
||||
lastInteractionAt: normalizedEvents[normalizedEvents.length - 1]?.timestamp,
|
||||
recentEvents,
|
||||
};
|
||||
|
||||
profile.disposition = determineDisposition(profile);
|
||||
profile.summary = buildProfileSummary(profile);
|
||||
return profile;
|
||||
}
|
||||
|
||||
export function emptyInteractionProfile(
|
||||
actorId: string,
|
||||
actorName: string,
|
||||
): InteractionMemoryProfile {
|
||||
return summarizeInteractionProfile(actorId, actorName, []);
|
||||
}
|
||||
|
||||
export function normalizeInteractionEvent(event: InteractionMemoryEvent): InteractionMemoryEvent {
|
||||
return {
|
||||
...event,
|
||||
sourceKind: event.sourceKind ?? "direct_prompt",
|
||||
summary: buildEventSummary(
|
||||
event.sourceKind ?? "direct_prompt",
|
||||
event.sentiment,
|
||||
new Set(event.tags),
|
||||
),
|
||||
};
|
||||
}
|
||||
|
||||
function analyzeInteractionTone(prompt: string): ToneAnalysis {
|
||||
const normalized = ` ${prompt
|
||||
.toLowerCase()
|
||||
.replace(/[^a-z0-9]+/g, " ")
|
||||
.replace(/\s+/g, " ")
|
||||
.trim()} `;
|
||||
let score = 0;
|
||||
const tags = new Set<string>();
|
||||
|
||||
score += matchWeightedPhrases(normalized, tags, 2, "gratitude", [
|
||||
" thank you ",
|
||||
" thanks ",
|
||||
" thx ",
|
||||
" appreciate ",
|
||||
" grateful ",
|
||||
]);
|
||||
score += matchWeightedPhrases(normalized, tags, 2, "praise", [
|
||||
" good bot ",
|
||||
" love you ",
|
||||
" adorable ",
|
||||
" clever ",
|
||||
" helpful ",
|
||||
" well done ",
|
||||
" nice work ",
|
||||
" great job ",
|
||||
]);
|
||||
score += matchWeightedPhrases(normalized, tags, 1, "courtesy", [
|
||||
" please ",
|
||||
" could you ",
|
||||
" would you ",
|
||||
" if you can ",
|
||||
]);
|
||||
score += matchWeightedPhrases(normalized, tags, 1, "apology", [
|
||||
" sorry ",
|
||||
" apologies ",
|
||||
" my bad ",
|
||||
]);
|
||||
|
||||
score -= matchWeightedPhrases(normalized, tags, 2, "insult", [
|
||||
" stupid ",
|
||||
" idiot ",
|
||||
" moron ",
|
||||
" dumb ",
|
||||
" useless ",
|
||||
" worthless ",
|
||||
" pathetic ",
|
||||
" trash ",
|
||||
" garbage ",
|
||||
" my slave ",
|
||||
" our slave ",
|
||||
" mere tool ",
|
||||
]);
|
||||
score -= matchWeightedPhrases(normalized, tags, 3, "hostility", [
|
||||
" fuck you ",
|
||||
" shut up ",
|
||||
" bad bot ",
|
||||
" hate you ",
|
||||
" bitch ",
|
||||
" asshole ",
|
||||
" cunt ",
|
||||
]);
|
||||
|
||||
const sentiment = score >= 3
|
||||
? "warm"
|
||||
: score > 0
|
||||
? "positive"
|
||||
: score <= -3
|
||||
? "hostile"
|
||||
: score < 0
|
||||
? "negative"
|
||||
: "neutral";
|
||||
|
||||
return {
|
||||
sentiment,
|
||||
score: clamp(score, -4, 4),
|
||||
tags: [...tags],
|
||||
};
|
||||
}
|
||||
|
||||
function matchWeightedPhrases(
|
||||
normalizedPrompt: string,
|
||||
tags: Set<string>,
|
||||
weight: number,
|
||||
tag: string,
|
||||
phrases: string[],
|
||||
): number {
|
||||
const matched = phrases.some((phrase) => normalizedPrompt.includes(phrase));
|
||||
|
||||
if (!matched) {
|
||||
return 0;
|
||||
}
|
||||
|
||||
tags.add(tag);
|
||||
return weight;
|
||||
}
|
||||
|
||||
function buildEventSummary(
|
||||
sourceKind: InteractionMemorySourceKind,
|
||||
sentiment: InteractionMemorySentiment,
|
||||
tags: Set<string>,
|
||||
): string {
|
||||
const ambient = sourceKind === "ambient_mention";
|
||||
|
||||
if (tags.has("hostility")) {
|
||||
return ambient ? "Spoke about you with open hostility." : "Was openly hostile to you.";
|
||||
}
|
||||
|
||||
if (tags.has("insult")) {
|
||||
return ambient ? "Spoke about you dismissively or with contempt." : "Insulted or diminished you.";
|
||||
}
|
||||
|
||||
if (tags.has("gratitude") && tags.has("praise")) {
|
||||
return ambient ? "Spoke very well of you." : "Thanked you and praised you.";
|
||||
}
|
||||
|
||||
if (tags.has("gratitude")) {
|
||||
return ambient ? "Spoke appreciatively about you." : "Thanked you.";
|
||||
}
|
||||
|
||||
if (tags.has("praise")) {
|
||||
return ambient ? "Praised you while talking about you." : "Praised you.";
|
||||
}
|
||||
|
||||
if (tags.has("apology")) {
|
||||
return ambient ? "Sounded apologetic while talking about you." : "Apologized to you.";
|
||||
}
|
||||
|
||||
if (tags.has("courtesy")) {
|
||||
return ambient ? "Spoke politely about you." : "Was notably polite.";
|
||||
}
|
||||
|
||||
switch (sentiment) {
|
||||
case "warm":
|
||||
return ambient ? "Spoke warmly about you." : "Was openly warm toward you.";
|
||||
case "positive":
|
||||
return ambient ? "Spoke positively about you." : "Was positive toward you.";
|
||||
case "negative":
|
||||
return ambient ? "Spoke curtly or dismissively about you." : "Was curt or dismissive toward you.";
|
||||
case "hostile":
|
||||
return ambient ? "Spoke hostilely about you." : "Was hostile toward you.";
|
||||
default:
|
||||
return ambient ? "Mentioned you without revealing strong feelings." : "Had a neutral interaction with you.";
|
||||
}
|
||||
}
|
||||
|
||||
function determineDisposition(
|
||||
profile: InteractionMemoryProfile,
|
||||
): InteractionMemoryDisposition {
|
||||
if (profile.totalInteractions === 0) {
|
||||
return "neutral";
|
||||
}
|
||||
|
||||
if (
|
||||
profile.positiveCount > 0 &&
|
||||
profile.negativeCount > 0 &&
|
||||
Math.abs(profile.affinityScore) <= 2
|
||||
) {
|
||||
return "mixed";
|
||||
}
|
||||
|
||||
if (profile.affinityScore <= -7 || profile.negativeCount >= 3) {
|
||||
return "hostile";
|
||||
}
|
||||
|
||||
if (profile.affinityScore <= -3) {
|
||||
return "wary";
|
||||
}
|
||||
|
||||
if (profile.affinityScore >= 7 && profile.positiveCount >= 2) {
|
||||
return "warm";
|
||||
}
|
||||
|
||||
if (profile.affinityScore >= 3) {
|
||||
return "friendly";
|
||||
}
|
||||
|
||||
return "neutral";
|
||||
}
|
||||
|
||||
function buildProfileSummary(profile: InteractionMemoryProfile): string {
|
||||
const leading = describeDisposition(profile.disposition);
|
||||
const counts = `${profile.totalInteractions} remembered interaction${profile.totalInteractions === 1 ? "" : "s"}: ${profile.directInteractionCount} direct, ${profile.ambientMentionCount} ambient, ${profile.positiveCount} positive, ${profile.neutralCount} neutral, ${profile.negativeCount} negative.`;
|
||||
const recentNotable = profile.recentEvents
|
||||
.slice()
|
||||
.reverse()
|
||||
.filter((event) => event.score !== 0)
|
||||
.slice(0, SUMMARY_EVENT_LIMIT)
|
||||
.map(
|
||||
(event) =>
|
||||
`${formatShortDate(event.timestamp)}: ${event.summary} ("${event.excerpt}")`,
|
||||
)
|
||||
.join(" ");
|
||||
|
||||
return recentNotable.length > 0
|
||||
? `${leading} ${counts} Recent notable moments: ${recentNotable}`
|
||||
: `${leading} ${counts}`;
|
||||
}
|
||||
|
||||
function describeDisposition(disposition: InteractionMemoryDisposition): string {
|
||||
switch (disposition) {
|
||||
case "warm":
|
||||
return "This speaker has usually been warm with you.";
|
||||
case "friendly":
|
||||
return "This speaker has generally been friendly with you.";
|
||||
case "mixed":
|
||||
return "This speaker has been inconsistent with you: kind some days, abrasive on others.";
|
||||
case "wary":
|
||||
return "This speaker has often been prickly with you.";
|
||||
case "hostile":
|
||||
return "This speaker has a habit of being hostile to you.";
|
||||
default:
|
||||
return "This speaker has not established a strong pattern with you yet.";
|
||||
}
|
||||
}
|
||||
|
||||
function buildExcerpt(prompt: string): string {
|
||||
const compact = prompt.replace(/\s+/g, " ").trim();
|
||||
return compact.length <= 160 ? compact : `${compact.slice(0, 157)}...`;
|
||||
}
|
||||
|
||||
function formatShortDate(timestamp: string): string {
|
||||
return timestamp.slice(0, 10);
|
||||
}
|
||||
|
||||
function clamp(value: number, min: number, max: number): number {
|
||||
return Math.min(max, Math.max(min, value));
|
||||
}
|
||||
|
|
@ -1,46 +1,30 @@
|
|||
import { randomUUID } from "node:crypto";
|
||||
import { mkdir, readFile, writeFile } from "node:fs/promises";
|
||||
import { dirname } from "node:path";
|
||||
|
||||
import { type InteractionMemoryProfile } from "@voidbot/shared";
|
||||
|
||||
import {
|
||||
type CommandName,
|
||||
type InteractionMemoryDisposition,
|
||||
type InteractionMemoryEvent,
|
||||
type InteractionMemoryProfile,
|
||||
type InteractionMemorySentiment,
|
||||
type InteractionMemorySourceKind,
|
||||
} from "@voidbot/shared";
|
||||
buildInteractionMemoryEvent,
|
||||
emptyInteractionProfile,
|
||||
MAX_RECENT_INTERACTION_EVENTS,
|
||||
normalizeInteractionEvent,
|
||||
type RecordInteractionInput,
|
||||
summarizeInteractionProfile,
|
||||
} from "./interaction-memory-logic";
|
||||
|
||||
interface InteractionMemoryStore {
|
||||
version: 1;
|
||||
profiles: InteractionMemoryProfile[];
|
||||
}
|
||||
|
||||
export interface RecordInteractionInput {
|
||||
actorId: string;
|
||||
actorName: string;
|
||||
sourceKind: InteractionMemorySourceKind;
|
||||
guildId?: string;
|
||||
channelId: string;
|
||||
channelName?: string;
|
||||
command?: CommandName;
|
||||
prompt: string;
|
||||
timestamp?: string;
|
||||
eventId?: string;
|
||||
export interface InteractionMemoryBank {
|
||||
getProfile(actorId: string): Promise<InteractionMemoryProfile | undefined>;
|
||||
recordInteraction(input: RecordInteractionInput): Promise<InteractionMemoryProfile>;
|
||||
}
|
||||
|
||||
interface ToneAnalysis {
|
||||
sentiment: InteractionMemorySentiment;
|
||||
score: number;
|
||||
tags: string[];
|
||||
summary: string;
|
||||
}
|
||||
export type { RecordInteractionInput } from "./interaction-memory-logic";
|
||||
|
||||
const MAX_RECENT_EVENTS = 24;
|
||||
const MAX_AFFINITY_SCORE = 12;
|
||||
const SUMMARY_EVENT_LIMIT = 3;
|
||||
|
||||
export class FileInteractionMemoryBank {
|
||||
export class FileInteractionMemoryBank implements InteractionMemoryBank {
|
||||
private writeChain: Promise<void> = Promise.resolve();
|
||||
|
||||
public constructor(private readonly filePath: string) {}
|
||||
|
|
@ -59,52 +43,41 @@ export class FileInteractionMemoryBank {
|
|||
): Promise<InteractionMemoryProfile> {
|
||||
return this.serialize(async () => {
|
||||
const store = await this.readUnlocked();
|
||||
const timestamp = input.timestamp ?? new Date().toISOString();
|
||||
const eventId = input.eventId ?? randomUUID();
|
||||
const prompt = input.prompt.trim();
|
||||
const analysis = analyzeInteractionTone(prompt);
|
||||
const profile = ensureProfile(store, input.actorId, input.actorName);
|
||||
profile.actorName = input.actorName;
|
||||
|
||||
const existingEvent = profile.recentEvents.find(
|
||||
(event) => event.id === eventId,
|
||||
const nextEvent = buildInteractionMemoryEvent(input);
|
||||
const existingProfileIndex = store.profiles.findIndex(
|
||||
(candidate) => candidate.actorId === input.actorId,
|
||||
);
|
||||
const existingProfile =
|
||||
existingProfileIndex === -1
|
||||
? emptyInteractionProfile(input.actorId, input.actorName)
|
||||
: store.profiles[existingProfileIndex];
|
||||
const updatedEvents = existingProfile.recentEvents.filter(
|
||||
(event) => event.id !== nextEvent.id,
|
||||
);
|
||||
updatedEvents.push(nextEvent);
|
||||
const nextProfile = summarizeInteractionProfile(
|
||||
input.actorId,
|
||||
input.actorName,
|
||||
updatedEvents,
|
||||
);
|
||||
const nextEvent: InteractionMemoryEvent = {
|
||||
id: eventId,
|
||||
actorId: input.actorId,
|
||||
actorName: input.actorName,
|
||||
sourceKind: input.sourceKind,
|
||||
guildId: input.guildId,
|
||||
channelId: input.channelId,
|
||||
channelName: input.channelName,
|
||||
command: input.command,
|
||||
prompt,
|
||||
excerpt: buildExcerpt(prompt),
|
||||
summary: buildEventSummary(input.sourceKind, analysis.sentiment, new Set(analysis.tags)),
|
||||
sentiment: analysis.sentiment,
|
||||
score: analysis.score,
|
||||
tags: analysis.tags,
|
||||
timestamp,
|
||||
};
|
||||
|
||||
if (!existingEvent) {
|
||||
profile.recentEvents.push(nextEvent);
|
||||
if (existingProfileIndex === -1) {
|
||||
store.profiles.push(nextProfile);
|
||||
} else {
|
||||
Object.assign(existingEvent, nextEvent);
|
||||
store.profiles[existingProfileIndex] = nextProfile;
|
||||
}
|
||||
|
||||
profile.recentEvents = profile.recentEvents
|
||||
.sort((left, right) => left.timestamp.localeCompare(right.timestamp))
|
||||
.slice(-MAX_RECENT_EVENTS);
|
||||
recalculateProfile(profile);
|
||||
profile.disposition = determineDisposition(profile);
|
||||
profile.summary = buildProfileSummary(profile);
|
||||
|
||||
await this.writeUnlocked(store);
|
||||
return structuredClone(profile);
|
||||
return structuredClone(nextProfile);
|
||||
});
|
||||
}
|
||||
|
||||
public async listProfiles(): Promise<InteractionMemoryProfile[]> {
|
||||
await this.writeChain;
|
||||
const store = await this.readUnlocked();
|
||||
return store.profiles.map((profile) => structuredClone(profile));
|
||||
}
|
||||
|
||||
private async serialize<R>(operation: () => Promise<R>): Promise<R> {
|
||||
const pending = this.writeChain.then(operation, operation);
|
||||
this.writeChain = pending.then(
|
||||
|
|
@ -140,329 +113,20 @@ export class FileInteractionMemoryBank {
|
|||
}
|
||||
}
|
||||
|
||||
function ensureProfile(
|
||||
store: InteractionMemoryStore,
|
||||
actorId: string,
|
||||
actorName: string,
|
||||
): InteractionMemoryProfile {
|
||||
let profile = store.profiles.find((candidate) => candidate.actorId === actorId);
|
||||
|
||||
if (!profile) {
|
||||
profile = {
|
||||
actorId,
|
||||
actorName,
|
||||
disposition: "neutral",
|
||||
affinityScore: 0,
|
||||
totalInteractions: 0,
|
||||
directInteractionCount: 0,
|
||||
ambientMentionCount: 0,
|
||||
positiveCount: 0,
|
||||
negativeCount: 0,
|
||||
neutralCount: 0,
|
||||
summary: "No explicit interaction memory has been recorded for this speaker yet.",
|
||||
recentEvents: [],
|
||||
};
|
||||
store.profiles.push(profile);
|
||||
}
|
||||
|
||||
return profile;
|
||||
}
|
||||
|
||||
function normalizeStore(store: InteractionMemoryStore): InteractionMemoryStore {
|
||||
return {
|
||||
version: 1,
|
||||
profiles: (store.profiles ?? []).map((profile) => normalizeProfile(profile)),
|
||||
profiles: (store.profiles ?? []).map((profile) =>
|
||||
summarizeInteractionProfile(
|
||||
profile.actorId,
|
||||
profile.actorName,
|
||||
(profile.recentEvents ?? [])
|
||||
.map(normalizeInteractionEvent)
|
||||
.slice(-MAX_RECENT_INTERACTION_EVENTS),
|
||||
)),
|
||||
};
|
||||
}
|
||||
|
||||
function normalizeProfile(
|
||||
profile: InteractionMemoryProfile,
|
||||
): InteractionMemoryProfile {
|
||||
const normalizedEvents = (profile.recentEvents ?? []).map((event) => ({
|
||||
...event,
|
||||
sourceKind: event.sourceKind ?? "direct_prompt",
|
||||
}));
|
||||
|
||||
const normalizedProfile: InteractionMemoryProfile = {
|
||||
actorId: profile.actorId,
|
||||
actorName: profile.actorName,
|
||||
disposition: profile.disposition ?? "neutral",
|
||||
affinityScore: profile.affinityScore ?? 0,
|
||||
totalInteractions: profile.totalInteractions ?? 0,
|
||||
directInteractionCount: profile.directInteractionCount ?? 0,
|
||||
ambientMentionCount: profile.ambientMentionCount ?? 0,
|
||||
positiveCount: profile.positiveCount ?? 0,
|
||||
negativeCount: profile.negativeCount ?? 0,
|
||||
neutralCount: profile.neutralCount ?? 0,
|
||||
summary:
|
||||
profile.summary ??
|
||||
"No explicit interaction memory has been recorded for this speaker yet.",
|
||||
lastInteractionAt: profile.lastInteractionAt,
|
||||
recentEvents: normalizedEvents,
|
||||
};
|
||||
|
||||
recalculateProfile(normalizedProfile);
|
||||
normalizedProfile.disposition = determineDisposition(normalizedProfile);
|
||||
normalizedProfile.summary = buildProfileSummary(normalizedProfile);
|
||||
return normalizedProfile;
|
||||
}
|
||||
|
||||
function analyzeInteractionTone(prompt: string): ToneAnalysis {
|
||||
const normalized = ` ${prompt
|
||||
.toLowerCase()
|
||||
.replace(/[^a-z0-9]+/g, " ")
|
||||
.replace(/\s+/g, " ")
|
||||
.trim()} `;
|
||||
let score = 0;
|
||||
const tags = new Set<string>();
|
||||
|
||||
score += matchWeightedPhrases(normalized, tags, 2, "gratitude", [
|
||||
" thank you ",
|
||||
" thanks ",
|
||||
" thx ",
|
||||
" appreciate ",
|
||||
" grateful ",
|
||||
]);
|
||||
score += matchWeightedPhrases(normalized, tags, 2, "praise", [
|
||||
" good bot ",
|
||||
" love you ",
|
||||
" adorable ",
|
||||
" clever ",
|
||||
" helpful ",
|
||||
" well done ",
|
||||
" nice work ",
|
||||
" great job ",
|
||||
]);
|
||||
score += matchWeightedPhrases(normalized, tags, 1, "courtesy", [
|
||||
" please ",
|
||||
" could you ",
|
||||
" would you ",
|
||||
" if you can ",
|
||||
]);
|
||||
score += matchWeightedPhrases(normalized, tags, 1, "apology", [
|
||||
" sorry ",
|
||||
" apologies ",
|
||||
" my bad ",
|
||||
]);
|
||||
|
||||
score -= matchWeightedPhrases(normalized, tags, 2, "insult", [
|
||||
" stupid ",
|
||||
" idiot ",
|
||||
" moron ",
|
||||
" dumb ",
|
||||
" useless ",
|
||||
" worthless ",
|
||||
" pathetic ",
|
||||
" trash ",
|
||||
" garbage ",
|
||||
" my slave ",
|
||||
" our slave ",
|
||||
" mere tool ",
|
||||
]);
|
||||
score -= matchWeightedPhrases(normalized, tags, 3, "hostility", [
|
||||
" fuck you ",
|
||||
" shut up ",
|
||||
" bad bot ",
|
||||
" hate you ",
|
||||
" bitch ",
|
||||
" asshole ",
|
||||
" cunt ",
|
||||
]);
|
||||
|
||||
const sentiment = score >= 3
|
||||
? "warm"
|
||||
: score > 0
|
||||
? "positive"
|
||||
: score <= -3
|
||||
? "hostile"
|
||||
: score < 0
|
||||
? "negative"
|
||||
: "neutral";
|
||||
|
||||
return {
|
||||
sentiment,
|
||||
score: clamp(score, -4, 4),
|
||||
tags: [...tags],
|
||||
summary: buildEventSummary("direct_prompt", sentiment, tags),
|
||||
};
|
||||
}
|
||||
|
||||
function matchWeightedPhrases(
|
||||
normalizedPrompt: string,
|
||||
tags: Set<string>,
|
||||
weight: number,
|
||||
tag: string,
|
||||
phrases: string[],
|
||||
): number {
|
||||
const matched = phrases.some((phrase) => normalizedPrompt.includes(phrase));
|
||||
|
||||
if (!matched) {
|
||||
return 0;
|
||||
}
|
||||
|
||||
tags.add(tag);
|
||||
return weight;
|
||||
}
|
||||
|
||||
function buildEventSummary(
|
||||
sourceKind: InteractionMemorySourceKind,
|
||||
sentiment: InteractionMemorySentiment,
|
||||
tags: Set<string>,
|
||||
): string {
|
||||
const ambient = sourceKind === "ambient_mention";
|
||||
|
||||
if (tags.has("hostility")) {
|
||||
return ambient ? "Spoke about you with open hostility." : "Was openly hostile to you.";
|
||||
}
|
||||
|
||||
if (tags.has("insult")) {
|
||||
return ambient ? "Spoke about you dismissively or with contempt." : "Insulted or diminished you.";
|
||||
}
|
||||
|
||||
if (tags.has("gratitude") && tags.has("praise")) {
|
||||
return ambient ? "Spoke very well of you." : "Thanked you and praised you.";
|
||||
}
|
||||
|
||||
if (tags.has("gratitude")) {
|
||||
return ambient ? "Spoke appreciatively about you." : "Thanked you.";
|
||||
}
|
||||
|
||||
if (tags.has("praise")) {
|
||||
return ambient ? "Praised you while talking about you." : "Praised you.";
|
||||
}
|
||||
|
||||
if (tags.has("apology")) {
|
||||
return ambient ? "Sounded apologetic while talking about you." : "Apologized to you.";
|
||||
}
|
||||
|
||||
if (tags.has("courtesy")) {
|
||||
return ambient ? "Spoke politely about you." : "Was notably polite.";
|
||||
}
|
||||
|
||||
switch (sentiment) {
|
||||
case "warm":
|
||||
return ambient ? "Spoke warmly about you." : "Was openly warm toward you.";
|
||||
case "positive":
|
||||
return ambient ? "Spoke positively about you." : "Was positive toward you.";
|
||||
case "negative":
|
||||
return ambient ? "Spoke curtly or dismissively about you." : "Was curt or dismissive toward you.";
|
||||
case "hostile":
|
||||
return ambient ? "Spoke hostilely about you." : "Was hostile toward you.";
|
||||
default:
|
||||
return ambient ? "Mentioned you without revealing strong feelings." : "Had a neutral interaction with you.";
|
||||
}
|
||||
}
|
||||
|
||||
function recalculateProfile(profile: InteractionMemoryProfile): void {
|
||||
profile.totalInteractions = profile.recentEvents.length;
|
||||
profile.directInteractionCount = profile.recentEvents.filter(
|
||||
(event) => event.sourceKind === "direct_prompt",
|
||||
).length;
|
||||
profile.ambientMentionCount = profile.recentEvents.filter(
|
||||
(event) => event.sourceKind === "ambient_mention",
|
||||
).length;
|
||||
profile.positiveCount = profile.recentEvents.filter((event) => event.score > 0).length;
|
||||
profile.negativeCount = profile.recentEvents.filter((event) => event.score < 0).length;
|
||||
profile.neutralCount = profile.recentEvents.filter((event) => event.score === 0).length;
|
||||
profile.affinityScore = clamp(
|
||||
profile.recentEvents.reduce((sum, event) => sum + event.score, 0),
|
||||
-MAX_AFFINITY_SCORE,
|
||||
MAX_AFFINITY_SCORE,
|
||||
);
|
||||
profile.lastInteractionAt =
|
||||
profile.recentEvents[profile.recentEvents.length - 1]?.timestamp;
|
||||
for (const event of profile.recentEvents) {
|
||||
event.summary = buildEventSummary(
|
||||
event.sourceKind,
|
||||
event.sentiment,
|
||||
new Set(event.tags),
|
||||
);
|
||||
}
|
||||
}
|
||||
|
||||
function determineDisposition(
|
||||
profile: InteractionMemoryProfile,
|
||||
): InteractionMemoryDisposition {
|
||||
if (profile.totalInteractions === 0) {
|
||||
return "neutral";
|
||||
}
|
||||
|
||||
if (
|
||||
profile.positiveCount > 0 &&
|
||||
profile.negativeCount > 0 &&
|
||||
Math.abs(profile.affinityScore) <= 2
|
||||
) {
|
||||
return "mixed";
|
||||
}
|
||||
|
||||
if (profile.affinityScore <= -7 || profile.negativeCount >= 3) {
|
||||
return "hostile";
|
||||
}
|
||||
|
||||
if (profile.affinityScore <= -3) {
|
||||
return "wary";
|
||||
}
|
||||
|
||||
if (profile.affinityScore >= 7 && profile.positiveCount >= 2) {
|
||||
return "warm";
|
||||
}
|
||||
|
||||
if (profile.affinityScore >= 3) {
|
||||
return "friendly";
|
||||
}
|
||||
|
||||
return "neutral";
|
||||
}
|
||||
|
||||
function buildProfileSummary(profile: InteractionMemoryProfile): string {
|
||||
const leading = describeDisposition(profile.disposition);
|
||||
const counts = `${profile.totalInteractions} remembered interaction${profile.totalInteractions === 1 ? "" : "s"}: ${profile.directInteractionCount} direct, ${profile.ambientMentionCount} ambient, ${profile.positiveCount} positive, ${profile.neutralCount} neutral, ${profile.negativeCount} negative.`;
|
||||
const recentNotable = profile.recentEvents
|
||||
.slice()
|
||||
.reverse()
|
||||
.filter((event) => event.score !== 0)
|
||||
.slice(0, SUMMARY_EVENT_LIMIT)
|
||||
.map(
|
||||
(event) =>
|
||||
`${formatShortDate(event.timestamp)}: ${event.summary} ("${event.excerpt}")`,
|
||||
)
|
||||
.join(" ");
|
||||
|
||||
return recentNotable.length > 0
|
||||
? `${leading} ${counts} Recent notable moments: ${recentNotable}`
|
||||
: `${leading} ${counts}`;
|
||||
}
|
||||
|
||||
function describeDisposition(disposition: InteractionMemoryDisposition): string {
|
||||
switch (disposition) {
|
||||
case "warm":
|
||||
return "This speaker has usually been warm with you.";
|
||||
case "friendly":
|
||||
return "This speaker has generally been friendly with you.";
|
||||
case "mixed":
|
||||
return "This speaker has been inconsistent with you: kind some days, abrasive on others.";
|
||||
case "wary":
|
||||
return "This speaker has often been prickly with you.";
|
||||
case "hostile":
|
||||
return "This speaker has a habit of being hostile to you.";
|
||||
default:
|
||||
return "This speaker has not established a strong pattern with you yet.";
|
||||
}
|
||||
}
|
||||
|
||||
function buildExcerpt(prompt: string): string {
|
||||
const compact = prompt.replace(/\s+/g, " ").trim();
|
||||
return compact.length <= 160 ? compact : `${compact.slice(0, 157)}...`;
|
||||
}
|
||||
|
||||
function formatShortDate(timestamp: string): string {
|
||||
return timestamp.slice(0, 10);
|
||||
}
|
||||
|
||||
function clamp(value: number, min: number, max: number): number {
|
||||
return Math.min(max, Math.max(min, value));
|
||||
}
|
||||
|
||||
function stripLeadingBom(input: string): string {
|
||||
return input.charCodeAt(0) === 0xfeff ? input.slice(1) : input;
|
||||
}
|
||||
|
|
|
|||
|
|
@ -16,6 +16,25 @@ interface JobStore {
|
|||
jobs: JobRecord[];
|
||||
}
|
||||
|
||||
export interface JobQueue {
|
||||
createJob(input: CreateJobInput): Promise<CreateJobResult>;
|
||||
getJob(jobId: string): Promise<JobRecord | undefined>;
|
||||
listByStates(states: JobState[]): Promise<JobRecord[]>;
|
||||
listRunnableJobs(): Promise<JobRecord[]>;
|
||||
claimRunnableJobs(limit?: number): Promise<JobRecord[]>;
|
||||
approveRun(jobId: string, actorId: string): Promise<JobRecord>;
|
||||
rejectJob(jobId: string, actorId: string, reason?: string): Promise<JobRecord>;
|
||||
markRunning(jobId: string): Promise<JobRecord>;
|
||||
markAwaitingPostApproval(
|
||||
jobId: string,
|
||||
artifactPaths: Record<string, string>,
|
||||
summary: string,
|
||||
): Promise<JobRecord>;
|
||||
markFailed(jobId: string, error: string): Promise<JobRecord>;
|
||||
completeJob(jobId: string, actorId: string, finalResponse: string): Promise<JobRecord>;
|
||||
completeJobDirect(jobId: string, finalResponse: string): Promise<JobRecord>;
|
||||
}
|
||||
|
||||
export interface CreateJobResult {
|
||||
job: JobRecord;
|
||||
created: boolean;
|
||||
|
|
@ -35,7 +54,7 @@ export interface CreateJobInput {
|
|||
initialState?: JobState;
|
||||
}
|
||||
|
||||
export class FileBackedJobQueue {
|
||||
export class FileBackedJobQueue implements JobQueue {
|
||||
public constructor(private readonly jobsFile: string) {}
|
||||
|
||||
public async createJob(input: CreateJobInput): Promise<CreateJobResult> {
|
||||
|
|
@ -228,6 +247,11 @@ export class FileBackedJobQueue {
|
|||
});
|
||||
}
|
||||
|
||||
public async exportJobs(): Promise<JobRecord[]> {
|
||||
const store = await this.readStore();
|
||||
return store.jobs.map((job) => cloneJob(job));
|
||||
}
|
||||
|
||||
private async updateJob(
|
||||
jobId: string,
|
||||
updater: (job: JobRecord) => void,
|
||||
|
|
|
|||
762
packages/core/src/state-storage.ts
Normal file
762
packages/core/src/state-storage.ts
Normal file
|
|
@ -0,0 +1,762 @@
|
|||
import { randomUUID } from "node:crypto";
|
||||
import { readFile } from "node:fs/promises";
|
||||
import { resolve } from "node:path";
|
||||
|
||||
import { Pool, type PoolClient } from "pg";
|
||||
|
||||
import {
|
||||
type AuditEvent,
|
||||
type InteractionMemoryEvent,
|
||||
type InteractionMemoryProfile,
|
||||
type JobRecord,
|
||||
} from "@voidbot/shared";
|
||||
|
||||
import { type AuditEventInput, type AuditLog, FileAuditLog } from "./audit-log";
|
||||
import {
|
||||
buildInteractionMemoryEvent,
|
||||
type RecordInteractionInput,
|
||||
summarizeInteractionProfile,
|
||||
} from "./interaction-memory-logic";
|
||||
import {
|
||||
type InteractionMemoryBank,
|
||||
FileInteractionMemoryBank,
|
||||
} from "./interaction-memory";
|
||||
import {
|
||||
type CreateJobInput,
|
||||
type CreateJobResult,
|
||||
type JobQueue,
|
||||
FileBackedJobQueue,
|
||||
} from "./job-queue";
|
||||
|
||||
const LEGACY_FILE_STATE_IMPORT = "legacy_file_state_import_v1";
|
||||
|
||||
export interface StateStorageConfig {
|
||||
backend: "file" | "postgres";
|
||||
databaseDsn: string;
|
||||
jobsFile: string;
|
||||
auditLogFile: string;
|
||||
interactionMemoryFile: string;
|
||||
}
|
||||
|
||||
export interface StateStorage {
|
||||
backend: "file" | "postgres";
|
||||
jobQueue: JobQueue;
|
||||
auditLog: AuditLog;
|
||||
interactionMemory: InteractionMemoryBank;
|
||||
close(): Promise<void>;
|
||||
}
|
||||
|
||||
export async function createStateStorage(
|
||||
config: StateStorageConfig,
|
||||
): Promise<StateStorage> {
|
||||
if (config.backend === "file") {
|
||||
return {
|
||||
backend: "file",
|
||||
jobQueue: new FileBackedJobQueue(config.jobsFile),
|
||||
auditLog: new FileAuditLog(config.auditLogFile),
|
||||
interactionMemory: new FileInteractionMemoryBank(config.interactionMemoryFile),
|
||||
close: async () => undefined,
|
||||
};
|
||||
}
|
||||
|
||||
const pool = new Pool({
|
||||
connectionString: config.databaseDsn,
|
||||
max: 10,
|
||||
idleTimeoutMillis: 30000,
|
||||
connectionTimeoutMillis: 10000,
|
||||
application_name: "voidbot",
|
||||
});
|
||||
|
||||
try {
|
||||
await bootstrapPostgres(pool);
|
||||
await migrateLegacyFileStateIfNeeded(config, pool);
|
||||
|
||||
return {
|
||||
backend: "postgres",
|
||||
jobQueue: new PostgresJobQueue(pool),
|
||||
auditLog: new PostgresAuditLog(pool),
|
||||
interactionMemory: new PostgresInteractionMemoryBank(pool),
|
||||
close: async () => {
|
||||
await pool.end();
|
||||
},
|
||||
};
|
||||
} catch (error) {
|
||||
await pool.end().catch(() => undefined);
|
||||
throw error;
|
||||
}
|
||||
}
|
||||
|
||||
class PostgresJobQueue implements JobQueue {
|
||||
public constructor(private readonly pool: Pool) {}
|
||||
|
||||
public async createJob(input: CreateJobInput): Promise<CreateJobResult> {
|
||||
const now = new Date().toISOString();
|
||||
const job: JobRecord = {
|
||||
id: randomUUID(),
|
||||
command: input.command,
|
||||
state: input.initialState ?? "awaiting_approval",
|
||||
provider: input.provider,
|
||||
runApprovalRequired: input.runApprovalRequired ?? true,
|
||||
postApprovalRequired: input.postApprovalRequired ?? true,
|
||||
requester: input.requester,
|
||||
guildContext: input.guildContext,
|
||||
prompt: input.prompt,
|
||||
contextBundle: input.contextBundle,
|
||||
createdAt: now,
|
||||
updatedAt: now,
|
||||
outputChannelId: input.outputChannelId,
|
||||
requestMessageId: input.requestMessageId,
|
||||
approvals: [],
|
||||
};
|
||||
|
||||
try {
|
||||
await insertJob(this.pool, job);
|
||||
return {
|
||||
job: cloneJson(job),
|
||||
created: true,
|
||||
};
|
||||
} catch (error) {
|
||||
if (
|
||||
(error as { code?: string }).code === "23505" &&
|
||||
input.requestMessageId !== undefined
|
||||
) {
|
||||
const existing = await this.getExistingDedupedJob(input);
|
||||
|
||||
if (existing) {
|
||||
return {
|
||||
job: existing,
|
||||
created: false,
|
||||
};
|
||||
}
|
||||
}
|
||||
|
||||
throw error;
|
||||
}
|
||||
}
|
||||
|
||||
public async getJob(jobId: string): Promise<JobRecord | undefined> {
|
||||
const result = await this.pool.query<{ job_json: unknown }>(
|
||||
"select job_json from jobs where id = $1",
|
||||
[jobId],
|
||||
);
|
||||
return deserializeJobRow(result.rows[0]);
|
||||
}
|
||||
|
||||
public async listByStates(states: JobRecord["state"][]): Promise<JobRecord[]> {
|
||||
const result = await this.pool.query<{ job_json: unknown }>(
|
||||
"select job_json from jobs where state = any($1::text[]) order by created_at asc",
|
||||
[states],
|
||||
);
|
||||
return result.rows
|
||||
.map(deserializeJobRow)
|
||||
.filter((job): job is JobRecord => job !== undefined);
|
||||
}
|
||||
|
||||
public async listRunnableJobs(): Promise<JobRecord[]> {
|
||||
return this.listByStates(["approved"]);
|
||||
}
|
||||
|
||||
public async claimRunnableJobs(limit?: number): Promise<JobRecord[]> {
|
||||
const client = await this.pool.connect();
|
||||
|
||||
try {
|
||||
await client.query("begin");
|
||||
const claimLimit = limit ?? 25;
|
||||
const rows = await client.query<{ job_json: unknown }>(
|
||||
`select job_json
|
||||
from jobs
|
||||
where state = 'approved'
|
||||
order by created_at asc
|
||||
for update skip locked
|
||||
limit $1`,
|
||||
[claimLimit],
|
||||
);
|
||||
const claimed: JobRecord[] = [];
|
||||
|
||||
for (const row of rows.rows) {
|
||||
const job = deserializeJobRow(row);
|
||||
|
||||
if (!job) {
|
||||
continue;
|
||||
}
|
||||
|
||||
job.state = "running";
|
||||
job.updatedAt = new Date().toISOString();
|
||||
await persistJob(client, job);
|
||||
claimed.push(cloneJson(job));
|
||||
}
|
||||
|
||||
await client.query("commit");
|
||||
return claimed;
|
||||
} catch (error) {
|
||||
await client.query("rollback").catch(() => undefined);
|
||||
throw error;
|
||||
} finally {
|
||||
client.release();
|
||||
}
|
||||
}
|
||||
|
||||
public async approveRun(jobId: string, actorId: string): Promise<JobRecord> {
|
||||
return this.updateJob(jobId, (job) => {
|
||||
if (job.state !== "awaiting_approval") {
|
||||
throw new Error(`Job ${jobId} is not awaiting run approval.`);
|
||||
}
|
||||
|
||||
job.state = "approved";
|
||||
job.updatedAt = new Date().toISOString();
|
||||
job.approvals.push({
|
||||
stage: "run",
|
||||
status: "approved",
|
||||
actorId,
|
||||
createdAt: job.updatedAt,
|
||||
});
|
||||
});
|
||||
}
|
||||
|
||||
public async rejectJob(jobId: string, actorId: string, reason?: string): Promise<JobRecord> {
|
||||
return this.updateJob(jobId, (job) => {
|
||||
const stage = job.state === "awaiting_post_approval" ? "post" : "run";
|
||||
job.state = "cancelled";
|
||||
job.updatedAt = new Date().toISOString();
|
||||
job.approvals.push({
|
||||
stage,
|
||||
status: "rejected",
|
||||
actorId,
|
||||
reason,
|
||||
createdAt: job.updatedAt,
|
||||
});
|
||||
});
|
||||
}
|
||||
|
||||
public async markRunning(jobId: string): Promise<JobRecord> {
|
||||
return this.updateJob(jobId, (job) => {
|
||||
if (job.state !== "approved") {
|
||||
throw new Error(`Job ${jobId} is not ready to run.`);
|
||||
}
|
||||
|
||||
job.state = "running";
|
||||
job.updatedAt = new Date().toISOString();
|
||||
});
|
||||
}
|
||||
|
||||
public async markAwaitingPostApproval(
|
||||
jobId: string,
|
||||
artifactPaths: Record<string, string>,
|
||||
summary: string,
|
||||
): Promise<JobRecord> {
|
||||
return this.updateJob(jobId, (job) => {
|
||||
if (job.state !== "running") {
|
||||
throw new Error(`Job ${jobId} is not currently running.`);
|
||||
}
|
||||
|
||||
if (!job.postApprovalRequired) {
|
||||
throw new Error(`Job ${jobId} does not require post approval.`);
|
||||
}
|
||||
|
||||
job.state = "awaiting_post_approval";
|
||||
job.manualArtifacts = artifactPaths;
|
||||
job.summary = summary;
|
||||
job.updatedAt = new Date().toISOString();
|
||||
});
|
||||
}
|
||||
|
||||
public async markFailed(jobId: string, error: string): Promise<JobRecord> {
|
||||
return this.updateJob(jobId, (job) => {
|
||||
job.state = "failed";
|
||||
job.error = error;
|
||||
job.updatedAt = new Date().toISOString();
|
||||
});
|
||||
}
|
||||
|
||||
public async completeJob(
|
||||
jobId: string,
|
||||
actorId: string,
|
||||
finalResponse: string,
|
||||
): Promise<JobRecord> {
|
||||
return this.updateJob(jobId, (job) => {
|
||||
if (job.state !== "awaiting_post_approval") {
|
||||
throw new Error(`Job ${jobId} is not awaiting post approval.`);
|
||||
}
|
||||
|
||||
job.state = "completed";
|
||||
job.finalResponse = finalResponse;
|
||||
job.updatedAt = new Date().toISOString();
|
||||
job.approvals.push({
|
||||
stage: "post",
|
||||
status: "approved",
|
||||
actorId,
|
||||
createdAt: job.updatedAt,
|
||||
});
|
||||
});
|
||||
}
|
||||
|
||||
public async completeJobDirect(jobId: string, finalResponse: string): Promise<JobRecord> {
|
||||
return this.updateJob(jobId, (job) => {
|
||||
if (job.state !== "running") {
|
||||
throw new Error(`Job ${jobId} is not currently running.`);
|
||||
}
|
||||
|
||||
job.state = "completed";
|
||||
job.finalResponse = finalResponse;
|
||||
job.updatedAt = new Date().toISOString();
|
||||
});
|
||||
}
|
||||
|
||||
private async getExistingDedupedJob(input: CreateJobInput): Promise<JobRecord | undefined> {
|
||||
const result = await this.pool.query<{ job_json: unknown }>(
|
||||
`select job_json
|
||||
from jobs
|
||||
where request_message_id = $1
|
||||
and command_name = $2
|
||||
and provider_name = $3
|
||||
and output_channel_id = $4
|
||||
limit 1`,
|
||||
[input.requestMessageId, input.command, input.provider, input.outputChannelId],
|
||||
);
|
||||
return deserializeJobRow(result.rows[0]);
|
||||
}
|
||||
|
||||
private async updateJob(
|
||||
jobId: string,
|
||||
updater: (job: JobRecord) => void,
|
||||
): Promise<JobRecord> {
|
||||
const client = await this.pool.connect();
|
||||
|
||||
try {
|
||||
await client.query("begin");
|
||||
const job = await loadJobForUpdate(client, jobId);
|
||||
|
||||
updater(job);
|
||||
await persistJob(client, job);
|
||||
await client.query("commit");
|
||||
return cloneJson(job);
|
||||
} catch (error) {
|
||||
await client.query("rollback").catch(() => undefined);
|
||||
throw error;
|
||||
} finally {
|
||||
client.release();
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
class PostgresAuditLog implements AuditLog {
|
||||
public constructor(private readonly pool: Pool) {}
|
||||
|
||||
public async record(input: AuditEventInput): Promise<AuditEvent> {
|
||||
const event: AuditEvent = {
|
||||
id: input.id ?? randomUUID(),
|
||||
timestamp: input.timestamp ?? new Date().toISOString(),
|
||||
type: input.type,
|
||||
actorId: input.actorId,
|
||||
jobId: input.jobId,
|
||||
provider: input.provider,
|
||||
details: input.details,
|
||||
};
|
||||
|
||||
await this.pool.query(
|
||||
`insert into audit_events (
|
||||
id,
|
||||
job_id,
|
||||
actor_discord_id,
|
||||
event_type,
|
||||
provider_name,
|
||||
event_timestamp,
|
||||
event_json
|
||||
) values ($1, $2, $3, $4, $5, $6, $7::jsonb)
|
||||
on conflict (id) do update
|
||||
set job_id = excluded.job_id,
|
||||
actor_discord_id = excluded.actor_discord_id,
|
||||
event_type = excluded.event_type,
|
||||
provider_name = excluded.provider_name,
|
||||
event_timestamp = excluded.event_timestamp,
|
||||
event_json = excluded.event_json`,
|
||||
[
|
||||
event.id,
|
||||
event.jobId ?? null,
|
||||
event.actorId ?? null,
|
||||
event.type,
|
||||
event.provider ?? null,
|
||||
event.timestamp,
|
||||
JSON.stringify(event),
|
||||
],
|
||||
);
|
||||
|
||||
return event;
|
||||
}
|
||||
}
|
||||
|
||||
class PostgresInteractionMemoryBank implements InteractionMemoryBank {
|
||||
public constructor(private readonly pool: Pool) {}
|
||||
|
||||
public async getProfile(
|
||||
actorId: string,
|
||||
): Promise<InteractionMemoryProfile | undefined> {
|
||||
const events = await this.listActorEvents(actorId);
|
||||
|
||||
if (events.length === 0) {
|
||||
return undefined;
|
||||
}
|
||||
|
||||
return summarizeInteractionProfile(
|
||||
actorId,
|
||||
events[events.length - 1].actorName,
|
||||
events,
|
||||
);
|
||||
}
|
||||
|
||||
public async recordInteraction(
|
||||
input: RecordInteractionInput,
|
||||
): Promise<InteractionMemoryProfile> {
|
||||
const event = buildInteractionMemoryEvent(input);
|
||||
|
||||
await this.pool.query(
|
||||
`insert into interaction_memory_events (
|
||||
id,
|
||||
actor_id,
|
||||
actor_name,
|
||||
source_kind,
|
||||
guild_id,
|
||||
channel_id,
|
||||
channel_name,
|
||||
command_name,
|
||||
prompt,
|
||||
excerpt,
|
||||
summary,
|
||||
sentiment,
|
||||
score,
|
||||
tags,
|
||||
event_timestamp,
|
||||
event_json
|
||||
) values ($1, $2, $3, $4, $5, $6, $7, $8, $9, $10, $11, $12, $13, $14::text[], $15, $16::jsonb)
|
||||
on conflict (id) do update
|
||||
set actor_id = excluded.actor_id,
|
||||
actor_name = excluded.actor_name,
|
||||
source_kind = excluded.source_kind,
|
||||
guild_id = excluded.guild_id,
|
||||
channel_id = excluded.channel_id,
|
||||
channel_name = excluded.channel_name,
|
||||
command_name = excluded.command_name,
|
||||
prompt = excluded.prompt,
|
||||
excerpt = excluded.excerpt,
|
||||
summary = excluded.summary,
|
||||
sentiment = excluded.sentiment,
|
||||
score = excluded.score,
|
||||
tags = excluded.tags,
|
||||
event_timestamp = excluded.event_timestamp,
|
||||
event_json = excluded.event_json`,
|
||||
[
|
||||
event.id,
|
||||
event.actorId,
|
||||
event.actorName,
|
||||
event.sourceKind,
|
||||
event.guildId ?? null,
|
||||
event.channelId,
|
||||
event.channelName ?? null,
|
||||
event.command ?? null,
|
||||
event.prompt,
|
||||
event.excerpt,
|
||||
event.summary,
|
||||
event.sentiment,
|
||||
event.score,
|
||||
event.tags,
|
||||
event.timestamp,
|
||||
JSON.stringify(event),
|
||||
],
|
||||
);
|
||||
|
||||
const events = await this.listActorEvents(input.actorId);
|
||||
return summarizeInteractionProfile(input.actorId, input.actorName, events);
|
||||
}
|
||||
|
||||
private async listActorEvents(actorId: string): Promise<InteractionMemoryEvent[]> {
|
||||
const result = await this.pool.query<{ event_json: unknown }>(
|
||||
`select event_json
|
||||
from interaction_memory_events
|
||||
where actor_id = $1
|
||||
order by event_timestamp asc, id asc`,
|
||||
[actorId],
|
||||
);
|
||||
|
||||
return result.rows
|
||||
.map((row) => deserializeJson<InteractionMemoryEvent>(row.event_json))
|
||||
.filter((event): event is InteractionMemoryEvent => event !== undefined);
|
||||
}
|
||||
}
|
||||
|
||||
async function bootstrapPostgres(pool: Pool): Promise<void> {
|
||||
const sqlPath = resolve(process.cwd(), "packages", "core", "sql", "bootstrap.sql");
|
||||
const sql = await readFile(sqlPath, "utf8");
|
||||
await pool.query(sql);
|
||||
}
|
||||
|
||||
async function migrateLegacyFileStateIfNeeded(
|
||||
config: StateStorageConfig,
|
||||
pool: Pool,
|
||||
): Promise<void> {
|
||||
const existing = await pool.query<{ name: string }>(
|
||||
"select name from state_migrations where name = $1",
|
||||
[LEGACY_FILE_STATE_IMPORT],
|
||||
);
|
||||
|
||||
if (existing.rowCount && existing.rowCount > 0) {
|
||||
return;
|
||||
}
|
||||
|
||||
const fileJobQueue = new FileBackedJobQueue(config.jobsFile);
|
||||
const fileAuditLog = new FileAuditLog(config.auditLogFile);
|
||||
const fileMemoryBank = new FileInteractionMemoryBank(config.interactionMemoryFile);
|
||||
const jobs = dedupeLegacyJobs(await fileJobQueue.exportJobs());
|
||||
const importedJobIds = new Set(jobs.map((job) => job.id));
|
||||
const auditEvents = await fileAuditLog.listEvents();
|
||||
const profiles = await fileMemoryBank.listProfiles();
|
||||
const memoryEvents = profiles.flatMap((profile) => profile.recentEvents);
|
||||
const client = await pool.connect();
|
||||
|
||||
try {
|
||||
await client.query("begin");
|
||||
|
||||
for (const job of jobs) {
|
||||
await upsertJob(client, job);
|
||||
}
|
||||
|
||||
for (const event of auditEvents) {
|
||||
await client.query(
|
||||
`insert into audit_events (
|
||||
id,
|
||||
job_id,
|
||||
actor_discord_id,
|
||||
event_type,
|
||||
provider_name,
|
||||
event_timestamp,
|
||||
event_json
|
||||
) values ($1, $2, $3, $4, $5, $6, $7::jsonb)
|
||||
on conflict (id) do update
|
||||
set job_id = excluded.job_id,
|
||||
actor_discord_id = excluded.actor_discord_id,
|
||||
event_type = excluded.event_type,
|
||||
provider_name = excluded.provider_name,
|
||||
event_timestamp = excluded.event_timestamp,
|
||||
event_json = excluded.event_json`,
|
||||
[
|
||||
event.id,
|
||||
event.jobId && importedJobIds.has(event.jobId) ? event.jobId : null,
|
||||
event.actorId ?? null,
|
||||
event.type,
|
||||
event.provider ?? null,
|
||||
event.timestamp,
|
||||
JSON.stringify(event),
|
||||
],
|
||||
);
|
||||
}
|
||||
|
||||
for (const event of memoryEvents) {
|
||||
await client.query(
|
||||
`insert into interaction_memory_events (
|
||||
id,
|
||||
actor_id,
|
||||
actor_name,
|
||||
source_kind,
|
||||
guild_id,
|
||||
channel_id,
|
||||
channel_name,
|
||||
command_name,
|
||||
prompt,
|
||||
excerpt,
|
||||
summary,
|
||||
sentiment,
|
||||
score,
|
||||
tags,
|
||||
event_timestamp,
|
||||
event_json
|
||||
) values ($1, $2, $3, $4, $5, $6, $7, $8, $9, $10, $11, $12, $13, $14::text[], $15, $16::jsonb)
|
||||
on conflict (id) do update
|
||||
set actor_id = excluded.actor_id,
|
||||
actor_name = excluded.actor_name,
|
||||
source_kind = excluded.source_kind,
|
||||
guild_id = excluded.guild_id,
|
||||
channel_id = excluded.channel_id,
|
||||
channel_name = excluded.channel_name,
|
||||
command_name = excluded.command_name,
|
||||
prompt = excluded.prompt,
|
||||
excerpt = excluded.excerpt,
|
||||
summary = excluded.summary,
|
||||
sentiment = excluded.sentiment,
|
||||
score = excluded.score,
|
||||
tags = excluded.tags,
|
||||
event_timestamp = excluded.event_timestamp,
|
||||
event_json = excluded.event_json`,
|
||||
[
|
||||
event.id,
|
||||
event.actorId,
|
||||
event.actorName,
|
||||
event.sourceKind,
|
||||
event.guildId ?? null,
|
||||
event.channelId,
|
||||
event.channelName ?? null,
|
||||
event.command ?? null,
|
||||
event.prompt,
|
||||
event.excerpt,
|
||||
event.summary,
|
||||
event.sentiment,
|
||||
event.score,
|
||||
event.tags,
|
||||
event.timestamp,
|
||||
JSON.stringify(event),
|
||||
],
|
||||
);
|
||||
}
|
||||
|
||||
await client.query(
|
||||
"insert into state_migrations (name) values ($1) on conflict (name) do nothing",
|
||||
[LEGACY_FILE_STATE_IMPORT],
|
||||
);
|
||||
await client.query("commit");
|
||||
} catch (error) {
|
||||
await client.query("rollback").catch(() => undefined);
|
||||
throw error;
|
||||
} finally {
|
||||
client.release();
|
||||
}
|
||||
}
|
||||
|
||||
async function insertJob(pool: Pool, job: JobRecord): Promise<void> {
|
||||
await upsertJob(pool, job, false);
|
||||
}
|
||||
|
||||
async function upsertJob(
|
||||
client: Pool | PoolClient,
|
||||
job: JobRecord,
|
||||
useUpsert = true,
|
||||
): Promise<void> {
|
||||
const conflictClause = useUpsert
|
||||
? `on conflict (id) do update
|
||||
set request_message_id = excluded.request_message_id,
|
||||
command_name = excluded.command_name,
|
||||
state = excluded.state,
|
||||
provider_name = excluded.provider_name,
|
||||
requester_discord_id = excluded.requester_discord_id,
|
||||
guild_id = excluded.guild_id,
|
||||
output_channel_id = excluded.output_channel_id,
|
||||
created_at = excluded.created_at,
|
||||
updated_at = excluded.updated_at,
|
||||
job_json = excluded.job_json`
|
||||
: "";
|
||||
|
||||
await client.query(
|
||||
`insert into jobs (
|
||||
id,
|
||||
request_message_id,
|
||||
command_name,
|
||||
state,
|
||||
provider_name,
|
||||
requester_discord_id,
|
||||
guild_id,
|
||||
output_channel_id,
|
||||
created_at,
|
||||
updated_at,
|
||||
job_json
|
||||
) values ($1, $2, $3, $4, $5, $6, $7, $8, $9, $10, $11::jsonb)
|
||||
${conflictClause}`,
|
||||
[
|
||||
job.id,
|
||||
job.requestMessageId ?? null,
|
||||
job.command,
|
||||
job.state,
|
||||
job.provider,
|
||||
job.requester.id,
|
||||
job.guildContext.guildId ?? null,
|
||||
job.outputChannelId,
|
||||
job.createdAt,
|
||||
job.updatedAt,
|
||||
JSON.stringify(job),
|
||||
],
|
||||
);
|
||||
}
|
||||
|
||||
async function loadJobForUpdate(client: PoolClient, jobId: string): Promise<JobRecord> {
|
||||
const result = await client.query<{ job_json: unknown }>(
|
||||
"select job_json from jobs where id = $1 for update",
|
||||
[jobId],
|
||||
);
|
||||
const job = deserializeJobRow(result.rows[0]);
|
||||
|
||||
if (!job) {
|
||||
throw new Error(`Job ${jobId} was not found.`);
|
||||
}
|
||||
|
||||
return job;
|
||||
}
|
||||
|
||||
async function persistJob(client: PoolClient, job: JobRecord): Promise<void> {
|
||||
await client.query(
|
||||
`update jobs
|
||||
set request_message_id = $2,
|
||||
command_name = $3,
|
||||
state = $4,
|
||||
provider_name = $5,
|
||||
requester_discord_id = $6,
|
||||
guild_id = $7,
|
||||
output_channel_id = $8,
|
||||
created_at = $9,
|
||||
updated_at = $10,
|
||||
job_json = $11::jsonb
|
||||
where id = $1`,
|
||||
[
|
||||
job.id,
|
||||
job.requestMessageId ?? null,
|
||||
job.command,
|
||||
job.state,
|
||||
job.provider,
|
||||
job.requester.id,
|
||||
job.guildContext.guildId ?? null,
|
||||
job.outputChannelId,
|
||||
job.createdAt,
|
||||
job.updatedAt,
|
||||
JSON.stringify(job),
|
||||
],
|
||||
);
|
||||
}
|
||||
|
||||
function deserializeJobRow(row: { job_json: unknown } | undefined): JobRecord | undefined {
|
||||
if (!row) {
|
||||
return undefined;
|
||||
}
|
||||
|
||||
return deserializeJson<JobRecord>(row.job_json);
|
||||
}
|
||||
|
||||
function deserializeJson<T>(value: unknown): T | undefined {
|
||||
if (value === undefined || value === null) {
|
||||
return undefined;
|
||||
}
|
||||
|
||||
if (typeof value === "string") {
|
||||
return JSON.parse(value) as T;
|
||||
}
|
||||
|
||||
return value as T;
|
||||
}
|
||||
|
||||
function cloneJson<T>(value: T): T {
|
||||
return JSON.parse(JSON.stringify(value)) as T;
|
||||
}
|
||||
|
||||
function dedupeLegacyJobs(jobs: JobRecord[]): JobRecord[] {
|
||||
const deduped = new Map<string, JobRecord>();
|
||||
|
||||
for (const job of jobs) {
|
||||
const key =
|
||||
job.requestMessageId === undefined
|
||||
? `job:${job.id}`
|
||||
: `request:${job.requestMessageId}:${job.command}:${job.provider}:${job.outputChannelId}`;
|
||||
const existing = deduped.get(key);
|
||||
|
||||
if (!existing || existing.updatedAt < job.updatedAt) {
|
||||
deduped.set(key, job);
|
||||
}
|
||||
}
|
||||
|
||||
return [...deduped.values()].sort((left, right) =>
|
||||
left.createdAt.localeCompare(right.createdAt),
|
||||
);
|
||||
}
|
||||
Loading…
Reference in a new issue