From 469f2f2c35edb20c8a202feac2ce9a6c80f551df Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?=E9=9B=B7=E7=94=B5=E8=8A=BD=E8=A1=A3?= Date: Thu, 8 Oct 2026 04:21:57 -0400 Subject: [PATCH] Q15: show which messages each agent actually read Runs record a context manifest once a model turn succeeds; the chat marks each human message with the agents that read its current version. Co-Authored-By: Claude Opus 5.5 --- .../migrations/061-task-run-context.ts | 6 +++ src/main/database/schema.ts | 2 + src/main/services/agent-run-loop.ts | 3 ++ src/main/services/chat-wake-context.ts | 38 +++++++++++++++++- src/main/services/task-execution-service.ts | 6 ++- src/main/services/task-prompts.ts | 1 + src/renderer/src/i18n/zh.ts | 1 + src/renderer/src/pages/ChatPage.tsx | 7 +++- .../src/pages/chat/ChatMessageRow.tsx | 6 +++ src/renderer/src/pages/chat/chat-helpers.ts | 27 ++++++++++++- src/shared/domain.ts | 2 + tests/integration/chat-context.test.ts | 13 ++++-- tests/unit/domain-services.test.ts | 2 +- tests/unit/message-readers.test.ts | 40 +++++++++++++++++++ 14 files changed, 144 insertions(+), 10 deletions(-) create mode 100644 src/main/database/migrations/061-task-run-context.ts create mode 100644 tests/unit/message-readers.test.ts diff --git a/src/main/database/migrations/061-task-run-context.ts b/src/main/database/migrations/061-task-run-context.ts new file mode 100644 index 00000000..0c5afc89 --- /dev/null +++ b/src/main/database/migrations/061-task-run-context.ts @@ -0,0 +1,6 @@ +export const taskRunContextSchema = ` + alter table task_runs add column context_json text; + + insert into schema_migrations(version, name, applied_at) + values (61, 'task_run_context', datetime('now')); +`; diff --git a/src/main/database/schema.ts b/src/main/database/schema.ts index f5d93a0e..2d9f9768 100644 --- a/src/main/database/schema.ts +++ b/src/main/database/schema.ts @@ -59,6 +59,7 @@ import { kimiCodeProviderSchema } from "./migrations/057-kimi-code-provider"; import { integrationKeysSchema } from "./migrations/058-integration-keys"; import { taskHandoffSchema } from "./migrations/059-task-handoff"; import { chatWakeTextPurgeSchema } from "./migrations/060-chat-wake-text-purge"; +import { taskRunContextSchema } from "./migrations/061-task-run-context"; interface Migration { version: number; @@ -127,6 +128,7 @@ const MIGRATIONS: Migration[] = [ { version: 58, name: "integration_keys", sql: integrationKeysSchema }, { version: 59, name: "task_handoff", sql: taskHandoffSchema }, { version: 60, name: "chat_wake_text_purge", sql: chatWakeTextPurgeSchema }, + { version: 61, name: "task_run_context", sql: taskRunContextSchema }, ]; const SCHEMA_VERSION = MIGRATIONS[MIGRATIONS.length - 1].version; diff --git a/src/main/services/agent-run-loop.ts b/src/main/services/agent-run-loop.ts index 02a8a44d..416f414e 100644 --- a/src/main/services/agent-run-loop.ts +++ b/src/main/services/agent-run-loop.ts @@ -19,6 +19,8 @@ export interface AgentRunLoopInput { externalTools?: ToolDefinition[]; /** Messages that arrived mid-run, or null. */ pollNewMessages?: () => string | null; + /** Called after each model turn the provider answered. */ + onModelTurn?: () => void; } export interface AgentRunLoopResult { @@ -57,6 +59,7 @@ export async function runAgentLoop(deps: AgentRunLoopDeps, input: AgentRunLoopIn tools: allTools, }); lastRequestId = turn.requestId; + input.onModelTurn?.(); if (turn.toolCalls.length === 0) { if (toolCallCount === 0 && !turn.text.trim()) { diff --git a/src/main/services/chat-wake-context.ts b/src/main/services/chat-wake-context.ts index 36acbc36..9da33c15 100644 --- a/src/main/services/chat-wake-context.ts +++ b/src/main/services/chat-wake-context.ts @@ -50,9 +50,44 @@ export function loadChatWakeContext(db: DatabaseSync, orgId: string, agentId: st spaceId: thread.space_id, history: [...(omitted > 0 ? [`(${omitted} earlier messages in this thread are not shown.)`] : []), ...history.map((m) => formatMessage(db, m))].join("\n"), fresh: fresh.map((m) => formatMessage(db, m)).join("\n"), + manifest: { history: history.map(readEntry), fresh: fresh.map(readEntry), omittedCount: omitted }, }; } +/** A message as the model saw it: id plus edit-or-create time. */ +export interface ReadEntry { + id: string; + version: string; +} + +/** What a run read: thread messages up front plus anything injected mid-run. */ +export interface RunContextManifest { + threadId: string | null; + history: ReadEntry[]; + fresh: ReadEntry[]; + omittedCount: number; + midRun: ReadEntry[]; +} + +/** Collects what the run read; commit() persists it once a model turn succeeds. */ +export function createRunContextRecorder(db: DatabaseSync, runId: string, chatContext: ChatContextBlock | null) { + const manifest: RunContextManifest = { + threadId: chatContext?.threadId ?? null, + history: chatContext?.manifest.history ?? [], + fresh: chatContext?.manifest.fresh ?? [], + omittedCount: chatContext?.manifest.omittedCount ?? 0, + midRun: [], + }; + return { + addMidRun: (entries: ReadEntry[]) => { manifest.midRun.push(...entries); }, + commit: () => { db.prepare("update task_runs set context_json = ? where id = ?").run(JSON.stringify(manifest), runId); }, + }; +} + +function readEntry(m: ThreadMessageRow): ReadEntry { + return { id: m.id, version: m.edited_at ?? m.created_at }; +} + function formatMessage(db: DatabaseSync, m: ThreadMessageRow): string { const who = m.sender_role ? `${m.sender_name} (${m.sender_type}, ${m.sender_role})` : `${m.sender_name} (${m.sender_type})`; const edited = m.edited_at ? ` (edited ${utcMinute(m.edited_at)})` : ""; @@ -66,7 +101,7 @@ function utcMinute(iso: string): string { /** Polls the threads this run is part of for messages from others. */ export function createRunMessagePoller( db: DatabaseSync, - input: { orgId: string; agentId: string; sourceThreadId: string | null; runStartedAt: string }, + input: { orgId: string; agentId: string; sourceThreadId: string | null; runStartedAt: string; onInjected: (entries: ReadEntry[]) => void }, ): () => string | null { let watermark = input.runStartedAt; return () => { @@ -81,6 +116,7 @@ export function createRunMessagePoller( `).all(input.orgId, watermark, input.agentId, input.sourceThreadId, input.orgId, input.agentId, input.runStartedAt) as unknown as ThreadMessageRow[]; if (rows.length === 0) return null; watermark = rows[rows.length - 1].created_at; + input.onInjected(rows.map(readEntry)); console.info(`[run-inbox] agent ${input.agentId}: ${rows.length} new message(s) injected mid-run`); return ["Messages that arrived while you were working (they may change what you should do):", ...rows.map((m) => formatMessage(db, m))].join("\n"); }; diff --git a/src/main/services/task-execution-service.ts b/src/main/services/task-execution-service.ts index 63209cb4..c405c971 100644 --- a/src/main/services/task-execution-service.ts +++ b/src/main/services/task-execution-service.ts @@ -24,7 +24,7 @@ import { type WorkspaceRow, } from "./task-prompts"; import { renderTaskHandoff } from "./agent-task-handoff"; -import { createRunMessagePoller, loadChatWakeContext } from "./chat-wake-context"; +import { createRunContextRecorder, createRunMessagePoller, loadChatWakeContext } from "./chat-wake-context"; import { runningWorkspaceWriter } from "./workspace-write-lock"; import { loadComposioTools } from "./composio/composio-tool-bridge"; import { connectorDisplayName } from "./composio/connector-brands"; @@ -159,6 +159,7 @@ export class TaskExecutionService { // Watermark for reply delivery and mid-run message polling. const runStartedAt = new Date().toISOString(); const chatContext = loadChatWakeContext(this.db, orgId, agent.id, isolation.secretScopeJson); + const contextRecorder = createRunContextRecorder(this.db, runId, chatContext); const toolContext: AgentToolContext = { orgId, agent: { id: agent.id, modelProviderId: agent.model_provider_id, modelId: agent.model_id }, @@ -206,7 +207,8 @@ export class TaskExecutionService { }), toolContext, externalTools, - pollNewMessages: createRunMessagePoller(this.db, { orgId, agentId: agent.id, sourceThreadId: chatContext?.threadId ?? null, runStartedAt }), + onModelTurn: contextRecorder.commit, + pollNewMessages: createRunMessagePoller(this.db, { orgId, onInjected: contextRecorder.addMidRun, agentId: agent.id, sourceThreadId: chatContext?.threadId ?? null, runStartedAt }), }, ); providerRequestId = result.lastRequestId; diff --git a/src/main/services/task-prompts.ts b/src/main/services/task-prompts.ts index fc6c298e..614aaf00 100644 --- a/src/main/services/task-prompts.ts +++ b/src/main/services/task-prompts.ts @@ -76,6 +76,7 @@ export interface ChatContextBlock { history: string; /** Messages after it: what this run answers. */ fresh: string; + manifest: { history: Array<{ id: string; version: string }>; fresh: Array<{ id: string; version: string }>; omittedCount: number }; } export interface AgentInstructionContext { diff --git a/src/renderer/src/i18n/zh.ts b/src/renderer/src/i18n/zh.ts index cd7611cc..723ee5fc 100644 --- a/src/renderer/src/i18n/zh.ts +++ b/src/renderer/src/i18n/zh.ts @@ -27,6 +27,7 @@ export const zh: Record = { "Advance": "推进", "Advanced": "高级", "Integrations": "集成密钥", + "Read by {names}": "已读：{names}", "Keys are stored on this device, encrypted by the operating system, and shared by every company.": "密钥保存在本机，由操作系统加密，所有公司共用。", "Composio API key": "Composio API 密钥", "Powers the Connectors page and agent connector tools. Create one at platform.composio.dev.": "用于连接器页面和 Agent 的连接器工具。在 platform.composio.dev 创建。", diff --git a/src/renderer/src/pages/ChatPage.tsx b/src/renderer/src/pages/ChatPage.tsx index 953522d6..69c706f5 100644 --- a/src/renderer/src/pages/ChatPage.tsx +++ b/src/renderer/src/pages/ChatPage.tsx @@ -8,6 +8,7 @@ import { buildConversations, collectParticipants, filterApprovalsForSpace, + messageReaders, } from "./chat/chat-helpers"; import type { PageProps } from "./types"; import { activeOrgId, requireActiveEmployee } from "./types"; @@ -27,6 +28,10 @@ export function ChatPage({ snapshot, actions }: PageProps) { const agents = useMemo(() => snapshot.agents.filter((item) => item.orgId === orgId), [snapshot.agents, orgId]); const files = useMemo(() => snapshot.files.filter((item) => item.orgId === orgId), [snapshot.files, orgId]); const messagesAll = useMemo(() => snapshot.chatMessages.filter((m) => m.orgId === orgId), [snapshot.chatMessages, orgId]); + const readersByMessage = useMemo( + () => messageReaders(snapshot.taskRuns.filter((run) => run.orgId === orgId), agents, messagesAll), + [snapshot.taskRuns, orgId, agents, messagesAll], + ); const messageIds = useMemo(() => new Set(messagesAll.map((message) => message.id)), [messagesAll]); const mentionsAll = useMemo( () => snapshot.chatMentions.filter((mention) => messageIds.has(mention.messageId)), @@ -202,7 +207,7 @@ export function ChatPage({ snapshot, actions }: PageProps) { mentions={mentionsAll} slashInvocations={slashInvocationsAll} files={files} - context={{ employees, agents }} + context={{ employees, agents, readersByMessage }} attachmentFileId={attachmentFileId} currentEmployeeId={employee.id} onAttachmentChange={setAttachmentFileId} diff --git a/src/renderer/src/pages/chat/ChatMessageRow.tsx b/src/renderer/src/pages/chat/ChatMessageRow.tsx index 067dbd1f..50681bae 100644 --- a/src/renderer/src/pages/chat/ChatMessageRow.tsx +++ b/src/renderer/src/pages/chat/ChatMessageRow.tsx @@ -41,6 +41,7 @@ export function ChatMessageRow({ const messageFiles = deleted ? [] : messageAttachments(message.id, attachments, files); const messageMentions = mentions.filter((mention) => mention.messageId === message.id); const slashInvocation = deleted ? null : slashInvocations.find((invocation) => invocation.messageId === message.id) ?? null; + const readers = bot || deleted ? [] : context.readersByMessage.get(message.id) ?? []; const submitEdit = (event: FormEvent) => { event.preventDefault(); const next = new FormData(event.currentTarget).get("body"); @@ -99,6 +100,11 @@ export function ChatMessageRow({ })} )} + {readers.length > 0 && ( +
+ {t("Read by {names}", { names: readers.join(", ") })} +
+ )} ); diff --git a/src/renderer/src/pages/chat/chat-helpers.ts b/src/renderer/src/pages/chat/chat-helpers.ts index 4d8e4682..f8e95453 100644 --- a/src/renderer/src/pages/chat/chat-helpers.ts +++ b/src/renderer/src/pages/chat/chat-helpers.ts @@ -8,15 +8,38 @@ import type { ChatSpaceRecord, EmployeeRecord, FileObjectRecord, + TaskRunRecord, } from "../../../../shared/domain"; import { t } from "../../i18n/i18n"; export interface MessageContext { employees: EmployeeRecord[]; agents: AgentRecord[]; + /** Agent names whose runs read each message. */ + readersByMessage: Map; } -export function senderName(message: ChatMessageRecord, context: MessageContext): string { +/** Which agents' runs read the current version of each message. */ +export function messageReaders(taskRuns: TaskRunRecord[], agents: AgentRecord[], messages: ChatMessageRecord[]): Map { + const currentVersion = new Map(messages.map((message) => [message.id, message.editedAt ?? message.createdAt])); + const readers = new Map(); + for (const run of taskRuns) { + if (!run.contextJson || !run.agentId) continue; + const agent = agents.find((item) => item.id === run.agentId); + if (!agent) continue; + type Entry = { id: string; version: string }; + const manifest = JSON.parse(run.contextJson) as { history: Entry[]; fresh: Entry[]; midRun: Entry[] }; + for (const entry of [...manifest.history, ...manifest.fresh, ...manifest.midRun]) { + if (currentVersion.get(entry.id) !== entry.version) continue; + const names = readers.get(entry.id) ?? []; + if (!names.includes(agent.name)) names.push(agent.name); + readers.set(entry.id, names); + } + } + return readers; +} + +export function senderName(message: ChatMessageRecord, context: Pick): string { // A message whose principal no longer resolves must degrade to a visible // placeholder, not throw: one orphaned row would otherwise blank the whole // Messages page mid-render. @@ -128,7 +151,7 @@ export function buildConversations(input: { if (group) group.push(membership); else membersBySpace.set(membership.spaceId, [membership]); } - const context: MessageContext = { employees: input.employees, agents: input.agents }; + const context = { employees: input.employees, agents: input.agents }; const items = input.spaces.map((space) => { const last = lastBySpace.get(space.id) ?? null; diff --git a/src/shared/domain.ts b/src/shared/domain.ts index 9f5b3df0..a3e58e8c 100644 --- a/src/shared/domain.ts +++ b/src/shared/domain.ts @@ -151,6 +151,8 @@ export interface TaskRunRecord { startedAt: string | null; completedAt: string | null; errorMessage: string | null; + /** JSON RunContextManifest: which messages the run read. */ + contextJson: string | null; } export interface ApprovalRecord { diff --git a/tests/integration/chat-context.test.ts b/tests/integration/chat-context.test.ts index d353d80e..cc26bbd7 100644 --- a/tests/integration/chat-context.test.ts +++ b/tests/integration/chat-context.test.ts @@ -182,6 +182,13 @@ describe("chat wake context", () => { expect(seenFollowUps[0]).toContain("Messages that arrived while you were working"); expect(seenFollowUps[0]).toContain("Stop — use Postgres instead."); expect(seenFollowUps[0]).not.toContain("Use MySQL."); + const entry = (body: string) => { + const row = connection.db.prepare("select id, created_at from chat_messages where body_md = ?").get(body) as { id: string; created_at: string }; + return { id: row.id, version: row.created_at }; + }; + const manifest = JSON.parse((connection.db.prepare("select context_json from task_runs where id = ?").get(wake.id) as { context_json: string }).context_json); + expect(manifest).toEqual({ threadId, history: [], fresh: [entry("Use MySQL.")], omittedCount: 0, midRun: [entry("Stop — use Postgres instead.")] }); + expect(firstId).toBe(entry("Use MySQL.").id); connection.close(); }); @@ -249,14 +256,14 @@ describe("chat wake context", () => { const threadId = h.say("Old plan: launch in March.", null); await h.runWake(); h.remove("Old plan: launch in March."); - h.say("Wrong number, ignore it: 4242.", threadId); - h.remove("Wrong number, ignore it: 4242."); + h.say("Wrong number, ignore it: QZX-WRONG.", threadId); + h.remove("Wrong number, ignore it: QZX-WRONG."); h.say("What is the launch date now?", threadId); await h.runWake(); const prompt = h.wakePrompts[1]; expect(prompt).toContain("What is the launch date now?"); expect(prompt).not.toContain("Old plan: launch in March."); - expect(prompt).not.toContain("4242"); + expect(prompt).not.toContain("QZX-WRONG"); h.connection.close(); }); diff --git a/tests/unit/domain-services.test.ts b/tests/unit/domain-services.test.ts index a65d12cb..2374df26 100644 --- a/tests/unit/domain-services.test.ts +++ b/tests/unit/domain-services.test.ts @@ -24,7 +24,7 @@ describe("domain services", () => { const connection = createConnection(); const rows = connection.db.prepare("select version from schema_migrations order by version asc").all() as Array<{ version: number }>; - expect(rows.map((row) => row.version)).toEqual([1, 2, 3, 4, 5, 6, 7, 8, 9, 10, 11, 12, 13, 14, 15, 16, 17, 18, 19, 20, 21, 22, 23, 24, 25, 26, 27, 28, 29, 30, 31, 32, 33, 34, 35, 36, 37, 38, 39, 40, 41, 42, 43, 44, 45, 46, 47, 48, 49, 50, 51, 52, 53, 54, 55, 56, 57, 58, 59, 60]); + expect(rows.map((row) => row.version)).toEqual([1, 2, 3, 4, 5, 6, 7, 8, 9, 10, 11, 12, 13, 14, 15, 16, 17, 18, 19, 20, 21, 22, 23, 24, 25, 26, 27, 28, 29, 30, 31, 32, 33, 34, 35, 36, 37, 38, 39, 40, 41, 42, 43, 44, 45, 46, 47, 48, 49, 50, 51, 52, 53, 54, 55, 56, 57, 58, 59, 60, 61]); connection.close(); }); diff --git a/tests/unit/message-readers.test.ts b/tests/unit/message-readers.test.ts new file mode 100644 index 00000000..b31d6755 --- /dev/null +++ b/tests/unit/message-readers.test.ts @@ -0,0 +1,40 @@ +import { describe, expect, test } from "vitest"; +import { messageReaders } from "../../src/renderer/src/pages/chat/chat-helpers"; +import type { AgentRecord, ChatMessageRecord, TaskRunRecord } from "../../src/shared/domain"; + +function run(agentId: string | null, contextJson: string | null): TaskRunRecord { + return { + id: `run_${Math.random()}`, orgId: "org", taskId: "tsk", agentId, providerId: null, providerRequestId: null, + executionContextId: null, serviceIdentityId: null, workspaceMountJson: "[]", secretScopeJson: "{}", + trigger: "chat_mention", state: "completed", createdAt: "", startedAt: null, completedAt: null, errorMessage: null, contextJson, + }; +} + +function message(id: string, createdAt: string, editedAt: string | null = null): ChatMessageRecord { + return { id, createdAt, editedAt } as ChatMessageRecord; +} + +const agents = [{ id: "a1", name: "CEO" }, { id: "a2", name: "Dana" }] as AgentRecord[]; +const entry = (id: string, version: string) => ({ id, version }); +const manifest = (history: object[], fresh: object[], midRun: object[]) => + JSON.stringify({ threadId: "t", history, fresh, omittedCount: 0, midRun }); + +describe("message readers", () => { + test("collects each agent once per message across history, fresh, and mid-run entries", () => { + const messages = [message("m1", "t1"), message("m2", "t2"), message("m3", "t3")]; + const readers = messageReaders([ + run("a1", manifest([entry("m1", "t1")], [entry("m2", "t2")], [entry("m3", "t3")])), + run("a1", manifest([entry("m2", "t2")], [], [])), + run("a2", manifest([], [entry("m2", "t2")], [])), + run("a2", null), + run(null, manifest([entry("m1", "t1")], [], [])), + ], agents, messages); + expect(Object.fromEntries(readers)).toEqual({ m1: ["CEO"], m2: ["CEO", "Dana"], m3: ["CEO"] }); + }); + + test("a message edited after it was read is unread until a run sees the new version", () => { + const messages = [message("m1", "t1", "t9")]; + expect(messageReaders([run("a1", manifest([], [entry("m1", "t1")], []))], agents, messages).size).toBe(0); + expect(Object.fromEntries(messageReaders([run("a1", manifest([], [entry("m1", "t9")], []))], agents, messages))).toEqual({ m1: ["CEO"] }); + }); +});