From 6d3bde1586b1d42dc793117079ff33a785770004 Mon Sep 17 00:00:00 2001 From: Nashchennc <59103980+NashChennc@users.noreply.github.com> Date: Fri, 6 Mar 2026 13:44:53 +0800 Subject: [PATCH] refactor(satori-bot): migrate persistence to adapter (#1155) --- pnpm-lock.yaml | 29 +- services/satori-bot/.gitignore | 16 ++ services/satori-bot/README.md | 2 +- services/satori-bot/docs/HANDLER.md | 11 +- services/satori-bot/docs/PERSISTENCE.md | 57 ++-- services/satori-bot/drizzle.config.ts | 11 + .../drizzle/0000_cooing_captain_flint.sql | 31 +++ .../drizzle/meta/0000_snapshot.json | 201 ++++++++++++++ .../satori-bot/drizzle/meta/_journal.json | 13 + services/satori-bot/package.json | 9 +- services/satori-bot/src/adapter/satori/api.ts | 16 +- .../satori-bot/src/adapter/satori/client.ts | 32 ++- .../satori-bot/src/adapter/satori/schema.ts | 102 +++++++ .../satori-bot/src/adapter/satori/types.ts | 14 +- .../src/capabilities/actions/read-messages.ts | 26 +- .../src/capabilities/actions/send-message.ts | 15 +- .../src/capabilities/actions/system.ts | 8 +- .../satori-bot/src/capabilities/definition.ts | 6 +- services/satori-bot/src/config.ts | 63 +++++ services/satori-bot/src/core/constants.ts | 3 +- services/satori-bot/src/core/dispatcher.ts | 22 +- services/satori-bot/src/core/loop/queue.ts | 11 +- .../satori-bot/src/core/loop/scheduler.ts | 259 +++++++++++------- .../satori-bot/src/core/planner/llm-client.ts | 52 ++-- .../satori-bot/src/core/session/context.ts | 16 +- services/satori-bot/src/core/types.ts | 91 +++--- services/satori-bot/src/core/utils.ts | 39 ++- services/satori-bot/src/index.ts | 32 ++- services/satori-bot/src/lib/db.ts | 218 ++++++++++----- services/satori-bot/src/lib/schema.ts | 35 +++ services/satori-bot/todolist.md | 34 +++ 31 files changed, 1120 insertions(+), 354 deletions(-) create mode 100644 services/satori-bot/.gitignore create mode 100644 services/satori-bot/drizzle.config.ts create mode 100644 services/satori-bot/drizzle/0000_cooing_captain_flint.sql create mode 100644 services/satori-bot/drizzle/meta/0000_snapshot.json create mode 100644 services/satori-bot/drizzle/meta/_journal.json create mode 100644 services/satori-bot/src/adapter/satori/schema.ts create mode 100644 services/satori-bot/src/config.ts create mode 100644 services/satori-bot/src/lib/schema.ts create mode 100644 services/satori-bot/todolist.md diff --git a/pnpm-lock.yaml b/pnpm-lock.yaml index 7e88366e9..4ffea3e6d 100644 --- a/pnpm-lock.yaml +++ b/pnpm-lock.yaml @@ -3375,6 +3375,9 @@ importers: services/satori-bot: dependencies: + '@electric-sql/pglite': + specifier: 'catalog:' + version: 0.3.15 '@guiiai/logg': specifier: 'catalog:' version: 1.2.11 @@ -3396,15 +3399,18 @@ importers: best-effort-json-parser: specifier: ^1.2.1 version: 1.2.1 + drizzle-orm: + specifier: ^0.45.1 + version: 0.45.1(@electric-sql/pglite@0.3.15)(@opentelemetry/api@1.9.0)(@prisma/client@5.22.0)(@types/pg@8.16.0)(better-sqlite3@12.5.0)(kysely@0.28.9)(pg@8.19.0)(postgres@3.4.8) es-toolkit: specifier: ^1.44.0 version: 1.44.0 - lowdb: - specifier: ^7.0.1 - version: 7.0.1 nanoid: specifier: ^5.1.6 version: 5.1.6 + valibot: + specifier: ^1.2.0 + version: 1.2.0(typescript@5.9.3) ws: specifier: ^8.19.0 version: 8.19.0(bufferutil@4.1.0)(utf-8-validate@5.0.10) @@ -3415,6 +3421,9 @@ importers: '@types/ws': specifier: ^8.18.1 version: 8.18.1 + drizzle-kit: + specifier: ^0.31.9 + version: 0.31.9 tsx: specifier: ^4.21.0 version: 4.21.0 @@ -13327,10 +13336,6 @@ packages: longest-streak@3.1.0: resolution: {integrity: sha512-9Ri+o0JYgehTaVBBDoMqIl8GXtbWg711O3srftcHhZ0dqnETqLaoIK0x17fUw9rFSlK/0NlsKe0Ahhyl5pXE2g==} - lowdb@7.0.1: - resolution: {integrity: sha512-neJAj8GwF0e8EpycYIDFqEPcx9Qz4GUho20jWFR7YiFeXzF1YMLdxB36PypcTSPMA+4+LvgyMacYhlr18Zlymw==} - engines: {node: '>=18'} - lowercase-keys@2.0.0: resolution: {integrity: sha512-tqNXrS78oMOE73NMxK4EMLQsQowWf8jKooH9g7xPavRT706R6bkQJ6DY2Te7QukaZsulxa30wQ7bk0pm4XiHmA==} engines: {node: '>=8'} @@ -15468,10 +15473,6 @@ packages: resolution: {integrity: sha512-reExS1kSGoElkextOcPkel4NE99S0BWxjUHQeDFnR8S993JxpPX7KU4MNmO19NXhlJp+8dmdCbKQVNgLJh2teA==} engines: {node: '>=18'} - steno@4.0.2: - resolution: {integrity: sha512-yhPIQXjrlt1xv7dyPQg2P17URmXbuM5pdGkpiMB3RenprfiBlvK415Lctfe0eshk90oA7/tNq7WEiMK8RSP39A==} - engines: {node: '>=18'} - store2@2.14.4: resolution: {integrity: sha512-srTItn1GOvyvOycgxjAnPA63FZNwy0PTyUBFMHRM+hVFltAeoh0LmNBz9SZqUS9mMqGk8rfyWyXn3GH5ReJ8Zw==} @@ -28088,10 +28089,6 @@ snapshots: longest-streak@3.1.0: {} - lowdb@7.0.1: - dependencies: - steno: 4.0.2 - lowercase-keys@2.0.0: {} lru-cache@10.4.3: {} @@ -30925,8 +30922,6 @@ snapshots: stdin-discarder@0.3.1: {} - steno@4.0.2: {} - store2@2.14.4: {} streamx@2.23.0: diff --git a/services/satori-bot/.gitignore b/services/satori-bot/.gitignore new file mode 100644 index 000000000..aa8ac3ce2 --- /dev/null +++ b/services/satori-bot/.gitignore @@ -0,0 +1,16 @@ +# Database Persistence (DO NOT COMMIT) +data/ +data/pglite-db/ +data/db.json + +# Environment Secrets +.env +.env.local + +# Node.js +node_modules/ +dist/ +*.log + +# OS Files +.DS_Store diff --git a/services/satori-bot/README.md b/services/satori-bot/README.md index a29d54a6e..b62a0aa96 100644 --- a/services/satori-bot/README.md +++ b/services/satori-bot/README.md @@ -72,5 +72,5 @@ pnpm --filter @proj-airi/satori-bot start ## Key Locations * **Persona & System Prompts**: `src/core/planner/prompts/*.velin.md` -* **Database (JSON)**: `data/db.json` (See *PERSISTENCE.md* for limitations) +* **Database (PGlite)**: `data/pglite-db` (See *PERSISTENCE.md* for architecture) * **Action Logic**: `src/capabilities/actions/` diff --git a/services/satori-bot/docs/HANDLER.md b/services/satori-bot/docs/HANDLER.md index f215e51b1..4a5318a94 100644 --- a/services/satori-bot/docs/HANDLER.md +++ b/services/satori-bot/docs/HANDLER.md @@ -21,7 +21,7 @@ The bot operates on a **Event-Driven + Autonomous Loop** hybrid model: 3. **Deduplication**: The system checks the `processedIds` set (key: `channelId-messageId`) to prevent double-processing. 4. **Enqueuing**: * The raw `event` is wrapped into a `{ event, status: 'ready' }` object. - * It is pushed into `botContext.eventQueue`. + * It is pushed into `botContext.eventQueue` and persisted to the database via `pushToEventQueue`. * **Key Data**: `event.message.content`, `event.user.id`, `event.channel.id`. ### Phase 2: Consumption & Anchoring @@ -34,10 +34,10 @@ When the system processing lock is free, it consumes events from the `eventQueue * Calls `ensureChatContext` (in `src/core/session/context.ts`) to load or create the in-memory `ChatContext` for that channel. * **Anchor**: The `event.channel.id` is the primary key for all context. 2. **Filtering**: - * Checks `selfId`. If the sender is the bot itself, the event is discarded (not counted as unread) to prevent infinite loops. + * Checks `selfId`. If the sender is the bot itself, the event is removed from the queue and discarded (not counted as unread). 3. **State Update (The "Unread Pool")**: - * The event is pushed into `botContext.unreadEvents[channelId]`. - * *Note:* This step does **not** just store the message; it marks the event as a "pending observation object". + * The event is pushed into `botContext.unreadEvents[channelId]` and persisted to the database via `pushToUnreadEvents`. + * The event is then removed from the database queue via `removeFromEventQueue`. 4. **Loop Trigger**: * Immediately calls `loopIterationForChannel`, waking up the Agent Loop for this specific channel. @@ -76,4 +76,5 @@ The system looks up the corresponding Handler in `globalRegistry` based on the J * `dispatchAction` returns an `ActionResult` containing a `shouldContinue` flag. * If `shouldContinue` is true (e.g., usually true after reading messages, as a reply is expected), the scheduler waits for `LOOP_CONTINUE_DELAY_MS` (default 2.5s) and then recursively calls `handleLoopStep`. -* **Termination**: The loop stops only when the LLM selects the `continue` action (Wait/Stop) or the `break` action. +* **Hard Limit**: To prevent infinite loops caused by LLM hallucinations or API abuse, each loop is capped at `MAX_LOOP_ITERATIONS = 5`. Reaching this limit will force the loop to break. +* **Termination**: The loop stops when the LLM selects a terminal action, the iteration limit is reached, or the `shouldContinue` flag becomes false. diff --git a/services/satori-bot/docs/PERSISTENCE.md b/services/satori-bot/docs/PERSISTENCE.md index ac4200296..bd291274f 100644 --- a/services/satori-bot/docs/PERSISTENCE.md +++ b/services/satori-bot/docs/PERSISTENCE.md @@ -1,46 +1,47 @@ ## **Architecture Status Report: Memory & Persistence** -**Date:** February 9, 2026 (Refactored) -**Component:** State Management Layer +**Date:** March 6, 2026 (Refactored) +**Component:** State Management Layer (Drizzle + PGlite) ### **1. Memory Architecture (RAM)** -The bot utilizes a **Memory-First** strategy, where the active state is fully resident in the Node.js heap. +The bot utilizes a **Memory-First** strategy for active chat sessions, while persisting critical queue and message data to disk. -* **Storage Mechanism**: All chat contexts are stored in a native `Map` within the `BotContext` object (`src/core/types.ts`). +* **Storage Mechanism**: Active chat contexts are stored in a native `Map` within the `BotContext` object (`src/core/types.ts`). * **Lifecycle Management**: * **Creation**: Contexts are lazy-loaded via `ensureChatContext` in `src/core/session/context.ts` upon receiving a message. - * **Retention**: There is currently **no garbage collection (GC)** mechanism. Once a channel is loaded, its context remains in memory indefinitely until the process terminates. + * **Retention**: Currently, contexts remain in memory until process termination. History is trimmed during the loop. * **Context Trimming**: * Executed within `handleLoopStep` in `src/core/loop/scheduler.ts`. - * Individual channels enforce a strict limit on history length (Default: 20 messages, 50 actions) to prevent single-channel bloat. - * **Risk**: The architecture is susceptible to memory leaks (OOM) as the number of unique channels increases over time. + * Individual channels enforce strict limits: `MAX_ACTIONS_IN_CONTEXT = 50`, `ACTIONS_KEEP_ON_TRIM = 20`. + * Message history is dynamically fetched from the database (last 10 messages) to keep the LLM context lean. -### **2. Persistence Architecture (Disk)** +### **2. Persistence Architecture (Database)** -The bot uses a file-based logging system primarily for archival purposes and basic metadata recovery upon restart, rather than for active state management. +The bot has migrated from `lowdb` (JSON) to **PGlite** (PostgreSQL in WASM/Node) with **Drizzle ORM** for robust state management and high-performance I/O. -* **Technology**: `lowdb` with a JSON file adapter. -* **Location**: `src/lib/db.ts` -> `data/db.json`. -* **Data Structure**: - * `channels`: Stores metadata like Channel ID, Platform, and SelfID. - * `messages`: A global, flattened array of messages. -* **Write Strategy**: **Synchronous full-file serialization**. Every new message triggers a complete rewrite of the JSON file to disk. -* **Retention Policy**: A global hard limit of **1000 messages** is enforced. When the limit is reached, the oldest messages are discarded regardless of which channel they belong to. -* **Recovery Logic**: Upon restart, `ensureChatContext` queries `db.channels` to restore the channel's `platform` and `selfId`, but it **does not** load historical messages into the in-memory context. +* **Technology**: [PGlite](https://pglite.dev/) + [Drizzle ORM](https://orm.drizzle.team/). +* **Location**: `data/` directory (configured via `DB_PATH` in `.env.local`). +* **Schema (`src/lib/schema.ts`)**: + * `channels`: Metadata for discovered channels (ID, name, platform, self_id). + * `messages`: Persistent message log with indexing on `channel_id` and `timestamp`. + * `event_queue`: Persistent queue for incoming Satori events awaiting processing. + * `unread_events`: Persistent store for events marked as unread for each channel. +* **Optimized I/O Strategy**: + * **Incremental Updates**: Unlike the previous "full-rewrite" approach, the bot now uses targeted SQL operations. + * **Queue Management**: Individual items are added (`pushToEventQueue`) and removed (`removeFromEventQueue`) by ID. + * **Unread Tracking**: Unread messages are persisted incrementally (`pushToUnreadEvents`) and cleared per channel (`clearUnreadEventsForChannel`). +* **Migrations**: Managed via `drizzle-kit`. Migrations are automatically applied on startup in `src/lib/db.ts`. -### **3. State Consistency** +### **3. State Consistency & Recovery** -There is a significant desynchronization between the ephemeral memory state and the persistent disk state. +The gap between ephemeral memory and persistent disk state has been significantly narrowed. -* **In-Memory State (Rich)**: Contains the full "Chain of Thought" (System prompts, reasoning steps, `AbortController` handles, pending Promises, `Action History`). -* **On-Disk State (Flat)**: Contains only raw user content and final bot responses. -* **Impact**: A process restart results in a **Hard Context Reset**. The bot loses all active "trains of thought" and task states, falling back to a state driven solely by new incoming messages. +* **Durable Queue**: The `eventQueue` and `unreadEvents` are fully persisted. If the bot crashes, it resumes processing the queue from where it left off. +* **Message History**: The LLM's conversation history is reconstructed from the indexed `messages` table in the database, ensuring continuity across restarts. +* **Hard Reset Mitigation**: While `AbortController` handles are still lost on restart, the core task queue and conversation context remain intact. -### **4. Future Roadmap (WIP)** +### **4. Configuration** -We are planning to implement a "Small Memory" storage scheme to improve robustness, featuring: - -1. **Indiscriminate Event Storage**: Storing all events without preemptive filtering. -2. **Event Activation Query**: Triggering queries based on specific event activation signals. -3. **Dynamic Context Filtering**: Reconstructing same-group contexts via query-time filtering rather than pre-computed buckets. +Database settings are managed through `src/config.ts`: +* `DB_PATH`: Path to the PGlite data directory (default: `data/pglite-db`). diff --git a/services/satori-bot/drizzle.config.ts b/services/satori-bot/drizzle.config.ts new file mode 100644 index 000000000..eef9d0d70 --- /dev/null +++ b/services/satori-bot/drizzle.config.ts @@ -0,0 +1,11 @@ +import { defineConfig } from 'drizzle-kit' + +export default defineConfig({ + schema: './src/lib/schema.ts', + out: './drizzle', + dialect: 'postgresql', + driver: 'pglite', + dbCredentials: { + url: './data/pglite-db', + }, +}) diff --git a/services/satori-bot/drizzle/0000_cooing_captain_flint.sql b/services/satori-bot/drizzle/0000_cooing_captain_flint.sql new file mode 100644 index 000000000..923178426 --- /dev/null +++ b/services/satori-bot/drizzle/0000_cooing_captain_flint.sql @@ -0,0 +1,31 @@ +CREATE TABLE "channels" ( + "id" text PRIMARY KEY NOT NULL, + "name" text NOT NULL, + "platform" text NOT NULL, + "self_id" text NOT NULL +); +--> statement-breakpoint +CREATE TABLE "event_queue" ( + "id" text PRIMARY KEY NOT NULL, + "event" json NOT NULL, + "status" text NOT NULL, + "created_at" bigint NOT NULL +); +--> statement-breakpoint +CREATE TABLE "messages" ( + "id" text PRIMARY KEY NOT NULL, + "channel_id" text NOT NULL, + "user_id" text NOT NULL, + "user_name" text NOT NULL, + "content" text NOT NULL, + "timestamp" bigint NOT NULL +); +--> statement-breakpoint +CREATE TABLE "unread_events" ( + "id" text PRIMARY KEY NOT NULL, + "channel_id" text NOT NULL, + "event" json NOT NULL, + "created_at" bigint NOT NULL +); +--> statement-breakpoint +CREATE INDEX "channel_timestamp_idx" ON "messages" USING btree ("channel_id","timestamp"); \ No newline at end of file diff --git a/services/satori-bot/drizzle/meta/0000_snapshot.json b/services/satori-bot/drizzle/meta/0000_snapshot.json new file mode 100644 index 000000000..1ecda6399 --- /dev/null +++ b/services/satori-bot/drizzle/meta/0000_snapshot.json @@ -0,0 +1,201 @@ +{ + "id": "e3501067-35f7-42db-9a70-4938f8d629a8", + "prevId": "00000000-0000-0000-0000-000000000000", + "version": "7", + "dialect": "postgresql", + "tables": { + "public.channels": { + "name": "channels", + "schema": "", + "columns": { + "id": { + "name": "id", + "type": "text", + "primaryKey": true, + "notNull": true + }, + "name": { + "name": "name", + "type": "text", + "primaryKey": false, + "notNull": true + }, + "platform": { + "name": "platform", + "type": "text", + "primaryKey": false, + "notNull": true + }, + "self_id": { + "name": "self_id", + "type": "text", + "primaryKey": false, + "notNull": true + } + }, + "indexes": {}, + "foreignKeys": {}, + "compositePrimaryKeys": {}, + "uniqueConstraints": {}, + "policies": {}, + "checkConstraints": {}, + "isRLSEnabled": false + }, + "public.event_queue": { + "name": "event_queue", + "schema": "", + "columns": { + "id": { + "name": "id", + "type": "text", + "primaryKey": true, + "notNull": true + }, + "event": { + "name": "event", + "type": "json", + "primaryKey": false, + "notNull": true + }, + "status": { + "name": "status", + "type": "text", + "primaryKey": false, + "notNull": true + }, + "created_at": { + "name": "created_at", + "type": "bigint", + "primaryKey": false, + "notNull": true + } + }, + "indexes": {}, + "foreignKeys": {}, + "compositePrimaryKeys": {}, + "uniqueConstraints": {}, + "policies": {}, + "checkConstraints": {}, + "isRLSEnabled": false + }, + "public.messages": { + "name": "messages", + "schema": "", + "columns": { + "id": { + "name": "id", + "type": "text", + "primaryKey": true, + "notNull": true + }, + "channel_id": { + "name": "channel_id", + "type": "text", + "primaryKey": false, + "notNull": true + }, + "user_id": { + "name": "user_id", + "type": "text", + "primaryKey": false, + "notNull": true + }, + "user_name": { + "name": "user_name", + "type": "text", + "primaryKey": false, + "notNull": true + }, + "content": { + "name": "content", + "type": "text", + "primaryKey": false, + "notNull": true + }, + "timestamp": { + "name": "timestamp", + "type": "bigint", + "primaryKey": false, + "notNull": true + } + }, + "indexes": { + "channel_timestamp_idx": { + "name": "channel_timestamp_idx", + "columns": [ + { + "expression": "channel_id", + "isExpression": false, + "asc": true, + "nulls": "last" + }, + { + "expression": "timestamp", + "isExpression": false, + "asc": true, + "nulls": "last" + } + ], + "isUnique": false, + "concurrently": false, + "method": "btree", + "with": {} + } + }, + "foreignKeys": {}, + "compositePrimaryKeys": {}, + "uniqueConstraints": {}, + "policies": {}, + "checkConstraints": {}, + "isRLSEnabled": false + }, + "public.unread_events": { + "name": "unread_events", + "schema": "", + "columns": { + "id": { + "name": "id", + "type": "text", + "primaryKey": true, + "notNull": true + }, + "channel_id": { + "name": "channel_id", + "type": "text", + "primaryKey": false, + "notNull": true + }, + "event": { + "name": "event", + "type": "json", + "primaryKey": false, + "notNull": true + }, + "created_at": { + "name": "created_at", + "type": "bigint", + "primaryKey": false, + "notNull": true + } + }, + "indexes": {}, + "foreignKeys": {}, + "compositePrimaryKeys": {}, + "uniqueConstraints": {}, + "policies": {}, + "checkConstraints": {}, + "isRLSEnabled": false + } + }, + "enums": {}, + "schemas": {}, + "sequences": {}, + "roles": {}, + "policies": {}, + "views": {}, + "_meta": { + "columns": {}, + "schemas": {}, + "tables": {} + } +} \ No newline at end of file diff --git a/services/satori-bot/drizzle/meta/_journal.json b/services/satori-bot/drizzle/meta/_journal.json new file mode 100644 index 000000000..31be1375b --- /dev/null +++ b/services/satori-bot/drizzle/meta/_journal.json @@ -0,0 +1,13 @@ +{ + "version": "7", + "dialect": "postgresql", + "entries": [ + { + "idx": 0, + "version": "7", + "when": 1772727599159, + "tag": "0000_cooing_captain_flint", + "breakpoints": true + } + ] +} \ No newline at end of file diff --git a/services/satori-bot/package.json b/services/satori-bot/package.json index a3339e0eb..4d44d47ee 100644 --- a/services/satori-bot/package.json +++ b/services/satori-bot/package.json @@ -17,9 +17,12 @@ "scripts": { "start": "tsx --env-file=.env --env-file-if-exists=.env.local src/index.ts", "dev": "tsx watch --env-file=.env --env-file-if-exists=.env.local src/index.ts", - "typecheck": "tsc --noEmit" + "typecheck": "tsc --noEmit", + "db:generate": "drizzle-kit generate", + "db:push": "drizzle-kit push" }, "dependencies": { + "@electric-sql/pglite": "catalog:", "@guiiai/logg": "catalog:", "@moeru/std": "catalog:", "@velin-dev/core": "^0.3.4", @@ -27,14 +30,16 @@ "@xsai/shared-chat": "catalog:", "@xsai/utils-chat": "catalog:", "best-effort-json-parser": "^1.2.1", + "drizzle-orm": "^0.45.1", "es-toolkit": "^1.44.0", - "lowdb": "^7.0.1", "nanoid": "^5.1.6", + "valibot": "^1.2.0", "ws": "^8.19.0" }, "devDependencies": { "@types/node": "^22.19.12", "@types/ws": "^8.18.1", + "drizzle-kit": "^0.31.9", "tsx": "^4.21.0", "typescript": "^5.9.3" } diff --git a/services/satori-bot/src/adapter/satori/api.ts b/services/satori-bot/src/adapter/satori/api.ts index a0b7a254f..1acb1e4ce 100644 --- a/services/satori-bot/src/adapter/satori/api.ts +++ b/services/satori-bot/src/adapter/satori/api.ts @@ -1,7 +1,10 @@ -import type { SatoriMessageCreateRequest, SatoriMessageCreateResponse } from './types' +import type { SatoriMessage, SatoriMessageCreateRequest, SatoriMessageCreateResponse } from './types' import { useLogg } from '@guiiai/logg' +import * as v from 'valibot' +import { SatoriMessageCreateResponseSchema, SatoriMessageSchema } from './schema' + const log = useLogg('SatoriAPI') export interface SatoriAPIConfig { @@ -34,7 +37,7 @@ export class SatoriAPI { private async request( endpoint: string, - body?: any, + body?: unknown, ): Promise { const url = `${this.config.baseUrl}${endpoint}` @@ -43,6 +46,7 @@ export class SatoriAPI { method: 'POST', headers: this.getHeaders(), body: body ? JSON.stringify(body) : undefined, + signal: AbortSignal.timeout(10000), }) if (!response.ok) { @@ -68,14 +72,16 @@ export class SatoriAPI { } log.log(`Sending message to channel ${channelId}: ${content}`) - return await this.request('/message.create', body) + const response = await this.request('/message.create', body) + return v.parse(v.array(SatoriMessageCreateResponseSchema), response) } - async getMessage(channelId: string, messageId: string): Promise { - return await this.request('/message.get', { + async getMessage(channelId: string, messageId: string): Promise { + const response = await this.request('/message.get', { channel_id: channelId, message_id: messageId, }) + return v.parse(SatoriMessageSchema, response) } async deleteMessage(channelId: string, messageId: string): Promise { diff --git a/services/satori-bot/src/adapter/satori/client.ts b/services/satori-bot/src/adapter/satori/client.ts index 6dc7f2169..5ef5787fc 100644 --- a/services/satori-bot/src/adapter/satori/client.ts +++ b/services/satori-bot/src/adapter/satori/client.ts @@ -7,6 +7,9 @@ import { useLogg } from '@guiiai/logg' import { SatoriAPI } from './api' import { SatoriOpcode } from './types' +import * as v from 'valibot' +import { SatoriEventSchema, SatoriReadyBodySchema, SatoriSignalSchema } from './schema' + const log = useLogg('SatoriClient') export interface SatoriClientConfig { @@ -34,6 +37,12 @@ export class SatoriClient { } async connect(): Promise { + if (this.ws) { + this.ws.removeAllListeners() + this.ws.close() + this.ws = undefined + } + if (this.connected) { log.warn('Already connected to Satori server') return @@ -107,11 +116,12 @@ export class SatoriClient { private async handleMessage(data: WebSocket.Data): Promise { try { - const signal = JSON.parse(data.toString()) as SatoriSignal + const rawData = JSON.parse(data.toString()) + const signal = v.parse(SatoriSignalSchema, rawData) switch (signal.op) { case SatoriOpcode.READY: { - const readyBody = signal.body as SatoriReadyBody + const readyBody = v.parse(SatoriReadyBodySchema, signal.body) log.log('Received READY signal') // Initialize API clients for each login @@ -124,7 +134,7 @@ export class SatoriClient { } case SatoriOpcode.EVENT: { - const event = signal.body as SatoriEvent + const event = v.parse(SatoriEventSchema, signal.body) this.lastSequenceNumber = event.id await this.handleEvent(event) break @@ -145,7 +155,15 @@ export class SatoriClient { } } catch (error) { - log.withError(error as Error).error('Failed to handle message') + if (v.isValiError(error)) { + log.error('Satori protocol validation failed:') + for (const issue of error.issues) { + log.error(` - ${issue.path?.map(p => p.key).join('.')}: ${issue.message}`) + } + } + else { + log.withError(error as Error).error('Failed to handle message') + } } } @@ -176,6 +194,11 @@ export class SatoriClient { this.connected = false this.stopHeartbeat() + if (this.ws) { + this.ws.removeAllListeners() + this.ws = undefined + } + if (this.shouldReconnect) { log.log('Attempting to reconnect in 5 seconds...') this.reconnectTimeout = setTimeout(() => { @@ -272,6 +295,7 @@ export class SatoriClient { } if (this.ws) { + this.ws.removeAllListeners() this.ws.close() this.ws = undefined } diff --git a/services/satori-bot/src/adapter/satori/schema.ts b/services/satori-bot/src/adapter/satori/schema.ts new file mode 100644 index 000000000..d64e17af1 --- /dev/null +++ b/services/satori-bot/src/adapter/satori/schema.ts @@ -0,0 +1,102 @@ +import * as v from 'valibot' + +export const SatoriUserSchema = v.object({ + id: v.string(), + name: v.optional(v.string()), + nick: v.optional(v.string()), + avatar: v.optional(v.string()), + is_bot: v.optional(v.boolean()), +}) + +export const SatoriChannelSchema = v.object({ + id: v.string(), + type: v.number(), + name: v.optional(v.string()), + parent_id: v.optional(v.string()), +}) + +export const SatoriGuildSchema = v.object({ + id: v.string(), + name: v.optional(v.string()), + avatar: v.optional(v.string()), +}) + +export const SatoriGuildMemberSchema = v.object({ + user: v.optional(SatoriUserSchema), + nick: v.optional(v.string()), + avatar: v.optional(v.string()), + joined_at: v.optional(v.number()), +}) + +export const SatoriMessageSchema = v.object({ + id: v.string(), + content: v.string(), + platform: v.optional(v.string()), + channel: v.optional(SatoriChannelSchema), + guild: v.optional(SatoriGuildSchema), + member: v.optional(SatoriGuildMemberSchema), + user: v.optional(SatoriUserSchema), + created_at: v.optional(v.number()), + updated_at: v.optional(v.number()), +}) + +export const SatoriLoginSchema = v.object({ + user: v.optional(SatoriUserSchema), + self_id: v.optional(v.string()), + platform: v.optional(v.string()), + status: v.number(), + features: v.optional(v.array(v.string())), + proxy_urls: v.optional(v.array(v.string())), +}) + +export const SatoriArgvSchema = v.object({ + name: v.string(), + arguments: v.array(v.unknown()), + options: v.record(v.string(), v.unknown()), +}) + +export const SatoriEventSchema = v.object({ + id: v.number(), + type: v.string(), + platform: v.string(), + self_id: v.string(), + timestamp: v.number(), + argv: v.optional(SatoriArgvSchema), + button: v.optional(v.object({ id: v.string() })), + channel: v.optional(SatoriChannelSchema), + guild: v.optional(SatoriGuildSchema), + login: v.optional(SatoriLoginSchema), + member: v.optional(SatoriGuildMemberSchema), + message: v.optional(SatoriMessageSchema), + operator: v.optional(SatoriUserSchema), + role: v.optional(v.object({ id: v.string(), name: v.optional(v.string()) })), + user: v.optional(SatoriUserSchema), + _type: v.optional(v.string()), + _data: v.optional(v.record(v.string(), v.unknown())), +}) + +export const SatoriMessageCreateResponseSchema = v.object({ + id: v.string(), + content: v.optional(v.string()), + channel: v.optional(SatoriChannelSchema), + guild: v.optional(SatoriGuildSchema), + member: v.optional(SatoriGuildMemberSchema), + user: v.optional(SatoriUserSchema), + created_at: v.optional(v.number()), + updated_at: v.optional(v.number()), +}) + +export const SatoriReadyBodySchema = v.object({ + logins: v.array(SatoriLoginSchema), + proxy_urls: v.optional(v.array(v.string())), +}) + +export const SatoriSignalSchema = v.object({ + op: v.number(), + body: v.optional(v.unknown()), +}) + +export const SatoriListSchema = >(itemSchema: T) => v.object({ + data: v.array(itemSchema), + next: v.optional(v.string()), +}) diff --git a/services/satori-bot/src/adapter/satori/types.ts b/services/satori-bot/src/adapter/satori/types.ts index 1d519c0d3..3f450fa24 100644 --- a/services/satori-bot/src/adapter/satori/types.ts +++ b/services/satori-bot/src/adapter/satori/types.ts @@ -14,7 +14,7 @@ export enum SatoriOpcode { } // WebSocket Signal Structure -export interface SatoriSignal { +export interface SatoriSignal { op: SatoriOpcode body?: T } @@ -28,12 +28,12 @@ export interface SatoriIdentifyBody { // READY signal body export interface SatoriReadyBody { logins: SatoriLogin[] - proxy_urls: string[] + proxy_urls?: string[] } // META signal body export interface SatoriMetaBody { - proxy_urls: string[] + proxy_urls?: string[] } // User resource @@ -100,8 +100,8 @@ export interface SatoriLogin { // Interaction Argv export interface SatoriArgv { name: string - arguments: any[] - options: Record + arguments: unknown[] + options: Record } // Interaction Button @@ -127,7 +127,7 @@ export interface SatoriEvent { role?: SatoriGuildRole user?: SatoriUser _type?: string - _data?: Record + _data?: Record } // API Request/Response types @@ -138,7 +138,7 @@ export interface SatoriMessageCreateRequest { export interface SatoriMessageCreateResponse { id: string - content: string + content?: string channel?: SatoriChannel guild?: SatoriGuild member?: SatoriGuildMember diff --git a/services/satori-bot/src/capabilities/actions/read-messages.ts b/services/satori-bot/src/capabilities/actions/read-messages.ts index 03d3aca2a..867fc2353 100644 --- a/services/satori-bot/src/capabilities/actions/read-messages.ts +++ b/services/satori-bot/src/capabilities/actions/read-messages.ts @@ -1,13 +1,21 @@ import type { ActionHandler, ActionResult } from '../definition' import { useLogg } from '@guiiai/logg' +import { deleteUnreadEventsByIds } from '../../lib/db' export const readMessagesAction: ActionHandler = { name: 'read_unread_messages', description: 'Read unread messages from a specific channel', execute: async (botContext, chatCtx, args): Promise => { + if (args.action !== 'read_unread_messages') { + return { + success: false, + shouldContinue: true, + result: 'System Error: Action mismatch for read_unread_messages.', + } + } const logger = useLogg('readMessagesAction').useGlobalConfig() - const channelId = args.channelId + const { channelId } = args if (!channelId) { return { @@ -28,13 +36,25 @@ export const readMessagesAction: ActionHandler = { } } - const formattedMessages = unreadEventsForThisChannel.map((event) => { + // Capture the IDs of the events we are about to "read" + const readEventIds = unreadEventsForThisChannel.map(item => item.id) + + const formattedMessages = unreadEventsForThisChannel.map((item) => { + const { event } = item const userName = event.user?.name || event.user?.id || 'Unknown' const content = event.message?.content || '[No content]' return `[${userName}]: ${content}` }).join('\n') - delete botContext.unreadEvents[channelId] + // Only remove the events we just read, preserving any that might have arrived during processing + botContext.unreadEvents[channelId] = (botContext.unreadEvents[channelId] || []) + .filter(item => !readEventIds.includes(item.id)) + + if (botContext.unreadEvents[channelId].length === 0) { + delete botContext.unreadEvents[channelId] + } + + await deleteUnreadEventsByIds(channelId, readEventIds) logger.log(`Read ${unreadEventsForThisChannel.length} unread events from channel ${channelId}`) diff --git a/services/satori-bot/src/capabilities/actions/send-message.ts b/services/satori-bot/src/capabilities/actions/send-message.ts index 29eb7fdf9..525765996 100644 --- a/services/satori-bot/src/capabilities/actions/send-message.ts +++ b/services/satori-bot/src/capabilities/actions/send-message.ts @@ -9,6 +9,13 @@ export function createSendMessageAction(client: SatoriClient): ActionHandler { return { name: 'send_message', execute: async (ctx, chatCtx, args) => { + if (args.action !== 'send_message') { + return { + success: false, + shouldContinue: true, + result: 'System Error: Action mismatch for send_message.', + } + } const logger = useLogg('Action:send_message').useGlobalConfig() const { channelId, content } = args @@ -28,13 +35,7 @@ export function createSendMessageAction(client: SatoriClient): ActionHandler { await client.sendMessage(chatCtx.platform, chatCtx.selfId, channelId, content) // Logic 3: Persistence - await recordMessage(channelId, 'bot', 'AIRI', content) - - // Logic 4: Memory State Update - chatCtx.messages.push({ - role: 'assistant', - content, - }) + await recordMessage(channelId, chatCtx.selfId, 'AIRI', content) return { success: true, diff --git a/services/satori-bot/src/capabilities/actions/system.ts b/services/satori-bot/src/capabilities/actions/system.ts index 32750eca4..4fe01bbef 100644 --- a/services/satori-bot/src/capabilities/actions/system.ts +++ b/services/satori-bot/src/capabilities/actions/system.ts @@ -19,7 +19,6 @@ export const continueAction: ActionHandler = { export const breakAction: ActionHandler = { name: 'break', execute: async (_ctx, chatCtx): Promise => { - chatCtx.messages = [] chatCtx.actions = [] return { success: true, @@ -33,6 +32,13 @@ export const breakAction: ActionHandler = { export const sleepAction: ActionHandler = { name: 'sleep', execute: async (_ctx, _chatCtx, args): Promise => { + if (args.action !== 'sleep') { + return { + success: false, + shouldContinue: true, + result: 'System Error: Action mismatch for sleep.', + } + } const duration = args.duration || SLEEP_DURATION_MS await new Promise(resolve => setTimeout(resolve, duration)) return { diff --git a/services/satori-bot/src/capabilities/definition.ts b/services/satori-bot/src/capabilities/definition.ts index 264e5237a..4dc6a1b4a 100644 --- a/services/satori-bot/src/capabilities/definition.ts +++ b/services/satori-bot/src/capabilities/definition.ts @@ -1,9 +1,9 @@ -import type { BotContext, ChatContext } from '../core/types' +import type { Action, BotContext, ChatContext } from '../core/types' export interface ActionResult { success: boolean shouldContinue: boolean - result: any + result: unknown } export interface ActionHandler { @@ -12,7 +12,7 @@ export interface ActionHandler { execute: ( ctx: BotContext, chatCtx: ChatContext, - args: any, + args: Action, abortSignal?: AbortSignal, ) => Promise } diff --git a/services/satori-bot/src/config.ts b/services/satori-bot/src/config.ts new file mode 100644 index 000000000..1de87841f --- /dev/null +++ b/services/satori-bot/src/config.ts @@ -0,0 +1,63 @@ +import { env } from 'node:process' +import * as v from 'valibot' + +const ConfigSchema = v.object({ + satori: v.object({ + wsUrl: v.string(), + token: v.optional(v.string()), + apiBaseUrl: v.optional(v.string()), + }), + llm: v.object({ + apiKey: v.string(), + baseUrl: v.string(), + model: v.string(), + ollamaDisableThink: v.optional(v.boolean(), false), + }), + db: v.object({ + path: v.optional(v.string(), '../../data/pglite-db'), + }), +}) + +export type Config = v.InferOutput + +function parseBoolean(value: string | undefined): boolean | undefined { + if (value === undefined) return undefined + return value.toLowerCase() === 'true' || value === '1' +} + +export function loadConfig(): Config { + const rawConfig = { + satori: { + wsUrl: env.SATORI_WS_URL || 'ws://localhost:5140/satori/v1/events', + token: env.SATORI_TOKEN, + apiBaseUrl: env.SATORI_API_BASE_URL, + }, + llm: { + apiKey: env.LLM_API_KEY, + baseUrl: env.LLM_API_BASE_URL, + model: env.LLM_MODEL, + ollamaDisableThink: parseBoolean(env.LLM_OLLAMA_DISABLE_THINK), + }, + db: { + path: env.DB_PATH, + }, + } + + try { + return v.parse(ConfigSchema, rawConfig) + } + catch (error) { + if (v.isValiError(error)) { + console.error('❌ Configuration validation failed:') + for (const issue of error.issues) { + console.error(` - ${issue.path?.map(p => p.key).join('.')}: ${issue.message}`) + } + } + else { + console.error('❌ Failed to load configuration:', error) + } + process.exit(1) + } +} + +export const config = loadConfig() diff --git a/services/satori-bot/src/core/constants.ts b/services/satori-bot/src/core/constants.ts index 076fd9c00..1e521cbd9 100644 --- a/services/satori-bot/src/core/constants.ts +++ b/services/satori-bot/src/core/constants.ts @@ -7,9 +7,9 @@ export const LOOP_CONTINUE_DELAY_MS = 2500 export const PERIODIC_LOOP_INTERVAL_MS = 60 * 1000 export const SLEEP_DURATION_MS = 30 * 1000 +export const MAX_LOOP_ITERATIONS = 5 // Context size limits -export const MAX_MESSAGES_IN_CONTEXT = 20 export const MAX_ACTIONS_IN_CONTEXT = 50 export const MAX_UNREAD_EVENTS = 100 @@ -17,5 +17,4 @@ export const MAX_UNREAD_EVENTS = 100 export const MAX_RECENT_INTERACTED_CHANNELS = 5 // Context trimming - how many items to keep when trimming -export const MESSAGES_KEEP_ON_TRIM = 5 export const ACTIONS_KEEP_ON_TRIM = 20 diff --git a/services/satori-bot/src/core/dispatcher.ts b/services/satori-bot/src/core/dispatcher.ts index 425783b33..8072be3e7 100644 --- a/services/satori-bot/src/core/dispatcher.ts +++ b/services/satori-bot/src/core/dispatcher.ts @@ -1,41 +1,47 @@ import type { ActionResult } from '../capabilities/definition' import type { BotContext, ChatContext } from './types' +import * as v from 'valibot' + import { globalRegistry } from '../capabilities/registry' +import { ActionSchema } from './types' export async function dispatchAction( ctx: BotContext, chatCtx: ChatContext, - actionPayload: any, + actionPayload: unknown, abortController: AbortController, ): Promise { const log = ctx.logger.useGlobalConfig() - if (!actionPayload || !actionPayload.action) { + const parseResult = v.safeParse(ActionSchema, actionPayload) + + if (!parseResult.success) { return { success: false, shouldContinue: true, - result: 'System Error: No valid action name provided in JSON.', + result: `System Error: Invalid action payload: ${parseResult.issues.map(i => i.message).join(', ')}`, } } - const handler = globalRegistry.get(actionPayload.action) + const validatedAction = parseResult.output + const handler = globalRegistry.get(validatedAction.action) if (!handler) { return { success: false, shouldContinue: true, - result: `System Error: Action "${actionPayload.action}" is not implemented.`, + result: `System Error: Action "${validatedAction.action}" is not implemented.`, } } try { - log.withField('action', actionPayload.action).debug('Executing action') + log.withField('action', validatedAction.action).debug('Executing action') - const result = await handler.execute(ctx, chatCtx, actionPayload, abortController.signal) + const result = await handler.execute(ctx, chatCtx, validatedAction, abortController.signal) chatCtx.actions.push({ - action: actionPayload, + action: validatedAction, result: result.result, }) diff --git a/services/satori-bot/src/core/loop/queue.ts b/services/satori-bot/src/core/loop/queue.ts index 6e4235ddf..e7130bc56 100644 --- a/services/satori-bot/src/core/loop/queue.ts +++ b/services/satori-bot/src/core/loop/queue.ts @@ -5,6 +5,7 @@ import type { SatoriEvent, SatoriReadyBody } from '../../adapter/satori/types' import type { BotContext } from '../types' import { onMessageArrival } from './scheduler' +import { pushToEventQueue } from '../../lib/db' /** * Set up the ready event handler @@ -51,9 +52,15 @@ export function setupMessageEventHandler( } // Add to message queue - botContext.eventQueue.push({ + const queueItem = { event, - status: 'ready', + status: 'ready' as const, + } + const id = await pushToEventQueue(queueItem) + + botContext.eventQueue.push({ + ...queueItem, + id, }) // Process message queue diff --git a/services/satori-bot/src/core/loop/scheduler.ts b/services/satori-bot/src/core/loop/scheduler.ts index 4f9e360a8..df7ed1fb1 100644 --- a/services/satori-bot/src/core/loop/scheduler.ts +++ b/services/satori-bot/src/core/loop/scheduler.ts @@ -1,21 +1,23 @@ +import type { Message as LLMMessage } from '@xsai/shared-chat' + import type { SatoriClient } from '../../adapter/satori/client' import type { SatoriEvent } from '../../adapter/satori/types' import type { BotContext, ChatContext } from '../types' -import { recordChannel, recordMessage } from '../../lib/db' +import { getRecentMessages, recordChannel, recordMessage, removeFromEventQueue, saveEventQueue, pushToUnreadEvents } from '../../lib/db' import { ACTIONS_KEEP_ON_TRIM, LOOP_CONTINUE_DELAY_MS, MAX_ACTIONS_IN_CONTEXT, - MAX_MESSAGES_IN_CONTEXT, + MAX_LOOP_ITERATIONS, MAX_RECENT_INTERACTED_CHANNELS, MAX_UNREAD_EVENTS, - MESSAGES_KEEP_ON_TRIM, PERIODIC_LOOP_INTERVAL_MS, } from '../constants' import { dispatchAction } from '../dispatcher' import { imagineAnAction } from '../planner/llm-client' import { ensureChatContext } from '../session/context' +import { trimActions } from '../utils' /** * Handle a single loop step @@ -27,80 +29,89 @@ export async function handleLoopStep( chatCtx: ChatContext, incomingEvents?: SatoriEvent, ): Promise { - ctx.currentProcessingStartTime = Date.now() + let shouldContinue = true + let currentIncoming = incomingEvents + let iterationCount = 0 - if (chatCtx?.currentAbortController) { - chatCtx.currentAbortController.abort() - } - - const currentController = new AbortController() - if (chatCtx) { - chatCtx.currentAbortController = currentController - - // Track message processing state - if (chatCtx.channelId && !ctx.lastInteractedChannelIds.includes(chatCtx.channelId)) { - ctx.lastInteractedChannelIds.push(chatCtx.channelId) + while (shouldContinue) { + if (iterationCount >= MAX_LOOP_ITERATIONS) { + ctx.logger + .withField('channelId', chatCtx?.channelId) + .withField('iterationCount', iterationCount) + .log('Reached maximum loop iterations, breaking to prevent infinite loop') + break } - if (ctx.lastInteractedChannelIds.length > MAX_RECENT_INTERACTED_CHANNELS) { - ctx.lastInteractedChannelIds = ctx.lastInteractedChannelIds.slice(-MAX_RECENT_INTERACTED_CHANNELS) + iterationCount++ + + ctx.currentProcessingStartTime = Date.now() + + if (chatCtx?.currentAbortController) { + chatCtx.currentAbortController.abort() } - // Manage context size - if (chatCtx.messages == null) { - chatCtx.messages = [] - } - if (chatCtx.messages.length > MAX_MESSAGES_IN_CONTEXT) { - const length = chatCtx.messages.length - chatCtx.messages = chatCtx.messages.slice(-MESSAGES_KEEP_ON_TRIM) - chatCtx.messages.push({ - role: 'user', - content: `AIRI System: Approaching to system context limit, reducing... memory..., reduced from ${length} to ${chatCtx.messages.length}, history may be lost.`, - }) + const currentController = new AbortController() + if (chatCtx) { + chatCtx.currentAbortController = currentController + + // Track message processing state + if (chatCtx.channelId && !ctx.lastInteractedChannelIds.includes(chatCtx.channelId)) { + ctx.lastInteractedChannelIds.push(chatCtx.channelId) + } + if (ctx.lastInteractedChannelIds.length > MAX_RECENT_INTERACTED_CHANNELS) { + ctx.lastInteractedChannelIds = ctx.lastInteractedChannelIds.slice(-MAX_RECENT_INTERACTED_CHANNELS) + } + + // Manage action context size + if (chatCtx.actions == null) { + chatCtx.actions = [] + } + chatCtx.actions = trimActions(chatCtx.actions, MAX_ACTIONS_IN_CONTEXT, ACTIONS_KEEP_ON_TRIM) } - if (chatCtx.actions == null) { - chatCtx.actions = [] - } - if (chatCtx.actions.length > MAX_ACTIONS_IN_CONTEXT) { - const length = chatCtx.actions.length - chatCtx.actions = chatCtx.actions.slice(-ACTIONS_KEEP_ON_TRIM) - chatCtx.messages.push({ - role: 'user', - content: `AIRI System: Approaching to system context limit, reducing... memory..., reduced from ${length} to ${chatCtx.actions.length}, history of actions may be lost.`, - }) - } - } + try { + // Dynamic history injection: Fetch last 10 messages from DB + const dbMessages = await getRecentMessages(chatCtx.channelId, 10) + const llmMessages: LLMMessage[] = dbMessages.map(m => ({ + role: m.userId === chatCtx.selfId ? 'assistant' : 'user', + content: m.content, + })) - try { - const actionPayload = await imagineAnAction( - currentController, - chatCtx?.messages || [], - chatCtx?.actions || [], - { - unreadEvents: ctx.unreadEvents, - incomingEvents: incomingEvents ? [incomingEvents] : [], - }, - ) + const actionPayload = await imagineAnAction( + currentController, + llmMessages, + chatCtx?.actions || [], + { + unreadEvents: ctx.unreadEvents, + incomingEvents: currentIncoming ? [currentIncoming] : [], + }, + ) - const result = await dispatchAction(ctx, chatCtx, actionPayload, currentController) - if (result.shouldContinue) { - await new Promise(r => setTimeout(r, LOOP_CONTINUE_DELAY_MS)) - // Recursively call next step and await it - await handleLoopStep(ctx, satoriClient, chatCtx) - } - } - catch (err) { - if ((err as Error).name === 'AbortError') { - ctx.logger.log('Operation was aborted due to interruption') - return - } + if (!actionPayload) { + shouldContinue = false + break + } - ctx.logger.withError(err as Error).log('Error occurred') - } - finally { - if (chatCtx && chatCtx.currentAbortController === currentController) { - chatCtx.currentAbortController = undefined - ctx.currentProcessingStartTime = undefined + const result = await dispatchAction(ctx, chatCtx, actionPayload, currentController) + shouldContinue = result.shouldContinue + + if (shouldContinue) { + await new Promise(r => setTimeout(r, LOOP_CONTINUE_DELAY_MS)) + currentIncoming = undefined // Only the first step uses the initial incoming event + } + } + catch (err) { + if ((err as Error).name === 'AbortError') { + ctx.logger.log('Operation was aborted due to interruption') + } else { + ctx.logger.withError(err as Error).log('Error occurred') + } + shouldContinue = false + } + finally { + if (chatCtx && chatCtx.currentAbortController === currentController) { + chatCtx.currentAbortController = undefined + ctx.currentProcessingStartTime = undefined + } } } } @@ -115,7 +126,7 @@ export async function loopIterationForChannel( chatContext: ChatContext, incomingEvent: SatoriEvent, ) { - // Directly await the recursive process + // Directly await the loop process await handleLoopStep(bot, satoriClient, chatContext, incomingEvent) } @@ -136,15 +147,32 @@ async function loopIterationPeriodicForExistingChannels(ctx: BotContext, satoriC ctx.logger.withField('channelCount', channelsWithUnread.length).log('Processing channels with unread events') - // Process channels sequentially to avoid overwhelming the LLM API + // Process channels in parallel but with their own locks for (const channelId of channelsWithUnread) { try { const chatCtx = await ensureChatContext(ctx, channelId) - await handleLoopStep(ctx, satoriClient, chatCtx) + + if (chatCtx.isProcessing) { + ctx.logger.withField('channelId', channelId).debug('Channel is already processing, skipping periodic loop for this channel') + continue + } + + // Start processing for this channel in background + chatCtx.isProcessing = true + ;(async () => { + try { + await handleLoopStep(ctx, satoriClient, chatCtx) + } + catch (err) { + ctx.logger.withError(err as Error).withField('channelId', channelId).log('Error processing channel in periodic loop') + } + finally { + chatCtx.isProcessing = false + } + })() } catch (err) { - ctx.logger.withError(err as Error).withField('channelId', channelId).log('Error processing channel in periodic loop') - // Continue to next channel instead of breaking the entire loop + ctx.logger.withError(err as Error).withField('channelId', channelId).log('Error ensuring chat context in periodic loop') continue } } @@ -181,6 +209,8 @@ export function startPeriodicLoop(botCtx: BotContext, satoriClient: SatoriClient loopPeriodic(botCtx, satoriClient) } +let isQueueConsumerRunning = false + /** * Handle message arrival event * Processes messages from the queue, records them, and triggers bot responses @@ -190,10 +220,10 @@ export async function onMessageArrival( botContext: BotContext, satoriClient: SatoriClient, ) { - if (botContext.processing) { + if (isQueueConsumerRunning) { return } - botContext.processing = true + isQueueConsumerRunning = true const log = botContext.logger @@ -209,6 +239,23 @@ export async function onMessageArrival( const sourceUserId = currMsg.event.user?.id || currMsg.event.member?.user?.id const sourceUserName = currMsg.event.user?.name || currMsg.event.member?.user?.name || 'unknown' + // Protocol-side persistence: Record channel and message at the very beginning + await recordChannel( + channelId, + currMsg.event.channel?.name || channelId, + platform, + selfId, + ) + + if (currMsg.event.user && currMsg.event.message?.content) { + await recordMessage( + channelId, + sourceUserId, + sourceUserName, + currMsg.event.message.content, + ) + } + const chatCtx = await ensureChatContext(botContext, channelId) if (!chatCtx.platform || chatCtx.platform === '') { @@ -218,24 +265,6 @@ export async function onMessageArrival( chatCtx.selfId = selfId } - // Record channel - await recordChannel( - chatCtx.channelId, - currMsg.event.channel?.name || chatCtx.channelId, - chatCtx.platform, - chatCtx.selfId, - ) - - // Record message - if (currMsg.event.user && currMsg.event.message?.content) { - await recordMessage( - chatCtx.channelId, - currMsg.event.user.id, - currMsg.event.user.name || currMsg.event.user.id, - currMsg.event.message.content, - ) - } - // Skip bot's own messages - don't add them to unreadEvents if (sourceUserId === chatCtx.selfId) { botContext.logger @@ -247,6 +276,12 @@ export async function onMessageArrival( }) .debug('[DEBUG] Skipping bot\'s own event in unreadEvents - filtered out') botContext.eventQueue.shift() + if (currMsg.id) { + await removeFromEventQueue(currMsg.id) + } + else { + await saveEventQueue(botContext.eventQueue) + } continue } @@ -261,7 +296,8 @@ export async function onMessageArrival( unreadEventsForThisChannel = [] } - unreadEventsForThisChannel.push(currMsg.event) + const unreadEventId = await pushToUnreadEvents(chatCtx.channelId, currMsg.event) + unreadEventsForThisChannel.push({ id: unreadEventId, event: currMsg.event }) if (unreadEventsForThisChannel.length > MAX_UNREAD_EVENTS) { unreadEventsForThisChannel = unreadEventsForThisChannel.slice(-MAX_UNREAD_EVENTS) @@ -269,17 +305,42 @@ export async function onMessageArrival( botContext.unreadEvents[chatCtx.channelId] = unreadEventsForThisChannel + // Consume the event from queue immediately + botContext.eventQueue.shift() + if (currMsg.id) { + await removeFromEventQueue(currMsg.id) + } + else { + await saveEventQueue(botContext.eventQueue) + } + + if (chatCtx.isProcessing) { + botContext.logger.withField('channelId', chatCtx.channelId).log('Channel is already processing, added to unreadEvents only') + continue + } + botContext.logger.withField('channelId', chatCtx.channelId).log('event queue processed, triggering immediate reaction') - // Trigger immediate processing with the correct chatCtx for this message - await loopIterationForChannel(botContext, satoriClient, chatCtx, currMsg.event) - botContext.eventQueue.shift() + // Trigger immediate processing without awaiting to allow other channels to proceed + chatCtx.isProcessing = true + // We use a self-invoking async function to handle the processing and lock release + ;(async () => { + try { + await loopIterationForChannel(botContext, satoriClient, chatCtx, currMsg.event) + } + catch (err) { + botContext.logger.withError(err as Error).withField('channelId', chatCtx.channelId).log('Error in channel-specific loop') + } + finally { + chatCtx.isProcessing = false + } + })() } } catch (err) { - botContext.logger.withError(err as Error).log('Error occurred') + botContext.logger.withError(err as Error).log('Error occurred in onMessageArrival') } finally { - botContext.processing = false + isQueueConsumerRunning = false } } diff --git a/services/satori-bot/src/core/planner/llm-client.ts b/services/satori-bot/src/core/planner/llm-client.ts index 2eada06d2..06f8a21d1 100644 --- a/services/satori-bot/src/core/planner/llm-client.ts +++ b/services/satori-bot/src/core/planner/llm-client.ts @@ -2,23 +2,25 @@ import type { GenerateTextOptions } from '@xsai/generate-text' import type { Message as LLMMessage } from '@xsai/shared-chat' import type { SatoriEvent } from '../../adapter/satori/types' -import type { Action } from '../types' - -import { env } from 'node:process' +import type { Action, StoredUnreadEvent } from '../types' import { useLogg } from '@guiiai/logg' import { generateText } from '@xsai/generate-text' import { message } from '@xsai/utils-chat' import { parse } from 'best-effort-json-parser' -import { personality, systemPrompt } from './prompts' +import * as v from 'valibot' + +import { config } from '../../config' +import { ActionSchema } from '../types' +import { personality, systemPrompt } from './prompts/index' export async function imagineAnAction( currentAbortController: AbortController | undefined, messages: LLMMessage[], actions: { action: Action, result: unknown }[], globalStates: { - unreadEvents: Record + unreadEvents: Record incomingEvents?: SatoriEvent[] }, ): Promise { @@ -54,26 +56,15 @@ export async function imagineAnAction( ) try { - // Validate API configuration - if (!env.LLM_API_KEY) { - throw new Error('LLM_API_KEY is not configured. Please set it in your .env.local file.') - } - if (!env.LLM_API_BASE_URL) { - throw new Error('LLM_API_BASE_URL is not configured. Please set it in your .env.local file.') - } - if (!env.LLM_MODEL) { - throw new Error('LLM_MODEL is not configured. Please set it in your .env.local file.') - } - const req = { - apiKey: env.LLM_API_KEY, - baseURL: env.LLM_API_BASE_URL, - model: env.LLM_MODEL, + apiKey: config.llm.apiKey, + baseURL: config.llm.baseUrl, + model: config.llm.model, messages: requestMessages, abortSignal: currentAbortController?.signal, } satisfies GenerateTextOptions - if (env.LLM_OLLAMA_DISABLE_THINK) { + if (config.llm.ollamaDisableThink) { (req as Record).think = false } @@ -98,16 +89,23 @@ export async function imagineAnAction( .replace(/\s*```\s*$/m, '') .trim() - const parsed = parse(responseText) as any + const parsed = parse(responseText) - // 如果 LLM 返回的 JSON 有 parameters 包装层，需要展开 + // Validate using valibot + // Handle the case where LLM might wrap parameters + let actionToValidate = parsed if (parsed.parameters && typeof parsed.parameters === 'object') { const { parameters, ...rest } = parsed - const action = { ...rest, ...parameters } as Action - return action + + if (parameters.channelId !== undefined) { + parameters.channelId = String(parameters.channelId) + } + + actionToValidate = { ...rest, ...parameters } } - return parsed as Action + const validated = v.parse(ActionSchema, actionToValidate) + return validated } catch (err) { const error = err as Error @@ -115,8 +113,8 @@ export async function imagineAnAction( // Check for API key errors if (error.message?.includes('API Key') || error.message?.includes('API key')) { logger.error('❌ LLM API Key Error: Please check your .env.local file and ensure LLM_API_KEY is set correctly.') - logger.error(` Current LLM_API_BASE_URL: ${env.LLM_API_BASE_URL}`) - logger.error(` Current LLM_MODEL: ${env.LLM_MODEL}`) + logger.error(` Current LLM_API_BASE_URL: ${config.llm.baseUrl}`) + logger.error(` Current LLM_MODEL: ${config.llm.model}`) } else if (error.message?.includes('LLM_')) { // Configuration error diff --git a/services/satori-bot/src/core/session/context.ts b/services/satori-bot/src/core/session/context.ts index 0145d7464..30532baad 100644 --- a/services/satori-bot/src/core/session/context.ts +++ b/services/satori-bot/src/core/session/context.ts @@ -2,19 +2,23 @@ import type { Logg } from '@guiiai/logg' import type { BotContext, ChatContext } from '../types' -import { listChannels } from '../../lib/db' +import { listChannels, loadEventQueue, loadUnreadEvents } from '../../lib/db' /** * Create a new bot context * Initializes all required data structures for the bot */ -export function createBotContext(logger: Logg): BotContext { +export async function createBotContext(logger: Logg): Promise { + const [eventQueue, unreadEvents] = await Promise.all([ + loadEventQueue(), + loadUnreadEvents(), + ]) + const botSelf: BotContext = { - eventQueue: [], - unreadEvents: {}, + eventQueue, + unreadEvents, processedIds: new Set(), logger, - processing: false, lastInteractedChannelIds: [], chats: new Map(), } @@ -47,9 +51,9 @@ export async function ensureChatContext(botCtx: BotContext, channelId: string): channelId, platform: channelInfo?.platform || '', selfId: channelInfo?.selfId || '', + isProcessing: false, currentTask: undefined, currentAbortController: undefined, - messages: [], actions: [], } diff --git a/services/satori-bot/src/core/types.ts b/services/satori-bot/src/core/types.ts index 35d36e1dd..30097a92b 100644 --- a/services/satori-bot/src/core/types.ts +++ b/services/satori-bot/src/core/types.ts @@ -1,8 +1,49 @@ import type { Logg } from '@guiiai/logg' -import type { Message as LLMMessage } from '@xsai/shared-chat' import type { SatoriEvent } from '../adapter/satori/types' +import * as v from 'valibot' + +// Action schemas +export const ContinueActionSchema = v.object({ + action: v.literal('continue'), +}) + +export const BreakActionSchema = v.object({ + action: v.literal('break'), +}) + +export const SleepActionSchema = v.object({ + action: v.literal('sleep'), + duration: v.optional(v.number()), +}) + +export const ListChannelsActionSchema = v.object({ + action: v.literal('list_channels'), +}) + +export const SendMessageActionSchema = v.object({ + action: v.literal('send_message'), + content: v.string(), + channelId: v.string(), +}) + +export const ReadUnreadMessagesActionSchema = v.object({ + action: v.literal('read_unread_messages'), + channelId: v.string(), +}) + +export const ActionSchema = v.union([ + ContinueActionSchema, + BreakActionSchema, + SleepActionSchema, + ListChannelsActionSchema, + SendMessageActionSchema, + ReadUnreadMessagesActionSchema, +]) + +export type Action = v.InferOutput + export interface CancellablePromise { promise: Promise cancel: () => void @@ -23,16 +64,21 @@ export function cancellable(promise: Promise): CancellablePromise { } export interface PendingEvent { + id: string event: SatoriEvent status: 'pending' | 'ready' } +export interface StoredUnreadEvent { + id: string + event: SatoriEvent +} + export interface BotContext { logger: Logg eventQueue: PendingEvent[] - unreadEvents: Record // channelId -> events + unreadEvents: Record // channelId -> events processedIds: Set - processing: boolean lastInteractedChannelIds: string[] currentProcessingStartTime?: number chats: Map @@ -42,47 +88,10 @@ export interface ChatContext { channelId: string platform: string selfId: string + isProcessing: boolean currentTask?: CancellablePromise currentAbortController?: AbortController - messages: LLMMessage[] actions: { action: Action, result: unknown }[] } - -// Action types -export interface ContinueAction { - action: 'continue' -} - -export interface BreakAction { - action: 'break' -} - -export interface SleepAction { - action: 'sleep' - seconds?: number -} - -export interface ListChannelsAction { - action: 'list_channels' -} - -export interface SendMessageAction { - action: 'send_message' - content: string - channelId: string -} - -export interface ReadUnreadMessagesAction { - action: 'read_unread_messages' - channelId: string -} - -export type Action - = | ContinueAction - | BreakAction - | SleepAction - | ListChannelsAction - | SendMessageAction - | ReadUnreadMessagesAction diff --git a/services/satori-bot/src/core/utils.ts b/services/satori-bot/src/core/utils.ts index b7360cc5c..291d6ae40 100644 --- a/services/satori-bot/src/core/utils.ts +++ b/services/satori-bot/src/core/utils.ts @@ -1,8 +1,38 @@ import type { SatoriEvent, SatoriMessage } from '../adapter/satori/types' -import type { BotContext, ChatContext } from './types' +import type { Action, BotContext, ChatContext } from './types' +/** + * Intelligently truncate action history while preserving logical chains. + * If the first action in the kept list is a 'continue', it backtracks to include + * the action that triggered it, ensuring the LLM has full context of its sequence. + */ +export function trimActions( + actions: { action: Action, result: unknown }[], + max: number, + keep: number, +): { action: Action, result: unknown }[] { + if (actions.length <= max) { + return actions + } + + let startIndex = actions.length - keep + + // Backtrack to avoid starting with a 'continue' action which lacks its previous context + while (startIndex > 0) { + const currentAction = actions[startIndex].action + if (currentAction.action === 'continue') { + startIndex-- + } + else { + break + } + } + + return actions.slice(startIndex) +} /** * Safely extract string content from message +... * Handles string, array, and other types */ export function getMessageContentString(content: unknown): string { @@ -56,15 +86,8 @@ export function formatDebugContext( if (chatCtx) { context.channelId = chatCtx.channelId - context.totalMessagesInContext = chatCtx.messages.length context.totalActionsInContext = chatCtx.actions.length - const lastMessages = chatCtx.messages.slice(-3).map(msg => ({ - role: msg.role, - content: getMessageContentString(msg.content).substring(0, 50), - })) - context.lastMessages = lastMessages - const lastActions = chatCtx.actions.slice(-3).map(action => ({ action: action.action.action, result: typeof action.result === 'string' ? action.result.substring(0, 100) : String(action.result).substring(0, 100), diff --git a/services/satori-bot/src/index.ts b/services/satori-bot/src/index.ts index 359f422d3..5718352a7 100644 --- a/services/satori-bot/src/index.ts +++ b/services/satori-bot/src/index.ts @@ -1,10 +1,11 @@ -import process, { env } from 'node:process' +import process from 'node:process' import { Format, LogLevel, setGlobalFormat, setGlobalLogLevel, useLogg } from '@guiiai/logg' +import { config } from './config' import { SatoriClient } from './adapter/satori/client' import { globalRegistry } from './capabilities/registry' -import { createBotContext, setupMessageEventHandler, setupReadyEventHandler, startPeriodicLoop } from './core' +import { createBotContext, setupMessageEventHandler, setupReadyEventHandler, startPeriodicLoop } from './core/index' import { initDb } from './lib/db' setGlobalFormat(Format.Pretty) @@ -19,13 +20,13 @@ async function main() { // Create Satori client const satoriClient = new SatoriClient({ - url: env.SATORI_WS_URL || 'ws://localhost:5140/satori/v1/events', - token: env.SATORI_TOKEN, - apiBaseUrl: env.SATORI_API_BASE_URL, + url: config.satori.wsUrl, + token: config.satori.token, + apiBaseUrl: config.satori.apiBaseUrl, }) // Create bot context - const botContext = createBotContext(log) + const botContext = await createBotContext(log) // Set up event handlers setupReadyEventHandler(satoriClient, log) @@ -44,10 +45,23 @@ async function main() { process.on('unhandledRejection', (err) => { const log = useLogg('UnhandledRejection').useGlobalConfig() + const cause = (err instanceof Error && 'cause' in err) ? err.cause : undefined log .withError(err as Error) - .withField('cause', (err as any).cause) - .error('Unhandled rejection') + .withField('cause', cause) + .error('Unhandled rejection occurred') +}) + +process.on('uncaughtException', (err) => { + const log = useLogg('UncaughtException').useGlobalConfig() + log + .withError(err) + .error('Uncaught exception occurred') +}) + +main().catch((err) => { + const log = useLogg('Main').useGlobalConfig() + log.withError(err).error('Fatal error in main loop') + process.exit(1) }) -main().catch(console.error) diff --git a/services/satori-bot/src/lib/db.ts b/services/satori-bot/src/lib/db.ts index 825283a6b..a3edb933d 100644 --- a/services/satori-bot/src/lib/db.ts +++ b/services/satori-bot/src/lib/db.ts @@ -1,90 +1,170 @@ -import { join } from 'node:path' +import type { SatoriEvent } from '../adapter/satori/types' +import type { StoredUnreadEvent } from '../core/types' + +import { resolve } from 'node:path' import { fileURLToPath } from 'node:url' -import { Low } from 'lowdb' -import { JSONFile } from 'lowdb/node' +import { PGlite } from '@electric-sql/pglite' +import { desc, eq, inArray } from 'drizzle-orm' +import { drizzle } from 'drizzle-orm/pglite' +import { migrate } from 'drizzle-orm/pglite/migrator' +import { nanoid } from 'nanoid' + +import { config } from '../config' + +import * as schema from './schema' const __dirname = fileURLToPath(new URL('.', import.meta.url)) +const dbPath = resolve(__dirname, config.db.path) -interface Channel { - id: string - name: string - platform: string - selfId: string -} - -interface Message { - id: string - channelId: string - userId: string - userName: string - content: string - timestamp: number -} - -interface Database { - channels: Channel[] - messages: Message[] -} - -const defaultData: Database = { - channels: [], - messages: [], -} - -const file = join(__dirname, '../../data/db.json') -const adapter = new JSONFile(file) -const db = new Low(adapter, defaultData) +// Initialize PGlite and Drizzle +const client = new PGlite(dbPath) +export const db = drizzle(client, { schema }) export async function initDb() { - await db.read() - db.data ||= defaultData - await db.write() + // Execute migrations + const migrationsPath = resolve(__dirname, '../../drizzle') + await migrate(db, { migrationsFolder: migrationsPath }) } +export const { channels, messages, eventQueue, unreadEvents } = schema + export async function recordChannel(id: string, name: string, platform: string, selfId: string) { - await db.read() - - // Find existing channel by ID only (platform/selfId may change on restart) - const existingIndex = db.data.channels.findIndex(c => c.id === id) - - if (existingIndex >= 0) { - // Update existing channel with new platform/selfId - db.data.channels[existingIndex] = { id, name, platform, selfId } - await db.write() - } - else { - // Create new channel - db.data.channels.push({ id, name, platform, selfId }) - await db.write() - } + await db.insert(channels) + .values({ id, name, platform, selfId }) + .onConflictDoUpdate({ + target: channels.id, + set: { name, platform, selfId }, + }) } -export async function listChannels(): Promise { - await db.read() - return db.data.channels +export async function listChannels() { + return await db.select().from(channels) } -export async function recordMessage(channelId: string, userId: string, userName: string, content: string) { - await db.read() +export async function recordMessage(channelId: string, userId: string, userName: string, content: string, timestamp?: number) { + const ts = timestamp || Date.now() + const id = nanoid() - const message: Message = { - id: `${channelId}-${userId}-${Date.now()}`, + await db.insert(messages).values({ + id, channelId, userId, userName, content, - timestamp: Date.now(), - } - - db.data.messages.push(message) - - // Keep only last 1000 messages - if (db.data.messages.length > 1000) { - db.data.messages = db.data.messages.slice(-1000) - } - - await db.write() + timestamp: ts, + }) } -export { db } +/** + * Retrieves the most recent messages for a specific channel. + */ +export async function getRecentMessages(channelId: string, limit: number = 10) { + return await db.select() + .from(messages) + .where(eq(messages.channelId, channelId)) + .orderBy(desc(messages.timestamp)) + .limit(limit) + .then(msgs => msgs.reverse()) +} + +// Event Queue Persistence + +export async function pushToEventQueue(item: { event: SatoriEvent, status: 'pending' | 'ready' }) { + const id = nanoid() + await db.insert(eventQueue).values({ + id, + event: item.event, + status: item.status, + createdAt: Date.now(), + }) + return id +} + +export async function removeFromEventQueue(id: string) { + await db.delete(eventQueue).where(eq(eventQueue.id, id)) +} + +export async function clearEventQueue() { + await db.delete(eventQueue) +} + +export async function saveEventQueue(queue: { id?: string, event: SatoriEvent, status: 'pending' | 'ready' }[]) { + // If we have IDs, we might be able to do something smarter, but for now let's just keep it as is + // but optimized for the common case where we might want to just replace all. + // Actually, the best way to handle this is to NOT use saveEventQueue for single items. + await db.delete(eventQueue) + if (queue.length > 0) { + await db.insert(eventQueue).values(queue.map(item => ({ + id: item.id || nanoid(), + event: item.event, + status: item.status, + createdAt: Date.now(), + }))) + } +} + +export async function loadEventQueue() { + const result = await db.select().from(eventQueue).orderBy(schema.eventQueue.createdAt) + return result.map(r => ({ + id: r.id, + event: r.event as SatoriEvent, + status: r.status as 'pending' | 'ready', + })) +} + +// Unread Events Persistence + +export async function pushToUnreadEvents(channelId: string, event: SatoriEvent) { + const id = nanoid() + await db.insert(unreadEvents).values({ + id, + channelId, + event, + createdAt: Date.now(), + }) + return id +} + +export async function deleteUnreadEventsByIds(channelId: string, ids: string[]) { + if (ids.length === 0) + return + await db.delete(unreadEvents).where(inArray(unreadEvents.id, ids)) +} + +export async function clearUnreadEventsForChannel(channelId: string) { + await db.delete(unreadEvents).where(eq(unreadEvents.channelId, channelId)) +} + +export async function saveUnreadEvents(allUnread: Record) { + await db.delete(unreadEvents) + const values = [] + for (const [channelId, events] of Object.entries(allUnread)) { + for (const item of events) { + values.push({ + id: item.id || nanoid(), + channelId, + event: item.event, + createdAt: Date.now(), + }) + } + } + if (values.length > 0) { + await db.insert(unreadEvents).values(values) + } +} + +export async function loadUnreadEvents() { + const result = await db.select().from(unreadEvents).orderBy(schema.unreadEvents.createdAt) + const allUnread: Record = {} + for (const r of result) { + if (!allUnread[r.channelId]) { + allUnread[r.channelId] = [] + } + allUnread[r.channelId].push({ + id: r.id, + event: r.event as SatoriEvent, + }) + } + return allUnread +} diff --git a/services/satori-bot/src/lib/schema.ts b/services/satori-bot/src/lib/schema.ts new file mode 100644 index 000000000..b285fc327 --- /dev/null +++ b/services/satori-bot/src/lib/schema.ts @@ -0,0 +1,35 @@ +import { pgTable, text, bigint, index, json } from 'drizzle-orm/pg-core' + +export const channels = pgTable('channels', { + id: text('id').primaryKey(), + name: text('name').notNull(), + platform: text('platform').notNull(), + selfId: text('self_id').notNull(), +}) + +export const messages = pgTable('messages', { + id: text('id').primaryKey(), + channelId: text('channel_id').notNull(), + userId: text('user_id').notNull(), + userName: text('user_name').notNull(), + content: text('content').notNull(), + timestamp: bigint('timestamp', { mode: 'number' }).notNull(), +}, (table) => { + return [ + index('channel_timestamp_idx').on(table.channelId, table.timestamp), + ] +}) + +export const eventQueue = pgTable('event_queue', { + id: text('id').primaryKey(), + event: json('event').notNull(), + status: text('status').notNull(), // 'pending' | 'ready' + createdAt: bigint('created_at', { mode: 'number' }).notNull(), +}) + +export const unreadEvents = pgTable('unread_events', { + id: text('id').primaryKey(), + channelId: text('channel_id').notNull(), + event: json('event').notNull(), + createdAt: bigint('created_at', { mode: 'number' }).notNull(), +}) diff --git a/services/satori-bot/todolist.md b/services/satori-bot/todolist.md new file mode 100644 index 000000000..4426d2929 --- /dev/null +++ b/services/satori-bot/todolist.md @@ -0,0 +1,34 @@ +# Satori Bot 待办事项 (优化版) + +## 🟡 P1: 架构完善与逻辑优化 (性能与体验) + +### 1. 重构 Action 截断逻辑，防止上下文失忆 +**目标**: 确保 LLM 的历史记录截断不会破坏逻辑链（如 `send_message` + `continue`）。 +- [ ] **策略层 (进阶)**: + - [ ] 尝试将较旧的 `actions` 压缩为文本 Summary 存入 LLM Context，而非直接丢弃。 + +## 🟢 P2: 增强功能与类型安全 + +- [ ] **监控增强**: + - [ ] 为所有 Action 执行增加更详细的 Trace 日志。 +- [ ] **类型收紧**: + - [ ] 持续检查并消除残留的 `as any` 类型断言。 + +--- + +## ✅ 已完成事项 (归档) + +### 🛡️ 1. 核心稳定性与并发架构 (P0) +- [x] **修复短时记忆清理的竞态条件**: 实现基于 ID 的精准删除，防止异步消息丢失。 +- [x] **消除全局锁死锁**: 将锁粒度下放到 Channel 级别，并引入 `try...finally` 强制释放机制。 +- [x] **实现非阻塞调度**: `onMessageArrival` 与周期性任务改为并发执行，提升吞吐量。 + +### ⚙️ 2. 逻辑链完整性与体验优化 (P1) +- [x] **智能 Action 截断**: 实现 `trimActions` 自动回溯，防止截断破坏 `continue` 等成对逻辑链。 +- [x] **阻断 API 滥用**: 在 `handleLoopStep` 中增加 `MAX_LOOP_ITERATIONS = 5` 的硬性上限。 + +### 💎 3. 类型安全与基础设施 (P2) +- [x] **修复 Satori API 类型不匹配**: 使 `SatoriMessageCreateResponse` 的接口定义与运行时 Schema 保持一致。 +- [x] **移除全局暴力退出**: 在 `process.on('unhandledRejection')` 中移除 `process.exit(1)`。 +- [x] **消除 Any 类型滥用**: 修复了 LLM 解析、数据库 Schema 等多处的类型退化。 +- [x] **重写队列持久化 I/O**: 实现 Drizzle ORM 的增量更新模式。