refactor(satori-bot): migrate persistence to adapter (#1155)

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:

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

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/`

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.

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<string, ChatContext>` within the `BotContext` object (`src/core/types.ts`).
+* **Storage Mechanism**: Active chat contexts are stored in a native `Map<string, ChatContext>` 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`).

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',
+  },
+})

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");

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": {}
+  }
+}

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
+    }
+  ]
+}

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"
   }

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<T>(
     endpoint: string,
-    body?: any,
+    body?: unknown,
   ): Promise<T> {
     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<SatoriMessageCreateResponse[]>('/message.create', body)
+    const response = await this.request<unknown[]>('/message.create', body)
+    return v.parse(v.array(SatoriMessageCreateResponseSchema), response)
   }
 
-  async getMessage(channelId: string, messageId: string): Promise<any> {
-    return await this.request('/message.get', {
+  async getMessage(channelId: string, messageId: string): Promise<SatoriMessage> {
+    const response = await this.request<unknown>('/message.get', {
       channel_id: channelId,
       message_id: messageId,
     })
+    return v.parse(SatoriMessageSchema, response)
   }
 
   async deleteMessage(channelId: string, messageId: string): Promise<void> {

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<void> {
+    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<void> {
     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
     }

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 = <T extends v.BaseSchema<any, any, any>>(itemSchema: T) => v.object({
+  data: v.array(itemSchema),
+  next: v.optional(v.string()),
+})

services/satori-bot/src/adapter/satori/types.ts

@@ -14,7 +14,7 @@ export enum SatoriOpcode {
 }
 
 // WebSocket Signal Structure
-export interface SatoriSignal<T = any> {
+export interface SatoriSignal<T = unknown> {
   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<string, any>
+  arguments: unknown[]
+  options: Record<string, unknown>
 }
 
 // Interaction Button
@@ -127,7 +127,7 @@ export interface SatoriEvent {
   role?: SatoriGuildRole
   user?: SatoriUser
   _type?: string
-  _data?: Record<string, any>
+  _data?: Record<string, unknown>
 }
 
 // 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

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<ActionResult> => {
+    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}`)
 

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,

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<ActionResult> => {
-    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<ActionResult> => {
+    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 {

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<ActionResult>
 }

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<typeof ConfigSchema>
+
+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()

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

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<ActionResult> {
   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,
     })
 

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

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<void> {
-  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
   }
 }

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<string, SatoriEvent[]>
+    unreadEvents: Record<string, StoredUnreadEvent[]>
     incomingEvents?: SatoriEvent[]
   },
 ): Promise<Action | undefined> {
@@ -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<string, unknown>).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

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<BotContext> {
+  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<string, ChatContext>(),
   }
@@ -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: [],
   }
 

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<typeof ActionSchema>
+
 export interface CancellablePromise<T> {
   promise: Promise<T>
   cancel: () => void
@@ -23,16 +64,21 @@ export function cancellable<T>(promise: Promise<T>): CancellablePromise<T> {
 }
 
 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<string, SatoriEvent[]> // channelId -> events
+  unreadEvents: Record<string, StoredUnreadEvent[]> // channelId -> events
   processedIds: Set<string>
-  processing: boolean
   lastInteractedChannelIds: string[]
   currentProcessingStartTime?: number
   chats: Map<string, ChatContext>
@@ -42,47 +88,10 @@ export interface ChatContext {
   channelId: string
   platform: string
   selfId: string
+  isProcessing: boolean
 
   currentTask?: CancellablePromise<void>
   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

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),

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)

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<Database>(file)
-const db = new Low<Database>(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<Channel[]> {
-  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<string, StoredUnreadEvent[]>) {
+  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<string, StoredUnreadEvent[]> = {}
+  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
+}

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(),
+})

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 的增量更新模式。