From f30ec740d8207bfab38f4986f1b70ff83f6f285e Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?=E9=9B=B7=E7=94=B5=E8=8A=BD=E8=A1=A3?= Date: Sun, 19 Apr 2026 18:34:49 -0400 Subject: [PATCH] Stop welding "delivered" onto "read" MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit The inbox projection used to call `markAgentInboxDelivered` after appending each message to the agent's session shadow. That helper sets `agent_messages.read_at = now()` for direct messages and inserts into `agent_message_reads` for channel messages — the SAME flags `listAgentInboxUnread` and the public `?unreadOnly=true` API use to define "unread". So the moment we put a message into the recipient's transcript, the recipient's own API call returned nothing pending. The IC prompt told the agent to "respond to ALL unread messages"; the API said zero unread; the agent posted a status comment instead of replying. Every silent-employee report traces back here. Replay safety lives entirely on `agent_message_projections` (installed in v46) — that ledger is keyed by `(message_id, session_id)` and `claimProjection` already short-circuits a re-run without touching `read_at` at all. The two concepts are now properly separated: - "delivered" (we put it in front of the agent) — tracked by the projection ledger, owned by the runtime bridge; - "read" (the agent acknowledged it) — tracked by `read_at` and `agent_message_reads`, only flipped by the explicit POST /messages/{id}/read endpoint the agent calls itself. `markAgentInboxDelivered` is removed because there is no remaining caller and its semantics were the bug — keeping it around would just invite the same mistake again. --- src/main/agent-runtime/runtime-bridge.ts | 25 ++++++++---------- .../agent-runtime/runtime-message-injector.ts | 26 ++++++++----------- src/main/database.ts | 1 - src/main/database/messages.ts | 23 ---------------- 4 files changed, 22 insertions(+), 53 deletions(-) diff --git a/src/main/agent-runtime/runtime-bridge.ts b/src/main/agent-runtime/runtime-bridge.ts index 6399d1e7..20ecbf3f 100644 --- a/src/main/agent-runtime/runtime-bridge.ts +++ b/src/main/agent-runtime/runtime-bridge.ts @@ -529,16 +529,17 @@ export class RuntimeBridge { // before loading priorMessages — the recipient must observe them // as plain user turns in its context. // - // The projection closure is now atomic: the ledger claim, the - // session-shadow append, the thread-event write, and the - // delivered-mark all commit inside one `BEGIN IMMEDIATE` - // transaction. Pre-v46 layout let those four writes drift apart, - // so a crash before the delivered-mark replayed the same inbox - // row on the next run and double-populated the transcript. The - // `agent_message_projections` ledger is the idempotent gate — - // `claimProjection` returns false when the `(message_id, - // session_id)` pair is already recorded, so a replay skips the - // whole closure. + // Replay safety lives entirely on the `agent_message_projections` + // ledger. We deliberately do NOT touch `agent_messages.read_at` + // here even though that column exists: "we put the message into + // the agent's transcript" and "the agent has acknowledged the + // message" are two different facts, and welding them onto one + // column made every projected message immediately invisible to + // the agent's `?unreadOnly=true` API call. The agent then read + // its IC prompt ("respond to ALL unread messages"), called the + // API, got an empty list, and posted a status comment instead of + // replying. The explicit POST /messages/{id}/read is now the + // sole signal that flips `read_at`. const store = this.getOrCreateStore(payload.taskSessionId); const inboxRecords = this.deps.db.listAgentInboxUnread(payload.companyId, payload.agentId); if (inboxRecords.length > 0) { @@ -562,10 +563,6 @@ export class RuntimeBridge { record, ); } - this.deps.db.markAgentInboxDelivered( - payload.agentId, - inboxRecords.map((r) => r.id), - ); }); } diff --git a/src/main/agent-runtime/runtime-message-injector.ts b/src/main/agent-runtime/runtime-message-injector.ts index 60177080..0a7d7b9e 100644 --- a/src/main/agent-runtime/runtime-message-injector.ts +++ b/src/main/agent-runtime/runtime-message-injector.ts @@ -128,11 +128,16 @@ export class RuntimeMessageInjector { } /** - * Routed from `worker-message-handler` on every `steer-ack`. Commits - * the inbox `delivered_at` write when the worker confirms the steer - * reached the live Agent; otherwise keeps the row unread and asks - * orchestration to wake the recipient so the next `prepareRun` picks - * the message up through the regular inbox projection. + * Routed from `worker-message-handler` on every `steer-ack`. Confirms + * the message reached the live Agent (the bridge already registered + * the inbound correlation when it built the projection). On rejection + * the recipient is woken so the next `prepareRun` picks the message + * up through the regular inbox projection. + * + * `read_at` is intentionally NOT touched here: "the worker queued + * this message onto the agent" is not "the agent replied". The + * agent flips `read_at` itself by POSTing /messages/{id}/read once + * it has actually processed the row. */ handleSteerAck(correlationId: string, accepted: boolean): void { this.resolvePending(correlationId, accepted, "worker_ack"); @@ -148,16 +153,7 @@ export class RuntimeMessageInjector { clearTimeout(entry.timer); this.pending.delete(correlationId); - if (accepted) { - try { - this.deps.db.markAgentInboxDelivered(entry.recipientAgentId, [entry.message.id]); - } catch (error) { - this.deps.logger.warn( - `[injector] markAgentInboxDelivered failed for ${entry.message.id}: ${error instanceof Error ? error.message : String(error)}`, - ); - } - return; - } + if (accepted) return; this.deps.logger.warn( `[injector] steer rejected (${source}) for message ${entry.message.id} to agent ${entry.recipientAgentId}; falling back to wake + inbox projection`, diff --git a/src/main/database.ts b/src/main/database.ts index 1890a45d..bd1a2835 100644 --- a/src/main/database.ts +++ b/src/main/database.ts @@ -20,7 +20,6 @@ export class AppDatabase extends AppDatabaseBase { searchMessages(companyId: string, query: string, opts?: { channel?: string; channelTargetId?: string | null; limit?: number }) { return this.messages.searchMessages(companyId, query, opts); } markMessageReadForReader(messageId: string, companyId: string, readerAgentId?: string | null) { return this.messages.markMessageReadForReader(messageId, companyId, readerAgentId); } listAgentInboxUnread(companyId: string, agentId: string) { return this.messages.listAgentInboxUnread(companyId, agentId); } - markAgentInboxDelivered(agentId: string, messageIds: string[]) { return this.messages.markAgentInboxDelivered(agentId, messageIds); } // ─── Social & Browser Actions ────────────────────────────────────────────── diff --git a/src/main/database/messages.ts b/src/main/database/messages.ts index 960ab1c1..785209a6 100644 --- a/src/main/database/messages.ts +++ b/src/main/database/messages.ts @@ -339,28 +339,5 @@ export function buildMessageMethods( }); }, - /** - * Transactionally mark a batch of messages as delivered to `agentId`. - * - Direct messages: sets agent_messages.read_at = now() if currently null. - * - All messages (direct + channel): insert into agent_message_reads with - * ON CONFLICT DO NOTHING so reprocessing is idempotent. - */ - markAgentInboxDelivered(agentId: string, messageIds: string[]): void { - if (messageIds.length === 0) return; - const ts = nowIso(); - deps.transaction(() => { - const db = getDb(); - const directUpdate = db.prepare( - "update agent_messages set read_at = ? where id = ? and to_agent_id = ? and read_at is null", - ); - const receiptInsert = db.prepare( - "insert into agent_message_reads (message_id, agent_id, read_at) values (?, ?, ?) on conflict(message_id, agent_id) do nothing", - ); - for (const id of messageIds) { - directUpdate.run(ts, id, agentId); - receiptInsert.run(id, agentId, ts); - } - }); - }, }; }