From 965fb1d9daf06064e34537a67c5fa99f54e36f1e Mon Sep 17 00:00:00 2001 From: Meta Vens Date: Sun, 26 Apr 2026 03:49:19 +0100 Subject: [PATCH] Refactor state storage seams --- notes/fresh-workspace-handoff.md | 3 +- notes/voidbot-current-system-map.md | 17 +- notes/voidbot-implementation-plan.md | 2 +- .../src/state-storage-postgres-audit-log.ts | 59 + .../src/state-storage-postgres-bootstrap.ts | 81 ++ ...ate-storage-postgres-interaction-memory.ts | 142 +++ .../src/state-storage-postgres-job-queue.ts | 406 +++++++ .../src/state-storage-rate-limit-stores.ts | 259 +++++ packages/core/src/state-storage-types.ts | 22 + packages/core/src/state-storage.ts | 1001 +---------------- state/evidence.jsonl | 1 + state/map.yaml | 5 +- 12 files changed, 1007 insertions(+), 991 deletions(-) create mode 100644 packages/core/src/state-storage-postgres-audit-log.ts create mode 100644 packages/core/src/state-storage-postgres-bootstrap.ts create mode 100644 packages/core/src/state-storage-postgres-interaction-memory.ts create mode 100644 packages/core/src/state-storage-postgres-job-queue.ts create mode 100644 packages/core/src/state-storage-rate-limit-stores.ts create mode 100644 packages/core/src/state-storage-types.ts diff --git a/notes/fresh-workspace-handoff.md b/notes/fresh-workspace-handoff.md index e32b7a5..11ca944 100644 --- a/notes/fresh-workspace-handoff.md +++ b/notes/fresh-workspace-handoff.md @@ -24,12 +24,13 @@ Git history and smoke artifacts carry routine proof. `state/evidence.jsonl` shou - The owner Codex lane is no longer one swollen file; orchestration, runtime/parsing, rendering, and shared helpers are split under `packages/providers/src/owner-codex-*.ts`. - The Discord bot lane is no longer one swollen entrypoint; gateway wiring remains in `apps/bot/src/discord-bot.ts` while command/prompt handlers and Discord-shape support helpers now live under `apps/bot/src/discord-bot-*.ts`. - Interaction-memory logic is no longer one thousand-line slab; event/tone analysis, shared constants, and profile synthesis are split across `packages/core/src/interaction-memory-*.ts`. +- State storage is no longer one persistence omnibus; `packages/core/src/state-storage.ts` is now a thin factory over domain-specific Postgres/file store modules plus bootstrap/migration helpers. - Ops health, backup verification, offsite sync, and dashboard surfaces already exist. - The new state/notes surfaces are now the continuity spine for future nontrivial work. ## Likely Next Bounded Move -- The next good structural cut is probably `packages/core/src/state-storage.ts`, `apps/worker/src/mcp-server.ts`, or `packages/providers/src/local-llm-provider.ts`. +- The next good structural cut is probably `apps/worker/src/mcp-server.ts` or `packages/providers/src/local-llm-provider.ts`. - Prefer the seam with the clearest concern split rather than chasing line count for its own sake. - Do not broaden scope into a heroic "refactor everything" binge. One organ at a time. diff --git a/notes/voidbot-current-system-map.md b/notes/voidbot-current-system-map.md index 74bf2f0..059ecf7 100644 --- a/notes/voidbot-current-system-map.md +++ b/notes/voidbot-current-system-map.md @@ -15,7 +15,17 @@ This note is the source-grounded description of how the live VoidBot stack is sh - `apps/worker/src/mcp-server.ts` - read-only retrieval tools and resource surfaces for archived history, indexed repos, lore, and owner notifications. - `packages/core/src/state-storage.ts` - - Postgres-backed durable state for jobs, audit events, interaction memory, and Void usage rate-limit counters. + - Thin state-storage factory that chooses file or Postgres backends and wires the domain stores together. +- `packages/core/src/state-storage-postgres-bootstrap.ts` + - Postgres schema bootstrap plus one-time legacy file-state import. +- `packages/core/src/state-storage-postgres-job-queue.ts` + - Postgres-backed job queue plus job row persistence helpers. +- `packages/core/src/state-storage-postgres-audit-log.ts` + - Postgres-backed audit log plus audit-event upsert helper. +- `packages/core/src/state-storage-postgres-interaction-memory.ts` + - Postgres-backed interaction-memory event storage and profile reconstruction. +- `packages/core/src/state-storage-rate-limit-stores.ts` + - File and Postgres implementations for Void usage rate-limit state. - `packages/core/src/interaction-memory-analysis.ts` - Event construction, tone/tag analysis, repetition detection, and persistence gating for remembered interactions. - `packages/core/src/interaction-memory-profile.ts` @@ -104,3 +114,8 @@ This note is the source-grounded description of how the live VoidBot stack is sh - backups and snapshots The important scar is that these stores are split on purpose. Do not casually weld them back into one convenient blob and act surprised when it becomes a moon. + +Within the Postgres path, the implementation is split on purpose now too: +- `state-storage.ts` chooses the backend +- `state-storage-postgres-bootstrap.ts` handles schema/bootstrap/import work +- per-domain modules own queue, audit, interaction-memory, and rate-limit behavior diff --git a/notes/voidbot-implementation-plan.md b/notes/voidbot-implementation-plan.md index dccec1c..45897ae 100644 --- a/notes/voidbot-implementation-plan.md +++ b/notes/voidbot-implementation-plan.md @@ -28,7 +28,7 @@ Use the persistent project-state surfaces to keep future work source-grounded an ### 4. Legibility Hardening - continue splitting oversized runtime files where concern boundaries are already obvious -- likely next cuts: `packages/core/src/state-storage.ts`, `apps/worker/src/mcp-server.ts`, or `packages/providers/src/local-llm-provider.ts` +- likely next cuts: `apps/worker/src/mcp-server.ts` or `packages/providers/src/local-llm-provider.ts` - keep behavior stable while moving concerns into smaller organs ### 5. The Still-Scaffolded Paths diff --git a/packages/core/src/state-storage-postgres-audit-log.ts b/packages/core/src/state-storage-postgres-audit-log.ts new file mode 100644 index 0000000..834d434 --- /dev/null +++ b/packages/core/src/state-storage-postgres-audit-log.ts @@ -0,0 +1,59 @@ +import { randomUUID } from "node:crypto"; + +import { type Pool, type PoolClient } from "pg"; + +import { type AuditEvent } from "@voidbot/shared"; + +import { type AuditEventInput, type AuditLog } from "./audit-log"; + +export class PostgresAuditLog implements AuditLog { + public constructor(private readonly pool: Pool) {} + + public async record(input: AuditEventInput): Promise { + 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 upsertAuditEvent(this.pool, event); + return event; + } +} + +export async function upsertAuditEvent( + client: Pool | PoolClient, + event: AuditEvent, +): Promise { + 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 ?? null, + event.actorId ?? null, + event.type, + event.provider ?? null, + event.timestamp, + JSON.stringify(event), + ], + ); +} diff --git a/packages/core/src/state-storage-postgres-bootstrap.ts b/packages/core/src/state-storage-postgres-bootstrap.ts new file mode 100644 index 0000000..2ca7f65 --- /dev/null +++ b/packages/core/src/state-storage-postgres-bootstrap.ts @@ -0,0 +1,81 @@ +import { readFile } from "node:fs/promises"; +import { resolve } from "node:path"; + +import { type Pool } from "pg"; + +import { FileAuditLog } from "./audit-log"; +import { FileInteractionMemoryBank } from "./interaction-memory"; +import { FileBackedJobQueue } from "./job-queue"; +import { + upsertAuditEvent, +} from "./state-storage-postgres-audit-log"; +import { + dedupeLegacyJobs, + upsertJob, +} from "./state-storage-postgres-job-queue"; +import { + upsertInteractionMemoryEvent, +} from "./state-storage-postgres-interaction-memory"; +import { type StateStorageConfig } from "./state-storage-types"; + +const LEGACY_FILE_STATE_IMPORT = "legacy_file_state_import_v1"; + +export async function bootstrapPostgres(pool: Pool): Promise { + const sqlPath = resolve(process.cwd(), "packages", "core", "sql", "bootstrap.sql"); + const sql = await readFile(sqlPath, "utf8"); + await pool.query(sql); +} + +export async function migrateLegacyFileStateIfNeeded( + config: StateStorageConfig, + pool: Pool, +): Promise { + 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 upsertAuditEvent(client, { + ...event, + jobId: event.jobId && importedJobIds.has(event.jobId) ? event.jobId : undefined, + }); + } + + for (const event of memoryEvents) { + await upsertInteractionMemoryEvent(client, 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(); + } +} diff --git a/packages/core/src/state-storage-postgres-interaction-memory.ts b/packages/core/src/state-storage-postgres-interaction-memory.ts new file mode 100644 index 0000000..69eddce --- /dev/null +++ b/packages/core/src/state-storage-postgres-interaction-memory.ts @@ -0,0 +1,142 @@ +import { type Pool, type PoolClient } from "pg"; + +import { + type InteractionMemoryEvent, + type InteractionMemoryProfile, +} from "@voidbot/shared"; + +import { + buildInteractionMemoryEvent, + type RecordInteractionInput, + shouldPersistInteractionEvent, + summarizeInteractionProfile, +} from "./interaction-memory-logic"; +import { type InteractionMemoryBank } from "./interaction-memory"; + +export class PostgresInteractionMemoryBank implements InteractionMemoryBank { + public constructor(private readonly pool: Pool) {} + + public async getProfile( + actorId: string, + ): Promise { + const events = await this.listActorEvents(actorId); + + if (events.length === 0) { + return undefined; + } + + const profile = summarizeInteractionProfile( + actorId, + events[events.length - 1].actorName, + events, + ); + + return profile.totalInteractions > 0 ? profile : undefined; + } + + public async recordInteraction( + input: RecordInteractionInput, + ): Promise { + const existingEvents = await this.listActorEvents(input.actorId); + const priorEvents = existingEvents.filter((event) => event.id !== input.eventId); + const event = buildInteractionMemoryEvent(input, priorEvents); + + if (!shouldPersistInteractionEvent(event)) { + if (input.eventId) { + await this.pool.query("delete from interaction_memory_events where id = $1", [input.eventId]); + } + + return summarizeInteractionProfile(input.actorId, input.actorName, priorEvents); + } + + await upsertInteractionMemoryEvent(this.pool, event); + + return summarizeInteractionProfile(input.actorId, input.actorName, [...priorEvents, event]); + } + + private async listActorEvents(actorId: string): Promise { + 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(row.event_json)) + .filter((event): event is InteractionMemoryEvent => event !== undefined); + } +} + +export async function upsertInteractionMemoryEvent( + client: Pool | PoolClient, + event: InteractionMemoryEvent, +): Promise { + 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), + ], + ); +} + +function deserializeJson(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; +} diff --git a/packages/core/src/state-storage-postgres-job-queue.ts b/packages/core/src/state-storage-postgres-job-queue.ts new file mode 100644 index 0000000..9ca8f58 --- /dev/null +++ b/packages/core/src/state-storage-postgres-job-queue.ts @@ -0,0 +1,406 @@ +import { randomUUID } from "node:crypto"; + +import { type Pool, type PoolClient } from "pg"; + +import { type JobRecord } from "@voidbot/shared"; + +import { + type CreateJobInput, + type CreateJobResult, + type JobQueue, +} from "./job-queue"; + +export class PostgresJobQueue implements JobQueue { + public constructor(private readonly pool: Pool) {} + + public async createJob(input: CreateJobInput): Promise { + 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 { + 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 { + 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 { + return this.listByStates(["approved"]); + } + + public async claimRunnableJobs(limit?: number): Promise { + 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 { + 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 { + 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 { + 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, + summary: string, + ): Promise { + 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 { + 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 { + 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 { + 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 { + 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 { + 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(); + } + } +} + +export async function upsertJob( + client: Pool | PoolClient, + job: JobRecord, + useUpsert = true, +): Promise { + 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), + ], + ); +} + +export function dedupeLegacyJobs(jobs: JobRecord[]): JobRecord[] { + const deduped = new Map(); + + 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), + ); +} + +async function insertJob(pool: Pool, job: JobRecord): Promise { + await upsertJob(pool, job, false); +} + +async function loadJobForUpdate(client: PoolClient, jobId: string): Promise { + 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 { + 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(row.job_json); +} + +function deserializeJson(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(value: T): T { + return JSON.parse(JSON.stringify(value)) as T; +} diff --git a/packages/core/src/state-storage-rate-limit-stores.ts b/packages/core/src/state-storage-rate-limit-stores.ts new file mode 100644 index 0000000..2529b62 --- /dev/null +++ b/packages/core/src/state-storage-rate-limit-stores.ts @@ -0,0 +1,259 @@ +import { mkdir, readFile, writeFile } from "node:fs/promises"; +import { dirname } from "node:path"; + +import { type Pool } from "pg"; + +import { + type ConsumeVoidUsageRateLimitInput, + type VoidUsageRateLimitDecision, + type VoidUsageRateLimitStore, +} from "./void-usage-rate-limiter"; + +interface VoidUsageRateLimitStateRow { + actor_id: string; + daily_bucket: string; + daily_count: number; + last_request_at: string | null; +} + +interface VoidUsageRateLimitFileStore { + version: 1; + actors: Record< + string, + { + dailyBucket: string; + dailyCount: number; + lastRequestAt?: string; + } + >; +} + +export class PostgresVoidUsageRateLimitStore implements VoidUsageRateLimitStore { + public constructor(private readonly pool: Pool) {} + + public async consume( + input: ConsumeVoidUsageRateLimitInput, + ): Promise { + const timestamp = input.timestamp ?? new Date().toISOString(); + const now = new Date(timestamp); + const bucket = timestamp.slice(0, 10); + const client = await this.pool.connect(); + + try { + await client.query("begin"); + await client.query( + `insert into void_usage_rate_limit_state ( + actor_id, + daily_bucket, + daily_count, + state_json + ) values ($1, $2, 0, '{}'::jsonb) + on conflict (actor_id) do nothing`, + [input.actorId, bucket], + ); + const stateResult = await client.query( + `select actor_id, daily_bucket::text, daily_count, last_request_at::text + from void_usage_rate_limit_state + where actor_id = $1 + for update`, + [input.actorId], + ); + const state = stateResult.rows[0]; + + if (!state) { + throw new Error(`Missing rate limit state for actor ${input.actorId}.`); + } + + const dailyCount = state.daily_bucket === bucket ? state.daily_count : 0; + + if (input.dailyLimit !== undefined && dailyCount >= input.dailyLimit) { + await client.query("commit"); + return { + allowed: false, + reason: "daily_limit", + dailyCount, + resetsAt: nextUtcDay(timestamp), + policy: { + modifier: "default", + matchedSubjects: [], + }, + }; + } + + if (input.cooldownSeconds !== undefined && state.last_request_at) { + const lastRequestAt = new Date(state.last_request_at); + const elapsedSeconds = Math.floor((now.getTime() - lastRequestAt.getTime()) / 1000); + const retryAfterSeconds = input.cooldownSeconds - elapsedSeconds; + + if (retryAfterSeconds > 0) { + await client.query("commit"); + return { + allowed: false, + reason: "cooldown", + dailyCount, + retryAfterSeconds, + resetsAt: now.toISOString(), + policy: { + modifier: "default", + matchedSubjects: [], + }, + }; + } + } + + const nextDailyCount = dailyCount + 1; + await client.query( + `update void_usage_rate_limit_state + set daily_bucket = $2, + daily_count = $3, + last_request_at = $4, + updated_at = now(), + state_json = $5::jsonb + where actor_id = $1`, + [ + input.actorId, + bucket, + nextDailyCount, + timestamp, + JSON.stringify({ + lastCommand: input.command, + lastProvider: input.provider, + guildId: input.guildId ?? null, + channelId: input.channelId, + }), + ], + ); + await client.query("commit"); + + return { + allowed: true, + dailyCount: nextDailyCount, + resetsAt: + input.dailyLimit === undefined ? undefined : nextUtcDay(timestamp), + policy: { + modifier: "default", + matchedSubjects: [], + }, + }; + } catch (error) { + await client.query("rollback").catch(() => undefined); + throw error; + } finally { + client.release(); + } + } +} + +export class FileBackedVoidUsageRateLimitStore implements VoidUsageRateLimitStore { + private writeChain: Promise = Promise.resolve(); + + public constructor(private readonly filePath: string) {} + + public async consume( + input: ConsumeVoidUsageRateLimitInput, + ): Promise { + return this.serialize(async () => { + const timestamp = input.timestamp ?? new Date().toISOString(); + const now = new Date(timestamp); + const bucket = timestamp.slice(0, 10); + const store = await this.readUnlocked(); + const existing = store.actors[input.actorId] ?? { + dailyBucket: bucket, + dailyCount: 0, + }; + const dailyCount = existing.dailyBucket === bucket ? existing.dailyCount : 0; + + if (input.dailyLimit !== undefined && dailyCount >= input.dailyLimit) { + return { + allowed: false, + reason: "daily_limit", + dailyCount, + resetsAt: nextUtcDay(timestamp), + policy: { + modifier: "default", + matchedSubjects: [], + }, + }; + } + + if (input.cooldownSeconds !== undefined && existing.lastRequestAt) { + const lastRequestAt = new Date(existing.lastRequestAt); + const elapsedSeconds = Math.floor((now.getTime() - lastRequestAt.getTime()) / 1000); + const retryAfterSeconds = input.cooldownSeconds - elapsedSeconds; + + if (retryAfterSeconds > 0) { + return { + allowed: false, + reason: "cooldown", + dailyCount, + retryAfterSeconds, + resetsAt: now.toISOString(), + policy: { + modifier: "default", + matchedSubjects: [], + }, + }; + } + } + + const nextDailyCount = dailyCount + 1; + store.actors[input.actorId] = { + dailyBucket: bucket, + dailyCount: nextDailyCount, + lastRequestAt: timestamp, + }; + await this.writeUnlocked(store); + + return { + allowed: true, + dailyCount: nextDailyCount, + resetsAt: + input.dailyLimit === undefined ? undefined : nextUtcDay(timestamp), + policy: { + modifier: "default", + matchedSubjects: [], + }, + }; + }); + } + + private async serialize(operation: () => Promise): Promise { + const pending = this.writeChain.then(operation, operation); + this.writeChain = pending.then( + () => undefined, + () => undefined, + ); + return pending; + } + + private async readUnlocked(): Promise { + try { + const raw = await readFile(this.filePath, "utf8"); + return JSON.parse(raw) as VoidUsageRateLimitFileStore; + } catch (error) { + const code = (error as NodeJS.ErrnoException).code; + + if (code === "ENOENT") { + return { + version: 1, + actors: {}, + }; + } + + throw error; + } + } + + private async writeUnlocked(store: VoidUsageRateLimitFileStore): Promise { + await mkdir(dirname(this.filePath), { recursive: true }); + await writeFile(this.filePath, `${JSON.stringify(store, null, 2)}\n`, "utf8"); + } +} + +function nextUtcDay(timestamp: string): string { + const date = new Date(timestamp); + const next = new Date( + Date.UTC(date.getUTCFullYear(), date.getUTCMonth(), date.getUTCDate() + 1, 0, 0, 0, 0), + ); + return next.toISOString(); +} diff --git a/packages/core/src/state-storage-types.ts b/packages/core/src/state-storage-types.ts new file mode 100644 index 0000000..e18ede7 --- /dev/null +++ b/packages/core/src/state-storage-types.ts @@ -0,0 +1,22 @@ +import type { AuditLog } from "./audit-log"; +import type { InteractionMemoryBank } from "./interaction-memory"; +import type { JobQueue } from "./job-queue"; +import type { VoidUsageRateLimitStore } from "./void-usage-rate-limiter"; + +export interface StateStorageConfig { + backend: "file" | "postgres"; + databaseDsn: string; + jobsFile: string; + auditLogFile: string; + interactionMemoryFile: string; + rateLimitStateFile: string; +} + +export interface StateStorage { + backend: "file" | "postgres"; + jobQueue: JobQueue; + auditLog: AuditLog; + interactionMemory: InteractionMemoryBank; + voidUsageRateLimits: VoidUsageRateLimitStore; + close(): Promise; +} diff --git a/packages/core/src/state-storage.ts b/packages/core/src/state-storage.ts index d7dee1e..2b26de6 100644 --- a/packages/core/src/state-storage.ts +++ b/packages/core/src/state-storage.ts @@ -1,58 +1,22 @@ -import { randomUUID } from "node:crypto"; -import { mkdir, readFile, writeFile } from "node:fs/promises"; -import { dirname, resolve } from "node:path"; - -import { Pool, type PoolClient } from "pg"; +import { Pool } from "pg"; +import { FileAuditLog } from "./audit-log"; +import { FileInteractionMemoryBank } from "./interaction-memory"; +import { FileBackedJobQueue } from "./job-queue"; +import { PostgresAuditLog } from "./state-storage-postgres-audit-log"; import { - type AuditEvent, - type InteractionMemoryEvent, - type InteractionMemoryProfile, - type JobRecord, -} from "@voidbot/shared"; - -import { type AuditEventInput, type AuditLog, FileAuditLog } from "./audit-log"; + bootstrapPostgres, + migrateLegacyFileStateIfNeeded, +} from "./state-storage-postgres-bootstrap"; +import { PostgresJobQueue } from "./state-storage-postgres-job-queue"; +import { PostgresInteractionMemoryBank } from "./state-storage-postgres-interaction-memory"; import { - buildInteractionMemoryEvent, - type RecordInteractionInput, - shouldPersistInteractionEvent, - summarizeInteractionProfile, -} from "./interaction-memory-logic"; -import { - type InteractionMemoryBank, - FileInteractionMemoryBank, -} from "./interaction-memory"; -import { - type CreateJobInput, - type CreateJobResult, - type JobQueue, - FileBackedJobQueue, -} from "./job-queue"; -import { - type ConsumeVoidUsageRateLimitInput, - type VoidUsageRateLimitDecision, - type VoidUsageRateLimitStore, -} from "./void-usage-rate-limiter"; + FileBackedVoidUsageRateLimitStore, + PostgresVoidUsageRateLimitStore, +} from "./state-storage-rate-limit-stores"; +import { type StateStorage, type StateStorageConfig } from "./state-storage-types"; -const LEGACY_FILE_STATE_IMPORT = "legacy_file_state_import_v1"; - -export interface StateStorageConfig { - backend: "file" | "postgres"; - databaseDsn: string; - jobsFile: string; - auditLogFile: string; - interactionMemoryFile: string; - rateLimitStateFile: string; -} - -export interface StateStorage { - backend: "file" | "postgres"; - jobQueue: JobQueue; - auditLog: AuditLog; - interactionMemory: InteractionMemoryBank; - voidUsageRateLimits: VoidUsageRateLimitStore; - close(): Promise; -} +export type { StateStorage, StateStorageConfig } from "./state-storage-types"; export async function createStateStorage( config: StateStorageConfig, @@ -95,938 +59,3 @@ export async function createStateStorage( throw error; } } - -class PostgresJobQueue implements JobQueue { - public constructor(private readonly pool: Pool) {} - - public async createJob(input: CreateJobInput): Promise { - 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 { - 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 { - 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 { - return this.listByStates(["approved"]); - } - - public async claimRunnableJobs(limit?: number): Promise { - 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 { - 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 { - 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 { - 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, - summary: string, - ): Promise { - 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 { - 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 { - 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 { - 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 { - 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 { - 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 { - 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 { - const events = await this.listActorEvents(actorId); - - if (events.length === 0) { - return undefined; - } - - const profile = summarizeInteractionProfile( - actorId, - events[events.length - 1].actorName, - events, - ); - - return profile.totalInteractions > 0 ? profile : undefined; - } - - public async recordInteraction( - input: RecordInteractionInput, - ): Promise { - const existingEvents = await this.listActorEvents(input.actorId); - const priorEvents = existingEvents.filter((event) => event.id !== input.eventId); - const event = buildInteractionMemoryEvent(input, priorEvents); - - if (!shouldPersistInteractionEvent(event)) { - if (input.eventId) { - await this.pool.query("delete from interaction_memory_events where id = $1", [input.eventId]); - } - - return summarizeInteractionProfile(input.actorId, input.actorName, priorEvents); - } - - 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), - ], - ); - - return summarizeInteractionProfile(input.actorId, input.actorName, [...priorEvents, event]); - } - - private async listActorEvents(actorId: string): Promise { - 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(row.event_json)) - .filter((event): event is InteractionMemoryEvent => event !== undefined); - } -} - -interface VoidUsageRateLimitStateRow { - actor_id: string; - daily_bucket: string; - daily_count: number; - last_request_at: string | null; -} - -class PostgresVoidUsageRateLimitStore implements VoidUsageRateLimitStore { - public constructor(private readonly pool: Pool) {} - - public async consume( - input: ConsumeVoidUsageRateLimitInput, - ): Promise { - const timestamp = input.timestamp ?? new Date().toISOString(); - const now = new Date(timestamp); - const bucket = timestamp.slice(0, 10); - const client = await this.pool.connect(); - - try { - await client.query("begin"); - await client.query( - `insert into void_usage_rate_limit_state ( - actor_id, - daily_bucket, - daily_count, - state_json - ) values ($1, $2, 0, '{}'::jsonb) - on conflict (actor_id) do nothing`, - [input.actorId, bucket], - ); - const stateResult = await client.query( - `select actor_id, daily_bucket::text, daily_count, last_request_at::text - from void_usage_rate_limit_state - where actor_id = $1 - for update`, - [input.actorId], - ); - const state = stateResult.rows[0]; - - if (!state) { - throw new Error(`Missing rate limit state for actor ${input.actorId}.`); - } - - const dailyCount = state.daily_bucket === bucket ? state.daily_count : 0; - - if (input.dailyLimit !== undefined && dailyCount >= input.dailyLimit) { - await client.query("commit"); - return { - allowed: false, - reason: "daily_limit", - dailyCount, - resetsAt: nextUtcDay(timestamp), - policy: { - modifier: "default", - matchedSubjects: [], - }, - }; - } - - if (input.cooldownSeconds !== undefined && state.last_request_at) { - const lastRequestAt = new Date(state.last_request_at); - const elapsedSeconds = Math.floor((now.getTime() - lastRequestAt.getTime()) / 1000); - const retryAfterSeconds = input.cooldownSeconds - elapsedSeconds; - - if (retryAfterSeconds > 0) { - await client.query("commit"); - return { - allowed: false, - reason: "cooldown", - dailyCount, - retryAfterSeconds, - resetsAt: now.toISOString(), - policy: { - modifier: "default", - matchedSubjects: [], - }, - }; - } - } - - const nextDailyCount = dailyCount + 1; - await client.query( - `update void_usage_rate_limit_state - set daily_bucket = $2, - daily_count = $3, - last_request_at = $4, - updated_at = now(), - state_json = $5::jsonb - where actor_id = $1`, - [ - input.actorId, - bucket, - nextDailyCount, - timestamp, - JSON.stringify({ - lastCommand: input.command, - lastProvider: input.provider, - guildId: input.guildId ?? null, - channelId: input.channelId, - }), - ], - ); - await client.query("commit"); - - return { - allowed: true, - dailyCount: nextDailyCount, - resetsAt: - input.dailyLimit === undefined ? undefined : nextUtcDay(timestamp), - policy: { - modifier: "default", - matchedSubjects: [], - }, - }; - } catch (error) { - await client.query("rollback").catch(() => undefined); - throw error; - } finally { - client.release(); - } - } -} - -interface VoidUsageRateLimitFileStore { - version: 1; - actors: Record< - string, - { - dailyBucket: string; - dailyCount: number; - lastRequestAt?: string; - } - >; -} - -class FileBackedVoidUsageRateLimitStore implements VoidUsageRateLimitStore { - private writeChain: Promise = Promise.resolve(); - - public constructor(private readonly filePath: string) {} - - public async consume( - input: ConsumeVoidUsageRateLimitInput, - ): Promise { - return this.serialize(async () => { - const timestamp = input.timestamp ?? new Date().toISOString(); - const now = new Date(timestamp); - const bucket = timestamp.slice(0, 10); - const store = await this.readUnlocked(); - const existing = store.actors[input.actorId] ?? { - dailyBucket: bucket, - dailyCount: 0, - }; - const dailyCount = existing.dailyBucket === bucket ? existing.dailyCount : 0; - - if (input.dailyLimit !== undefined && dailyCount >= input.dailyLimit) { - return { - allowed: false, - reason: "daily_limit", - dailyCount, - resetsAt: nextUtcDay(timestamp), - policy: { - modifier: "default", - matchedSubjects: [], - }, - }; - } - - if (input.cooldownSeconds !== undefined && existing.lastRequestAt) { - const lastRequestAt = new Date(existing.lastRequestAt); - const elapsedSeconds = Math.floor((now.getTime() - lastRequestAt.getTime()) / 1000); - const retryAfterSeconds = input.cooldownSeconds - elapsedSeconds; - - if (retryAfterSeconds > 0) { - return { - allowed: false, - reason: "cooldown", - dailyCount, - retryAfterSeconds, - resetsAt: now.toISOString(), - policy: { - modifier: "default", - matchedSubjects: [], - }, - }; - } - } - - const nextDailyCount = dailyCount + 1; - store.actors[input.actorId] = { - dailyBucket: bucket, - dailyCount: nextDailyCount, - lastRequestAt: timestamp, - }; - await this.writeUnlocked(store); - - return { - allowed: true, - dailyCount: nextDailyCount, - resetsAt: - input.dailyLimit === undefined ? undefined : nextUtcDay(timestamp), - policy: { - modifier: "default", - matchedSubjects: [], - }, - }; - }); - } - - private async serialize(operation: () => Promise): Promise { - const pending = this.writeChain.then(operation, operation); - this.writeChain = pending.then( - () => undefined, - () => undefined, - ); - return pending; - } - - private async readUnlocked(): Promise { - try { - const raw = await readFile(this.filePath, "utf8"); - return JSON.parse(raw) as VoidUsageRateLimitFileStore; - } catch (error) { - const code = (error as NodeJS.ErrnoException).code; - - if (code === "ENOENT") { - return { - version: 1, - actors: {}, - }; - } - - throw error; - } - } - - private async writeUnlocked(store: VoidUsageRateLimitFileStore): Promise { - await mkdir(dirname(this.filePath), { recursive: true }); - await writeFile(this.filePath, `${JSON.stringify(store, null, 2)}\n`, "utf8"); - } -} - -async function bootstrapPostgres(pool: Pool): Promise { - 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 { - 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 { - await upsertJob(pool, job, false); -} - -async function upsertJob( - client: Pool | PoolClient, - job: JobRecord, - useUpsert = true, -): Promise { - 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 { - 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 { - 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(row.job_json); -} - -function deserializeJson(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(value: T): T { - return JSON.parse(JSON.stringify(value)) as T; -} - -function dedupeLegacyJobs(jobs: JobRecord[]): JobRecord[] { - const deduped = new Map(); - - 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), - ); -} - -function nextUtcDay(timestamp: string): string { - const date = new Date(timestamp); - const next = new Date( - Date.UTC(date.getUTCFullYear(), date.getUTCMonth(), date.getUTCDate() + 1, 0, 0, 0, 0), - ); - return next.toISOString(); -} diff --git a/state/evidence.jsonl b/state/evidence.jsonl index 88217c9..c906594 100644 --- a/state/evidence.jsonl +++ b/state/evidence.jsonl @@ -4,3 +4,4 @@ {"ts":"2026-04-26T03:40:00+01:00","type":"architecture","status":"ok","note":"The owner Codex provider was split into orchestration, runtime/parsing, rendering, and shared-helper modules; the architecture was sound, but file bloat had become the likelier future failure surface."} {"ts":"2026-04-26T15:20:00+01:00","type":"architecture","status":"ok","note":"The Discord bot lane was split into a gateway shell plus dedicated command-handler and support-helper modules; behavior stayed stable, but line-count monarchy had become the likelier future failure surface here too."} {"ts":"2026-04-26T15:55:00+01:00","type":"architecture","status":"ok","note":"Interaction-memory logic was split into event/tone analysis, shared constants, and profile-synthesis modules; behavior stayed stable, but one file trying to be both the ear and the therapist had become the likelier future failure surface."} +{"ts":"2026-04-26T16:25:00+01:00","type":"architecture","status":"ok","note":"State storage was split into a thin factory, bootstrap/migration helper, per-domain Postgres stores, and dedicated rate-limit stores; behavior stayed stable, but one persistence omnibus had become the likelier future failure surface."} diff --git a/state/map.yaml b/state/map.yaml index 455a417..ec53607 100644 --- a/state/map.yaml +++ b/state/map.yaml @@ -75,6 +75,7 @@ landed_baseline: - The owner Codex lane is now split into orchestration, runtime/parsing, shared helpers, and rendering modules instead of one swollen provider file. - The Discord bot lane is now split into a gateway shell plus dedicated command-handler and support helper modules instead of one swollen entrypoint file. - Interaction-memory logic is now split into event/tone analysis and profile-synthesis modules instead of one slab trying to be both the ear and the therapist. + - State storage is now split into a thin factory plus per-domain Postgres/file store modules and bootstrap/migration helpers instead of one persistence omnibus. rejected_paths: - name: transcript_only_continuity @@ -85,5 +86,5 @@ rejected_paths: reason: vibes produce plausible slop that sounds smart and lies with a straight face current_status: - summary: VoidBot now has a real project-memory spine plus three legibility-hardening passes on major runtime organs. The owner Codex lane, Discord bot lane, and interaction-memory logic were all split into smaller seams, which lowers local edit risk without changing their behavioral contracts. The stack is still dense, but it no longer depends purely on transcript memory or one-file monarchies to stay coherent. - next_action: Keep using the canonical state files for re-entry, then continue decomposing the next oversized organ with the clearest concern boundaries, likely the state-storage persistence slab, MCP server, or local LLM provider. + summary: VoidBot now has a real project-memory spine plus four legibility-hardening passes on major runtime organs. The owner Codex lane, Discord bot lane, interaction-memory logic, and state-storage persistence layer were all split into smaller seams, which lowers local edit risk without changing their behavioral contracts. The stack is still dense, but it no longer depends purely on transcript memory or one-file monarchies to stay coherent. + next_action: Keep using the canonical state files for re-entry, then continue decomposing the next oversized organ with the clearest concern boundaries, likely the MCP server or local LLM provider.