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 <noreply@anthropic.com>
This commit is contained in:
ZacharyZhang-NYandClaude Opus 5.5 committed 2026-10-08 04:21:57 -04:00
1 parent 592ce69219
commit 469f2f2c35
14 files changed
+144 -10

No files matched your search

@@ -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'));
`;
+2
View File
@@ -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;
+3
View File
@@ -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()) {
+37 -1
View File
@@ -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");
};
+4 -2
View File
@@ -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;
+1
View File
@@ -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 {
+1
View File
@@ -27,6 +27,7 @@ export const zh: Record<string, string> = {
"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 创建。",
+6 -1
View File
@@ -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}
@@ -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<HTMLFormElement>) => {
event.preventDefault();
const next = new FormData(event.currentTarget).get("body");
@@ -99,6 +100,11 @@ export function ChatMessageRow({
})}
</div>
)}
{readers.length > 0 && (
<div style={{ fontSize: 12, color: "var(--text-3)", marginTop: 4 }}>
{t("Read by {names}", { names: readers.join(", ") })}
</div>
)}
</div>
</div>
);
+25 -2
View File
@@ -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<string, string[]>;
}
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<string, string[]> {
const currentVersion = new Map(messages.map((message) => [message.id, message.editedAt ?? message.createdAt]));
const readers = new Map<string, string[]>();
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<MessageContext, "employees" | "agents">): 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;
+2
View File
@@ -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 {
+10 -3
View File
@@ -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();
});
+1 -1
View File
@@ -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();
});
+40
View File
@@ -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"] });
});
});