diff --git a/.env.example b/.env.example index 5e6bd05..78d6d5e 100644 --- a/.env.example +++ b/.env.example @@ -3,6 +3,7 @@ DISCORD_BOT_TOKEN= DISCORD_APPLICATION_ID= DISCORD_GUILD_ID= DISCORD_OWNER_ID= +STATE_STORAGE_BACKEND=postgres DATABASE_DSN=postgres://voidbot:voidbot@localhost:5432/voidbot # Leave this as local_json until Qdrant is running and you've migrated the existing vectors. VECTOR_STORE_KIND=local_json diff --git a/README.md b/README.md index 840755c..23e626f 100644 --- a/README.md +++ b/README.md @@ -47,7 +47,7 @@ The repo is split into a bot, a worker, and a handful of focused packages: - `packages/shared`: shared contracts and types - `packages/sandbox`: policy-first sandbox scaffolding -The durable local state lives under `.voidbot/`. That includes jobs, audit logs, artifacts, archives, interaction memory, and detached indexing logs. +The durable local state is split on purpose: Postgres holds jobs, audit events, and interaction memory, while `.voidbot/` keeps artifacts, archives, logs, and detached indexing status. ## Quick Start @@ -103,7 +103,22 @@ ollama pull qwen3-embedding:0.6b `RAG_EMBEDDING_DIMENSIONS` only matters for the hash fallback backend. Ollama-backed embeddings report their own dimensions. -### 4. Start Qdrant +### 4. Start Postgres + +VoidBot now keeps queue state, audit events, and interaction memory in Postgres. `npm run stack:start` will launch the local Docker Postgres container automatically for the default local DSN, but you can start it yourself if you want the database up first: + +```bash +docker compose -f infra/postgres/docker-compose.yml up -d +``` + +The default local state-store config is: + +```dotenv +STATE_STORAGE_BACKEND=postgres +DATABASE_DSN=postgres://voidbot:voidbot@localhost:5432/voidbot +``` + +### 5. Start Qdrant Qdrant is the recommended vector backend now. It avoids the pain that comes from stuffing everything into local JSON and then acting surprised when the file turns into a small moon. @@ -144,7 +159,7 @@ npm run rag:migrate-qdrant Use `-- --wipe` only when you intentionally want to replace the target collections. -### 5. Start the stack +### 6. Start the stack The normal startup path is one command: @@ -154,6 +169,7 @@ npm run stack:start That script: +- verifies or starts local Postgres if `STATE_STORAGE_BACKEND=postgres` - verifies Qdrant if `VECTOR_STORE_KIND=qdrant` - checks the Ollama endpoints and required models - rebuilds the workspace @@ -373,16 +389,15 @@ Some commands are full working paths. Some are still thin MVPs. They exist becau Useful local state and docs: -- `.voidbot/jobs/jobs.json`: local job queue -- `.voidbot/audit/events.jsonl`: audit trail - `.voidbot/artifacts//`: handoff bundles, stdout, stderr, traces - `.voidbot/rag/messages.json`: archived Discord messages - `.voidbot/rag/source-documents.json`: archived source and lore documents - `.voidbot/rag/import-state.json`: log-backfill file state +- `.voidbot/status/runtime-stack.json`: live stack status, service reachability, and bot/worker PIDs - `config/system-messages.json`: rotating stock system messages - `styles/void-default.md`: public default style pack - `docs/architecture/overview.md`: higher-level architecture notes -- `packages/core/sql/bootstrap.sql`: draft PostgreSQL schema +- `packages/core/sql/bootstrap.sql`: PostgreSQL schema for jobs, audit events, and interaction memory ## Safety Boundaries @@ -397,11 +412,10 @@ Useful local state and docs: Reasonable next steps from here: -1. Move queue and audit storage into PostgreSQL. -2. Add health checks, backups, and operational guidance around Qdrant. -3. Expand worker processing into richer provider run records and moderation hooks. -4. Replace the remaining scaffolded provider paths with funded, production-grade implementations. -5. Add real constrained sandbox execution instead of policy-only dry runs. +1. Add health checks, backups, and operational guidance around Postgres and Qdrant. +2. Expand worker processing into richer provider run records and moderation hooks. +3. Replace the remaining scaffolded provider paths with funded, production-grade implementations. +4. Add real constrained sandbox execution instead of policy-only dry runs. ## Related Docs diff --git a/apps/bot/src/discord-bot.ts b/apps/bot/src/discord-bot.ts index 0eeb9fb..e94a035 100644 --- a/apps/bot/src/discord-bot.ts +++ b/apps/bot/src/discord-bot.ts @@ -16,12 +16,13 @@ import { import { loadConfig } from "@voidbot/config"; import { + type AuditLog, buildVoidMcpServerConfig, ContextBuilder, - FileAuditLog, - FileBackedJobQueue, - FileInteractionMemoryBank, PermissionEngine, + createStateStorage, + type InteractionMemoryBank, + type JobQueue, loadStylePack, loadSystemMessageCatalog, type SystemMessageCatalog, @@ -142,9 +143,17 @@ export async function startBot(): Promise { const permissionEngine = new PermissionEngine(config.ownerDiscordId, { localLlmAllowPublic: config.localLlm.allowPublic, }); - const jobQueue = new FileBackedJobQueue(config.jobsFile); - const auditLog = new FileAuditLog(config.auditLogFile); - const interactionMemory = new FileInteractionMemoryBank(config.interactionMemoryFile); + const { + jobQueue, + auditLog, + interactionMemory, + } = await createStateStorage({ + backend: config.stateStorageBackend, + databaseDsn: config.databaseDsn, + jobsFile: config.jobsFile, + auditLogFile: config.auditLogFile, + interactionMemoryFile: config.interactionMemoryFile, + }); const contextBuilder = new ContextBuilder(); let activeStylePack = await loadStylePack(config.stylePackPath); let activeSystemMessages = await loadSystemMessageCatalog(config.systemMessagesPath); @@ -563,9 +572,9 @@ interface PromptHandlerOptions { contextBuilder: ContextBuilder; retrievalService: RetrievalService; sourceArchiveRepository: FileSourceDocumentArchiveRepository; - jobQueue: FileBackedJobQueue; - auditLog: FileAuditLog; - interactionMemory: FileInteractionMemoryBank; + jobQueue: JobQueue; + auditLog: AuditLog; + interactionMemory: InteractionMemoryBank; providerRegistry: ProviderRegistry; stylePack?: StylePack; systemMessages: SystemMessageCatalog; @@ -715,8 +724,8 @@ async function handleApproveJob(options: { interaction: ChatInputCommandInteraction; client: Client; actor: Actor; - jobQueue: FileBackedJobQueue; - auditLog: FileAuditLog; + jobQueue: JobQueue; + auditLog: AuditLog; permissionEngine: PermissionEngine; systemMessages: SystemMessageCatalog; }): Promise { @@ -795,8 +804,8 @@ async function handleApproveJob(options: { async function handleRejectJob(options: { interaction: ChatInputCommandInteraction; actor: Actor; - jobQueue: FileBackedJobQueue; - auditLog: FileAuditLog; + jobQueue: JobQueue; + auditLog: AuditLog; permissionEngine: PermissionEngine; systemMessages: SystemMessageCatalog; }): Promise { @@ -994,7 +1003,7 @@ async function ingestIfIndexed( async function rememberAmbientVoidReference( message: Message, botUserId: string | undefined, - interactionMemory: FileInteractionMemoryBank, + interactionMemory: InteractionMemoryBank, ): Promise { if (!isAmbientVoidReference(message, botUserId)) { return; diff --git a/apps/worker/src/index.ts b/apps/worker/src/index.ts index e60405d..d0b4da7 100644 --- a/apps/worker/src/index.ts +++ b/apps/worker/src/index.ts @@ -5,9 +5,10 @@ import { join } from "node:path"; import { loadConfig } from "@voidbot/config"; import { + type AuditLog, buildVoidMcpServerConfig, - FileAuditLog, - FileBackedJobQueue, + createStateStorage, + type JobQueue, loadSystemMessageCatalog, type SystemMessageCatalog, } from "@voidbot/core"; @@ -33,8 +34,6 @@ import { } from "@voidbot/shared"; const config = loadConfig(); -const jobQueue = new FileBackedJobQueue(config.jobsFile); -const auditLog = new FileAuditLog(config.auditLogFile); const embedder = createTextEmbedder({ backend: config.ragEmbeddingBackend, hashDimensions: config.ragEmbeddingDimensions, @@ -72,6 +71,8 @@ const retrievalService = new RetrievalService( let isProcessing = false; let providerRegistry: ProviderRegistry; let systemMessages: SystemMessageCatalog; +let jobQueue: JobQueue; +let auditLog: AuditLog; void main().catch((error) => { console.error(error); @@ -79,6 +80,18 @@ void main().catch((error) => { }); async function main(): Promise { + const { + jobQueue: stateJobQueue, + auditLog: stateAuditLog, + } = await createStateStorage({ + backend: config.stateStorageBackend, + databaseDsn: config.databaseDsn, + jobsFile: config.jobsFile, + auditLogFile: config.auditLogFile, + interactionMemoryFile: config.interactionMemoryFile, + }); + jobQueue = stateJobQueue; + auditLog = stateAuditLog; systemMessages = await loadSystemMessageCatalog(config.systemMessagesPath); providerRegistry = await buildProviderRegistry(systemMessages); console.log(`VoidBot worker polling every ${config.workerPollIntervalMs}ms.`); diff --git a/docs/architecture/overview.md b/docs/architecture/overview.md index 34fad13..d950a2e 100644 --- a/docs/architecture/overview.md +++ b/docs/architecture/overview.md @@ -10,24 +10,23 @@ Discord Gateway -> Job Queue -> Provider Adapter -> Storage Layer - - file-backed queue and audit log for local development - - PostgreSQL schema draft in packages/core/sql/bootstrap.sql - - persistent local archive and vector index for RAG + - PostgreSQL-backed jobs, audit events, and interaction memory + - persistent local archives for RAG + - Qdrant or local JSON vector stores for retrieval ``` ## Current state - `apps/bot` owns Discord connectivity, routing, and command handling. - `apps/worker` polls owner jobs, runs the local Discord-safe Codex lane, serves bounded history-search tool requests from the RAG index, and prepares handoff bundles when needed. -- `packages/core` contains permissions, context assembly, job storage, audit logging, style loading, and the rotating system message catalog. +- `packages/core` contains permissions, context assembly, Postgres-backed state storage, style loading, and the rotating system message catalog. - `packages/providers` isolates provider contracts from Discord code. - `packages/rag` contains the message archive, chunking pipeline, local embedding backends, pluggable vector-store backends, and log import state. - `packages/sandbox` defines policy profiles and a dry-run policy runner. -## Deliberate shortcuts in this first pass +## Current shortcuts and tradeoffs -- Job storage and audit events are file-backed under `.voidbot/` so the bot and worker can function immediately. -- The relational schema is drafted but not yet wired into a real repository implementation. +- Postgres now owns jobs, audit events, and interaction memory, but the archive side of RAG is still file-backed under `.voidbot/`. - Retrieval now uses a persistent vector backend with a local embedding backend by default (`qwen3-embedding:0.6b` through Ollama). - `VECTOR_STORE_KIND=local_json` keeps the zero-service path available for local development, but `VECTOR_STORE_KIND=qdrant` is now supported for a proper vector database. - Discord history vectors and source-tree/lore vectors stay in separate physical partitions so source indexing no longer rewrites the entire history corpus. @@ -42,7 +41,7 @@ Discord Gateway ## Upgrade path -1. Replace the file-backed queue and audit log with PostgreSQL-backed repositories. -2. Cut production deployments over to Qdrant while keeping separate collections for Discord history and source/lore corpora. -3. Add moderation, budgeting, and rate limits before enabling `openai_api` for member traffic. +1. Add health checks, backup/restore guidance, and operational runbooks for Postgres and Qdrant. +2. Add moderation, budgeting, and rate limits before enabling `openai_api` for member traffic. +3. Expand worker-side run records and admin tooling around the interaction memory/event stores. 4. Expand sandbox execution from dry-run policy checks to real constrained runners. diff --git a/infra/postgres/docker-compose.yml b/infra/postgres/docker-compose.yml new file mode 100644 index 0000000..1546188 --- /dev/null +++ b/infra/postgres/docker-compose.yml @@ -0,0 +1,23 @@ +services: + postgres: + image: postgres:16-alpine + container_name: voidbot-postgres + ports: + - "${VOIDBOT_POSTGRES_PORT:-5432}:5432" + environment: + POSTGRES_DB: "${VOIDBOT_POSTGRES_DB:-voidbot}" + POSTGRES_USER: "${VOIDBOT_POSTGRES_USER:-voidbot}" + POSTGRES_PASSWORD: "${VOIDBOT_POSTGRES_PASSWORD:-voidbot}" + volumes: + - voidbot-postgres-data:/var/lib/postgresql/data + healthcheck: + test: + - CMD-SHELL + - pg_isready -U "$$POSTGRES_USER" -d "$$POSTGRES_DB" + interval: 5s + timeout: 5s + retries: 12 + start_period: 5s + +volumes: + voidbot-postgres-data: diff --git a/package-lock.json b/package-lock.json index 5e11dc3..f3a6a82 100644 --- a/package-lock.json +++ b/package-lock.json @@ -13,6 +13,7 @@ ], "devDependencies": { "@types/node": "^24.0.0", + "@types/pg": "^8.20.0", "tsx": "^4.20.0", "typescript": "^5.9.0" } @@ -747,6 +748,18 @@ "undici-types": "~7.16.0" } }, + "node_modules/@types/pg": { + "version": "8.20.0", + "resolved": "https://registry.npmjs.org/@types/pg/-/pg-8.20.0.tgz", + "integrity": "sha512-bEPFOaMAHTEP1EzpvHTbmwR8UsFyHSKsRisLIHVMXnpNefSbGA1bD6CVy+qKjGSqmZqNqBDV2azOBo8TgkcVow==", + "dev": true, + "license": "MIT", + "dependencies": { + "@types/node": "*", + "pg-protocol": "*", + "pg-types": "^2.2.0" + } + }, "node_modules/@types/ws": { "version": "8.18.1", "resolved": "https://registry.npmjs.org/@types/ws/-/ws-8.18.1.tgz", @@ -1680,6 +1693,95 @@ "url": "https://opencollective.com/express" } }, + "node_modules/pg": { + "version": "8.20.0", + "resolved": "https://registry.npmjs.org/pg/-/pg-8.20.0.tgz", + "integrity": "sha512-ldhMxz2r8fl/6QkXnBD3CR9/xg694oT6DZQ2s6c/RI28OjtSOpxnPrUCGOBJ46RCUxcWdx3p6kw/xnDHjKvaRA==", + "license": "MIT", + "dependencies": { + "pg-connection-string": "^2.12.0", + "pg-pool": "^3.13.0", + "pg-protocol": "^1.13.0", + "pg-types": "2.2.0", + "pgpass": "1.0.5" + }, + "engines": { + "node": ">= 16.0.0" + }, + "optionalDependencies": { + "pg-cloudflare": "^1.3.0" + }, + "peerDependencies": { + "pg-native": ">=3.0.1" + }, + "peerDependenciesMeta": { + "pg-native": { + "optional": true + } + } + }, + "node_modules/pg-cloudflare": { + "version": "1.3.0", + "resolved": "https://registry.npmjs.org/pg-cloudflare/-/pg-cloudflare-1.3.0.tgz", + "integrity": "sha512-6lswVVSztmHiRtD6I8hw4qP/nDm1EJbKMRhf3HCYaqud7frGysPv7FYJ5noZQdhQtN2xJnimfMtvQq21pdbzyQ==", + "license": "MIT", + "optional": true + }, + "node_modules/pg-connection-string": { + "version": "2.12.0", + "resolved": "https://registry.npmjs.org/pg-connection-string/-/pg-connection-string-2.12.0.tgz", + "integrity": "sha512-U7qg+bpswf3Cs5xLzRqbXbQl85ng0mfSV/J0nnA31MCLgvEaAo7CIhmeyrmJpOr7o+zm0rXK+hNnT5l9RHkCkQ==", + "license": "MIT" + }, + "node_modules/pg-int8": { + "version": "1.0.1", + "resolved": "https://registry.npmjs.org/pg-int8/-/pg-int8-1.0.1.tgz", + "integrity": "sha512-WCtabS6t3c8SkpDBUlb1kjOs7l66xsGdKpIPZsg4wR+B3+u9UAum2odSsF9tnvxg80h4ZxLWMy4pRjOsFIqQpw==", + "license": "ISC", + "engines": { + "node": ">=4.0.0" + } + }, + "node_modules/pg-pool": { + "version": "3.13.0", + "resolved": "https://registry.npmjs.org/pg-pool/-/pg-pool-3.13.0.tgz", + "integrity": "sha512-gB+R+Xud1gLFuRD/QgOIgGOBE2KCQPaPwkzBBGC9oG69pHTkhQeIuejVIk3/cnDyX39av2AxomQiyPT13WKHQA==", + "license": "MIT", + "peerDependencies": { + "pg": ">=8.0" + } + }, + "node_modules/pg-protocol": { + "version": "1.13.0", + "resolved": "https://registry.npmjs.org/pg-protocol/-/pg-protocol-1.13.0.tgz", + "integrity": "sha512-zzdvXfS6v89r6v7OcFCHfHlyG/wvry1ALxZo4LqgUoy7W9xhBDMaqOuMiF3qEV45VqsN6rdlcehHrfDtlCPc8w==", + "license": "MIT" + }, + "node_modules/pg-types": { + "version": "2.2.0", + "resolved": "https://registry.npmjs.org/pg-types/-/pg-types-2.2.0.tgz", + "integrity": "sha512-qTAAlrEsl8s4OiEQY69wDvcMIdQN6wdz5ojQiOy6YRMuynxenON0O5oCpJI6lshc6scgAY8qvJ2On/p+CXY0GA==", + "license": "MIT", + "dependencies": { + "pg-int8": "1.0.1", + "postgres-array": "~2.0.0", + "postgres-bytea": "~1.0.0", + "postgres-date": "~1.0.4", + "postgres-interval": "^1.1.0" + }, + "engines": { + "node": ">=4" + } + }, + "node_modules/pgpass": { + "version": "1.0.5", + "resolved": "https://registry.npmjs.org/pgpass/-/pgpass-1.0.5.tgz", + "integrity": "sha512-FdW9r/jQZhSeohs1Z3sI1yxFQNFvMcnmfuj4WBMUTxOrAyLMaTcE1aAMBiTlbMNaXvBCQuVi0R7hd8udDSP7ug==", + "license": "MIT", + "dependencies": { + "split2": "^4.1.0" + } + }, "node_modules/pkce-challenge": { "version": "5.0.1", "resolved": "https://registry.npmjs.org/pkce-challenge/-/pkce-challenge-5.0.1.tgz", @@ -1689,6 +1791,45 @@ "node": ">=16.20.0" } }, + "node_modules/postgres-array": { + "version": "2.0.0", + "resolved": "https://registry.npmjs.org/postgres-array/-/postgres-array-2.0.0.tgz", + "integrity": "sha512-VpZrUqU5A69eQyW2c5CA1jtLecCsN2U/bD6VilrFDWq5+5UIEVO7nazS3TEcHf1zuPYO/sqGvUvW62g86RXZuA==", + "license": "MIT", + "engines": { + "node": ">=4" + } + }, + "node_modules/postgres-bytea": { + "version": "1.0.1", + "resolved": "https://registry.npmjs.org/postgres-bytea/-/postgres-bytea-1.0.1.tgz", + "integrity": "sha512-5+5HqXnsZPE65IJZSMkZtURARZelel2oXUEO8rH83VS/hxH5vv1uHquPg5wZs8yMAfdv971IU+kcPUczi7NVBQ==", + "license": "MIT", + "engines": { + "node": ">=0.10.0" + } + }, + "node_modules/postgres-date": { + "version": "1.0.7", + "resolved": "https://registry.npmjs.org/postgres-date/-/postgres-date-1.0.7.tgz", + "integrity": "sha512-suDmjLVQg78nMK2UZ454hAG+OAW+HQPZ6n++TNDUX+L0+uUlLywnoxJKDou51Zm+zTCjrCl0Nq6J9C5hP9vK/Q==", + "license": "MIT", + "engines": { + "node": ">=0.10.0" + } + }, + "node_modules/postgres-interval": { + "version": "1.2.0", + "resolved": "https://registry.npmjs.org/postgres-interval/-/postgres-interval-1.2.0.tgz", + "integrity": "sha512-9ZhXKM/rw350N1ovuWHbGxnGh/SNJ4cnxHiM0rxE4VN41wsg8P8zWn9hv/buK00RP4WvlOyr/RBDiptyxVbkZQ==", + "license": "MIT", + "dependencies": { + "xtend": "^4.0.0" + }, + "engines": { + "node": ">=0.10.0" + } + }, "node_modules/proxy-addr": { "version": "2.0.7", "resolved": "https://registry.npmjs.org/proxy-addr/-/proxy-addr-2.0.7.tgz", @@ -1926,6 +2067,15 @@ "url": "https://github.com/sponsors/ljharb" } }, + "node_modules/split2": { + "version": "4.2.0", + "resolved": "https://registry.npmjs.org/split2/-/split2-4.2.0.tgz", + "integrity": "sha512-UcjcJOWknrNkF6PLX83qcHM6KHgVKNkV62Y8a5uYDVv9ydGQVwAHMKqHdJje1VTWpljG0WYpCDhrCdAOYH4TWg==", + "license": "ISC", + "engines": { + "node": ">= 10.x" + } + }, "node_modules/statuses": { "version": "2.0.2", "resolved": "https://registry.npmjs.org/statuses/-/statuses-2.0.2.tgz", @@ -2078,6 +2228,15 @@ } } }, + "node_modules/xtend": { + "version": "4.0.2", + "resolved": "https://registry.npmjs.org/xtend/-/xtend-4.0.2.tgz", + "integrity": "sha512-LKYU1iAXJXUgAXn9URjiu+MWhyUXHsvfp7mcuYm9dSUKK0/CjtrUwFAxD82/mCWbtLsGjFIad0wIsod4zrTAEQ==", + "license": "MIT", + "engines": { + "node": ">=0.4" + } + }, "node_modules/zod": { "version": "3.25.76", "resolved": "https://registry.npmjs.org/zod/-/zod-3.25.76.tgz", @@ -2108,7 +2267,8 @@ "name": "@voidbot/core", "version": "0.1.0", "dependencies": { - "@voidbot/shared": "0.1.0" + "@voidbot/shared": "0.1.0", + "pg": "^8.20.0" } }, "packages/providers": { diff --git a/package.json b/package.json index 565ca14..8a3f8aa 100644 --- a/package.json +++ b/package.json @@ -13,6 +13,7 @@ "dev:mcp": "tsx apps/worker/src/mcp-server.ts", "dev:worker": "tsx apps/worker/src/index.ts", "stack:start": "powershell -ExecutionPolicy Bypass -File scripts/start-voidbot-stack.ps1", + "postgres:up": "docker compose -f infra/postgres/docker-compose.yml up -d", "codex:mcp:install": "tsx scripts/install-codex-mcp.ts", "codex:mcp:remove": "tsx scripts/remove-codex-mcp.ts", "qdrant:setup": "powershell -ExecutionPolicy Bypass -File scripts/setup-docker-qdrant.ps1", @@ -26,6 +27,7 @@ }, "devDependencies": { "@types/node": "^24.0.0", + "@types/pg": "^8.20.0", "tsx": "^4.20.0", "typescript": "^5.9.0" } diff --git a/packages/config/src/index.ts b/packages/config/src/index.ts index 0f87e8a..5a4e5a1 100644 --- a/packages/config/src/index.ts +++ b/packages/config/src/index.ts @@ -42,6 +42,7 @@ const envSchema = z.object({ DISCORD_APPLICATION_ID: optionalNonEmptyString, DISCORD_GUILD_ID: optionalNonEmptyString, DISCORD_OWNER_ID: z.string().min(1, "DISCORD_OWNER_ID is required"), + STATE_STORAGE_BACKEND: z.enum(["file", "postgres"]).default("postgres"), DATABASE_DSN: z.string().min(1).default("postgres://voidbot:voidbot@localhost:5432/voidbot"), VECTOR_STORE_KIND: z.enum(["local_json", "qdrant"]).default("local_json"), VECTOR_STORE_PATH: z.string().min(1).default(".voidbot/history-vector-store.json"), @@ -102,6 +103,7 @@ export interface AppConfig { applicationId?: string; developmentGuildId?: string; ownerDiscordId: string; + stateStorageBackend: "file" | "postgres"; databaseDsn: string; enabledProviders: ProviderName[]; ownerCodexMode: OwnerCodexMode; @@ -260,6 +262,7 @@ export function loadConfig(env: NodeJS.ProcessEnv = process.env): AppConfig { applicationId: parsed.DISCORD_APPLICATION_ID, developmentGuildId: parsed.DISCORD_GUILD_ID, ownerDiscordId: parsed.DISCORD_OWNER_ID, + stateStorageBackend: parsed.STATE_STORAGE_BACKEND, databaseDsn: parsed.DATABASE_DSN, enabledProviders: parseProviders(parsed.ENABLED_PROVIDERS), ownerCodexMode: parseOwnerCodexMode(parsed.OWNER_CODEX_MODE), diff --git a/packages/core/package.json b/packages/core/package.json index 81d3a5a..9c24fa4 100644 --- a/packages/core/package.json +++ b/packages/core/package.json @@ -5,6 +5,7 @@ "main": "dist/index.js", "types": "dist/index.d.ts", "dependencies": { - "@voidbot/shared": "0.1.0" + "@voidbot/shared": "0.1.0", + "pg": "^8.20.0" } } diff --git a/packages/core/sql/bootstrap.sql b/packages/core/sql/bootstrap.sql index 2f6d835..28a700c 100644 --- a/packages/core/sql/bootstrap.sql +++ b/packages/core/sql/bootstrap.sql @@ -9,7 +9,7 @@ create table if not exists channel_policies ( guild_id text not null, channel_id text not null, indexing_enabled boolean not null default false, - provider_allowlist text[] not null default '{}', + provider_allowlist text[] not null default '{}'::text[], primary key (guild_id, channel_id) ); @@ -21,31 +21,39 @@ create table if not exists provider_configs ( updated_at timestamptz not null default now() ); +create table if not exists style_packs ( + name text primary key, + file_path text not null, + enabled boolean not null default true, + created_at timestamptz not null default now(), + updated_at timestamptz not null default now() +); + +create table if not exists state_migrations ( + name text primary key, + applied_at timestamptz not null default now() +); + create table if not exists jobs ( id uuid primary key, + request_message_id text, command_name text not null, state text not null, provider_name text not null, requester_discord_id text not null, guild_id text, - channel_id text not null, - prompt text not null, - context_json jsonb not null, - summary text, - final_response text, - created_at timestamptz not null default now(), - updated_at timestamptz not null default now() + output_channel_id text not null, + created_at timestamptz not null, + updated_at timestamptz not null, + job_json jsonb not null ); -create table if not exists job_approvals ( - id bigserial primary key, - job_id uuid not null references jobs(id) on delete cascade, - stage text not null, - status text not null, - actor_discord_id text not null, - reason text, - created_at timestamptz not null default now() -); +create index if not exists jobs_state_created_idx + on jobs (state, created_at); + +create unique index if not exists jobs_request_dedupe_idx + on jobs (request_message_id, command_name, provider_name, output_channel_id) + where request_message_id is not null; create table if not exists provider_runs ( id bigserial primary key, @@ -67,36 +75,36 @@ create table if not exists tool_invocations ( ); create table if not exists audit_events ( - id bigserial primary key, + id uuid primary key, job_id uuid references jobs(id) on delete set null, actor_discord_id text, + provider_name text, event_type text not null, - event_json jsonb not null, - created_at timestamptz not null default now() + event_timestamp timestamptz not null default now(), + event_json jsonb not null ); -create table if not exists message_archives ( - message_id text primary key, +create index if not exists audit_events_job_timestamp_idx + on audit_events (job_id, event_timestamp desc); + +create table if not exists interaction_memory_events ( + id text primary key, + actor_id text not null, + actor_name text not null, + source_kind text not null, guild_id text, channel_id text not null, - author_discord_id text not null, - content text not null, - metadata_json jsonb not null default '{}'::jsonb, - created_at timestamptz not null + channel_name text, + command_name text, + prompt text not null, + excerpt text not null, + summary text not null, + sentiment text not null, + score integer not null, + tags text[] not null default '{}'::text[], + event_timestamp timestamptz not null, + event_json jsonb not null ); -create table if not exists embedding_chunks ( - chunk_id text primary key, - message_id text not null references message_archives(message_id) on delete cascade, - text_content text not null, - metadata_json jsonb not null default '{}'::jsonb, - embedding vector(1536) -); - -create table if not exists style_packs ( - name text primary key, - file_path text not null, - enabled boolean not null default true, - created_at timestamptz not null default now(), - updated_at timestamptz not null default now() -); +create index if not exists interaction_memory_events_actor_timestamp_idx + on interaction_memory_events (actor_id, event_timestamp desc); diff --git a/packages/core/src/audit-log.ts b/packages/core/src/audit-log.ts index 075e1bb..a01c932 100644 --- a/packages/core/src/audit-log.ts +++ b/packages/core/src/audit-log.ts @@ -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 & Partial>; -export class FileAuditLog { +export interface AuditLog { + record(input: AuditEventInput): Promise; +} + +export class FileAuditLog implements AuditLog { public constructor(private readonly auditFile: string) {} public async record(input: AuditEventInput): Promise { @@ -26,5 +30,23 @@ export class FileAuditLog { return event; } -} + public async listEvents(): Promise { + 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; + } + } +} diff --git a/packages/core/src/index.ts b/packages/core/src/index.ts index 86d7bfe..fb10cb4 100644 --- a/packages/core/src/index.ts +++ b/packages/core/src/index.ts @@ -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"; diff --git a/packages/core/src/interaction-memory-logic.ts b/packages/core/src/interaction-memory-logic.ts new file mode 100644 index 0000000..26725c0 --- /dev/null +++ b/packages/core/src/interaction-memory-logic.ts @@ -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(); + + 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, + 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 { + 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)); +} diff --git a/packages/core/src/interaction-memory.ts b/packages/core/src/interaction-memory.ts index aa363a1..3a92485 100644 --- a/packages/core/src/interaction-memory.ts +++ b/packages/core/src/interaction-memory.ts @@ -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; + recordInteraction(input: RecordInteractionInput): Promise; } -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 = Promise.resolve(); public constructor(private readonly filePath: string) {} @@ -59,52 +43,41 @@ export class FileInteractionMemoryBank { ): Promise { 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 { + await this.writeChain; + const store = await this.readUnlocked(); + return store.profiles.map((profile) => structuredClone(profile)); + } + private async serialize(operation: () => Promise): Promise { 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(); - - 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, - 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 { - 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; } diff --git a/packages/core/src/job-queue.ts b/packages/core/src/job-queue.ts index 4a78968..55acdce 100644 --- a/packages/core/src/job-queue.ts +++ b/packages/core/src/job-queue.ts @@ -16,6 +16,25 @@ interface JobStore { jobs: JobRecord[]; } +export interface JobQueue { + createJob(input: CreateJobInput): Promise; + getJob(jobId: string): Promise; + listByStates(states: JobState[]): Promise; + listRunnableJobs(): Promise; + claimRunnableJobs(limit?: number): Promise; + approveRun(jobId: string, actorId: string): Promise; + rejectJob(jobId: string, actorId: string, reason?: string): Promise; + markRunning(jobId: string): Promise; + markAwaitingPostApproval( + jobId: string, + artifactPaths: Record, + summary: string, + ): Promise; + markFailed(jobId: string, error: string): Promise; + completeJob(jobId: string, actorId: string, finalResponse: string): Promise; + completeJobDirect(jobId: string, finalResponse: string): Promise; +} + 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 { @@ -228,6 +247,11 @@ export class FileBackedJobQueue { }); } + public async exportJobs(): Promise { + const store = await this.readStore(); + return store.jobs.map((job) => cloneJob(job)); + } + private async updateJob( jobId: string, updater: (job: JobRecord) => void, diff --git a/packages/core/src/state-storage.ts b/packages/core/src/state-storage.ts new file mode 100644 index 0000000..9b657e0 --- /dev/null +++ b/packages/core/src/state-storage.ts @@ -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; +} + +export async function createStateStorage( + config: StateStorageConfig, +): Promise { + 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 { + 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; + } + + return summarizeInteractionProfile( + actorId, + events[events.length - 1].actorName, + events, + ); + } + + public async recordInteraction( + input: RecordInteractionInput, + ): Promise { + 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 { + 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); + } +} + +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), + ); +} diff --git a/scripts/start-voidbot-stack.ps1 b/scripts/start-voidbot-stack.ps1 index f725a5b..eb9a448 100644 --- a/scripts/start-voidbot-stack.ps1 +++ b/scripts/start-voidbot-stack.ps1 @@ -117,6 +117,31 @@ function Wait-Until { throw $FailureMessage } +function Test-TcpEndpoint { + param( + [Parameter(Mandatory = $true)] + [string] $HostName, + [Parameter(Mandatory = $true)] + [int] $Port + ) + + $client = [System.Net.Sockets.TcpClient]::new() + + try { + $connectTask = $client.ConnectAsync($HostName, $Port) + + if (-not $connectTask.Wait(1500)) { + return $false + } + + return $client.Connected + } catch { + return $false + } finally { + $client.Dispose() + } +} + function Test-IsLocalUrl { param( [Parameter(Mandatory = $true)] @@ -127,6 +152,52 @@ function Test-IsLocalUrl { return $uri.Host -in @("127.0.0.1", "localhost", "::1") } +function Test-IsLocalHostName { + param( + [Parameter(Mandatory = $true)] + [string] $HostName + ) + + return $HostName -in @("127.0.0.1", "localhost", "::1") +} + +function Parse-DatabaseDsn { + param( + [Parameter(Mandatory = $true)] + [string] $Dsn + ) + + $uri = [Uri]$Dsn + $userInfo = if ([string]::IsNullOrWhiteSpace($uri.UserInfo)) { @() } else { $uri.UserInfo.Split(":", 2) } + + return @{ + host = $uri.Host + port = if ($uri.Port -gt 0) { $uri.Port } else { 5432 } + database = $uri.AbsolutePath.TrimStart("/") + username = if ($userInfo.Length -ge 1) { [Uri]::UnescapeDataString($userInfo[0]) } else { "voidbot" } + password = if ($userInfo.Length -ge 2) { [Uri]::UnescapeDataString($userInfo[1]) } else { "voidbot" } + } +} + +function Get-DockerContainerHealth { + param( + [Parameter(Mandatory = $true)] + [string] $ContainerName + ) + + try { + $status = & docker inspect --format "{{if .State.Health}}{{.State.Health.Status}}{{else}}none{{end}}" $ContainerName 2>$null + + if ($LASTEXITCODE -ne 0) { + return $null + } + + return $status.Trim() + } catch { + return $null + } +} + function Ensure-DockerOnPath { $dockerCliDir = "C:\Program Files\Docker\Docker\resources\bin" @@ -181,6 +252,65 @@ function Ensure-Qdrant { } } +function Ensure-Postgres { + param( + [Parameter(Mandatory = $true)] + [string] $RepoRoot, + [Parameter(Mandatory = $true)] + [string] $DatabaseDsn + ) + + $database = Parse-DatabaseDsn -Dsn $DatabaseDsn + + if (Test-TcpEndpoint -HostName $database.host -Port $database.port) { + return @{ + dsn = $DatabaseDsn + host = $database.host + port = $database.port + database = $database.database + startedByScript = $false + healthy = $true + } + } + + if (-not (Test-IsLocalHostName -HostName $database.host)) { + throw "Postgres is not reachable at $($database.host):$($database.port), and the host is not local enough for Docker auto-start." + } + + Ensure-DockerOnPath + + $composeFile = Join-Path $RepoRoot "infra\postgres\docker-compose.yml" + $env:VOIDBOT_POSTGRES_PORT = [string]$database.port + $env:VOIDBOT_POSTGRES_DB = $database.database + $env:VOIDBOT_POSTGRES_USER = $database.username + $env:VOIDBOT_POSTGRES_PASSWORD = $database.password + + & docker compose -f $composeFile up -d + + if ($LASTEXITCODE -ne 0) { + throw "Docker Compose failed to start Postgres." + } + + Wait-Until -Condition { + $health = Get-DockerContainerHealth -ContainerName "voidbot-postgres" + + if ([string]::IsNullOrWhiteSpace($health) -or $health -eq "none") { + return Test-TcpEndpoint -HostName $database.host -Port $database.port + } + + return $health -eq "healthy" + } -FailureMessage "Postgres did not become reachable at $($database.host):$($database.port) in time." + + return @{ + dsn = $DatabaseDsn + host = $database.host + port = $database.port + database = $database.database + startedByScript = $true + healthy = $true + } +} + function Ensure-OllamaEndpoint { param( [Parameter(Mandatory = $true)] @@ -332,6 +462,8 @@ $status = @{ Write-StatusFile -Path $statusPath -Status $status $vectorStoreKind = if ($config.ContainsKey("VECTOR_STORE_KIND")) { $config["VECTOR_STORE_KIND"] } else { "local_json" } +$stateStorageBackend = if ($config.ContainsKey("STATE_STORAGE_BACKEND")) { $config["STATE_STORAGE_BACKEND"] } else { "postgres" } +$databaseDsn = if ($config.ContainsKey("DATABASE_DSN")) { $config["DATABASE_DSN"] } else { "postgres://voidbot:voidbot@localhost:5432/voidbot" } $qdrantUrl = if ($config.ContainsKey("QDRANT_URL")) { $config["QDRANT_URL"] } else { "http://127.0.0.1:6333" } $ragBackend = if ($config.ContainsKey("RAG_EMBEDDING_BACKEND")) { $config["RAG_EMBEDDING_BACKEND"] } else { "ollama" } $ragOllamaBaseUrl = if ($config.ContainsKey("RAG_OLLAMA_BASE_URL")) { $config["RAG_OLLAMA_BASE_URL"] } else { "http://127.0.0.1:11434" } @@ -340,6 +472,16 @@ $localLlmEnabled = if ($config.ContainsKey("ENABLED_PROVIDERS")) { $config["ENAB $localLlmBaseUrl = if ($config.ContainsKey("LOCAL_LLM_OLLAMA_BASE_URL")) { $config["LOCAL_LLM_OLLAMA_BASE_URL"] } else { "http://127.0.0.1:11434" } $localLlmModel = if ($config.ContainsKey("LOCAL_LLM_OLLAMA_MODEL")) { $config["LOCAL_LLM_OLLAMA_MODEL"] } else { "qwen3.5:9b" } +if ($stateStorageBackend -eq "postgres") { + $status.postgres = Ensure-Postgres -RepoRoot $repoRoot -DatabaseDsn $databaseDsn + Write-StatusFile -Path $statusPath -Status $status +} else { + $status.postgres = @{ + skipped = $true + reason = "STATE_STORAGE_BACKEND is not postgres." + } +} + if ($vectorStoreKind -eq "qdrant") { $status.qdrant = Ensure-Qdrant -RepoRoot $repoRoot -QdrantUrl $qdrantUrl Write-StatusFile -Path $statusPath -Status $status