diff --git a/packages/server/src/server/agent/providers/claude/agent.request-journal.test.ts b/packages/server/src/server/agent/providers/claude/agent.request-journal.test.ts index 9be65da964..307365b5ac 100644 --- a/packages/server/src/server/agent/providers/claude/agent.request-journal.test.ts +++ b/packages/server/src/server/agent/providers/claude/agent.request-journal.test.ts @@ -7,6 +7,7 @@ import type { AgentTimelineItem } from "../../agent-sdk-types.js"; import { createTestLogger } from "../../../../test-utils/test-logger.js"; import { ClaudeAgentClient } from "./agent.js"; import { claudeProjectDirSync } from "./project-dir.js"; +import { projectTimelineRows } from "../../timeline-projection.js"; describe("Claude session request journal replay", () => { let root: string; @@ -137,4 +138,100 @@ describe("Claude session request journal replay", () => { }), ]); }); + + test.each(["remove", "user"])( + "replays a background completion once across queue enqueue and %s records", + async (delivery) => { + const content = + "\nbg-1\ncompleted\nBackground command completed\n"; + const items = await replay([ + { type: "queue-operation", operation: "enqueue", content }, + delivery === "remove" + ? { type: "queue-operation", operation: "remove", content } + : { + type: "user", + uuid: "notification-delivery", + origin: { kind: "task-notification" }, + message: { role: "user", content }, + }, + ]); + const projected = projectTimelineRows({ + rows: items.map((item, seq) => ({ item, seq, timestamp: "2026-10-23T00:00:00.000Z" })), + mode: "projected", + }); + expect(projected.map((row) => row.item)).toEqual([ + expect.objectContaining({ + type: "tool_call", + name: "task_notification", + status: "completed", + metadata: expect.objectContaining({ taskId: "bg-1" }), + }), + ]); + }, + ); + test("restores each Monitor event and completion of one task once", async () => { + const notifications = ["1", "2", "3", "4"].map( + (event) => + `\nmonitor-1\nMonitor event\n${event}\n`, + ); + notifications.push( + "\nmonitor-1\ncompleted\nMonitor stream ended\n", + ); + const items = await replay( + notifications.flatMap((content, index) => [ + { type: "queue-operation", operation: "enqueue", content }, + { type: "user", uuid: `delivery-${index}`, message: { role: "user", content } }, + ]), + ); + const projected = projectTimelineRows({ + rows: items.map((item, seq) => ({ item, seq, timestamp: "2026-10-23T00:00:00.000Z" })), + mode: "projected", + }); + expect(projected.map(({ item }) => item.type === "tool_call" && item.detail)).toEqual( + notifications.map((text, index) => + expect.objectContaining({ + label: index === 4 ? "Monitor stream ended" : "Monitor event", + text, + }), + ), + ); + }); + test.each(["user", "remove", "system"])( + "preserves identical notifications from separate occurrences with %s deliveries", + async (delivery) => { + const content = + "\nmonitor-repeat\ncompleted\nMonitor event\n"; + const queued = [1, 2].map(() => ({ type: "queue-operation", operation: "enqueue", content })); + const delivered = [1, 2].map((index) => { + if (delivery === "user") { + return { type: "user", uuid: `repeat-${index}`, message: { role: "user", content } }; + } + if (delivery === "remove") { + return { type: "queue-operation", operation: "remove", content }; + } + return { + type: "system", + subtype: "task_notification", + uuid: `repeat-${index}`, + task_id: "monitor-repeat", + status: "completed", + summary: "Monitor event", + }; + }); + const items = await replay([ + ...queued, + ...delivered, + { type: "queue-operation", operation: "enqueue", content }, + ]); + const projected = projectTimelineRows({ + rows: items.map((item, seq) => ({ item, seq, timestamp: "2026-10-23T00:00:00.000Z" })), + mode: "projected", + }); + expect(projected.map(({ item }) => item.type === "tool_call" && item.detail?.label)).toEqual([ + "Monitor event", + "Monitor event", + "Monitor event", + ]); + }, + ); }); diff --git a/packages/server/src/server/agent/providers/claude/agent.sub-agent-replay.test.ts b/packages/server/src/server/agent/providers/claude/agent.sub-agent-replay.test.ts index 8f2ba60907..666e701649 100644 --- a/packages/server/src/server/agent/providers/claude/agent.sub-agent-replay.test.ts +++ b/packages/server/src/server/agent/providers/claude/agent.sub-agent-replay.test.ts @@ -834,7 +834,7 @@ describe("ClaudeAgentSession persisted subagent replay", () => { .filter((item) => item.type === "tool_call" && item.name === "task_notification"), ).toEqual([ expect.objectContaining({ - callId: "task_notification_workflow-notification", + callId: expect.stringMatching(/^task_notification_[a-f0-9]+_1$/), status: "completed", }), ]); diff --git a/packages/server/src/server/agent/providers/claude/agent.ts b/packages/server/src/server/agent/providers/claude/agent.ts index 859e2d5f2e..a00f504f1a 100644 --- a/packages/server/src/server/agent/providers/claude/agent.ts +++ b/packages/server/src/server/agent/providers/claude/agent.ts @@ -6313,6 +6313,7 @@ export function convertClaudeHistoryEntry( const userTaskNotificationItem = mapTaskNotificationUserContentToToolCall({ content, messageId: userMessageId, + notificationId: typeof entry.notificationId === "string" ? entry.notificationId : undefined, }); if (userTaskNotificationItem) { return [userTaskNotificationItem]; diff --git a/packages/server/src/server/agent/providers/claude/task-notification-tool-call.test.ts b/packages/server/src/server/agent/providers/claude/task-notification-tool-call.test.ts index d538e14c29..18655a9d8d 100644 --- a/packages/server/src/server/agent/providers/claude/task-notification-tool-call.test.ts +++ b/packages/server/src/server/agent/providers/claude/task-notification-tool-call.test.ts @@ -97,4 +97,17 @@ describe("task-notification-tool-call", () => { expect(item).toBeNull(); }); + + it("keeps notifications without task IDs distinct by their message IDs", () => { + const items = ["message-1", "message-2"].map((messageId) => + mapTaskNotificationUserContentToToolCall({ + content: "completed", + messageId, + }), + ); + expect(items.map((item) => item?.callId)).toEqual([ + "task_notification_message-1", + "task_notification_message-2", + ]); + }); }); diff --git a/packages/server/src/server/agent/providers/claude/task-notification-tool-call.ts b/packages/server/src/server/agent/providers/claude/task-notification-tool-call.ts index 2db4beddb0..b269be70eb 100644 --- a/packages/server/src/server/agent/providers/claude/task-notification-tool-call.ts +++ b/packages/server/src/server/agent/providers/claude/task-notification-tool-call.ts @@ -12,6 +12,7 @@ const OptionalNonEmptyTrimmedStringSchema = z.preprocess( ); const TaskNotificationEnvelopeSchema = z.object({ + notificationId: z.string().optional(), messageId: z.string().nullable(), taskId: z.string().nullable(), toolUseId: z.string().nullable(), @@ -35,7 +36,9 @@ const TaskNotificationUserContentSchema = z.union([ const TaskNotificationHistoryRecordSchema = z .object({ + notificationId: z.string().optional(), type: z.string().optional(), + operation: z.string().optional(), subtype: z.string().optional(), uuid: z.string().optional(), message_id: z.string().optional(), @@ -52,6 +55,7 @@ const TaskNotificationHistoryRecordSchema = z const MapTaskNotificationUserContentToToolCallInputSchema = z.object({ content: z.unknown(), messageId: z.string().optional().nullable(), + notificationId: z.string().optional(), }); export type TaskNotificationEnvelope = z.infer; @@ -155,6 +159,7 @@ function parseTaskNotificationFromUserContent( } return TaskNotificationEnvelopeSchema.parse({ + notificationId: parsedInput.data.notificationId, messageId: toNonEmptyString(parsedInput.data.messageId), taskId: readTaskNotificationTagValue({ text: rawText, tagName: "task-id" }), toolUseId: @@ -186,6 +191,7 @@ function parseTaskNotificationFromSystemRecord(record: unknown): TaskNotificatio const rawText = toNonEmptyString(systemRecord.content); return TaskNotificationEnvelopeSchema.parse({ + notificationId: systemRecord.notificationId, messageId: toNonEmptyString(systemRecord.uuid) ?? toNonEmptyString(systemRecord.message_id), taskId: toNonEmptyString(systemRecord.task_id) ?? @@ -212,25 +218,76 @@ function parseTaskNotificationFromSystemRecord(record: unknown): TaskNotificatio }); } +/** Normalize notification identity for one transcript traversal. Queue bookkeeping and delivery + * are separate records of the same occurrence; each source counts repeated identical bodies + * independently so the second event is not merged with the first. */ +export function* assignTaskNotificationReplayIds(entries: Iterable): Generator { + const occurrences = new Map>(); + for (const entry of entries) { + const record = TaskNotificationHistoryRecordSchema.safeParse(entry); + if (!record.success) { + yield entry; + continue; + } + const data = record.data; + const envelope = + data.type === "user" + ? parseTaskNotificationFromUserContent({ content: data.message?.content }) + : parseTaskNotificationFromSystemRecord(data); + if (!envelope?.taskId || !data.type) { + yield entry; + continue; + } + const signature = taskNotificationReplaySignature(envelope); + const source = data.type === "queue-operation" ? `queue-${data.operation}` : data.type; + const counts = occurrences.get(signature) ?? new Map(); + const occurrence = (counts.get(source) ?? 0) + 1; + counts.set(source, occurrence); + occurrences.set(signature, counts); + yield { ...entry, notificationId: `task_notification_${signature}_${occurrence}` }; + } +} + +function taskNotificationReplaySignature(envelope: TaskNotificationEnvelope): string { + // Known fields are canonical across XML user/queue records and structured system records. + // Keep the remaining body too: Monitor events share a summary but carry different s. + const remainder = (envelope.rawText ?? "") + .replace(/<\/?task-notification>/gi, "") + .replace( + /<(task-id|tool-use-id|tool_use_id|status|summary|output-file|output_file)>[\s\S]*?<\/\1>/gi, + "", + ) + .trim(); + return createHash("sha1") + .update( + JSON.stringify([ + envelope.taskId, + envelope.toolUseId, + envelope.status, + envelope.summary, + envelope.outputFile, + remainder, + ]), + ) + .digest("hex"); +} + function normalizeTaskNotificationCallIdSegment(segment: string): string | null { const normalized = segment.trim().replace(/[^a-zA-Z0-9._:-]+/g, "_"); return normalized.length > 0 ? normalized : null; } function buildTaskNotificationCallId(envelope: TaskNotificationEnvelope): string { + if (envelope.notificationId) return envelope.notificationId; const messageSegment = envelope.messageId ? normalizeTaskNotificationCallIdSegment(envelope.messageId) : null; - if (messageSegment) { - return `task_notification_${messageSegment}`; - } + if (messageSegment) return `task_notification_${messageSegment}`; const taskSegment = envelope.taskId ? normalizeTaskNotificationCallIdSegment(envelope.taskId) : null; - if (taskSegment) { - return `task_notification_${taskSegment}`; - } + if (taskSegment) return `task_notification_${taskSegment}`; const seed = [envelope.status, envelope.summary, envelope.outputFile, envelope.rawText] diff --git a/packages/server/src/server/agent/providers/claude/transcript-history.ts b/packages/server/src/server/agent/providers/claude/transcript-history.ts index cba606b0e4..f8663f83d0 100644 --- a/packages/server/src/server/agent/providers/claude/transcript-history.ts +++ b/packages/server/src/server/agent/providers/claude/transcript-history.ts @@ -1,3 +1,4 @@ +import { assignTaskNotificationReplayIds } from "./task-notification-tool-call.js"; import fs from "node:fs"; import path from "node:path"; import { normalizeProviderReplayTimestamp } from "../../provider-history-timestamps.js"; @@ -88,21 +89,24 @@ function indexedRecords( ): Iterable { return { *[Symbol.iterator]() { - for (const location of locations) { - const fd = descriptors.get(location.file)!; - const buffer = Buffer.allocUnsafe(location.length); - let read = 0; - while (read < buffer.length) { - const count = fs.readSync(fd, buffer, read, buffer.length - read, location.offset + read); - if (count === 0) - throw new Error(`Claude transcript was truncated during replay: ${location.file}`); - read += count; - } - const entry = parseRecord(buffer.toString("utf8")); - if (entry) yield entry; - } + yield* assignTaskNotificationReplayIds(readEntries()); }, }; + function* readEntries(): Generator { + for (const location of locations) { + const fd = descriptors.get(location.file)!; + const buffer = Buffer.allocUnsafe(location.length); + let read = 0; + while (read < buffer.length) { + const count = fs.readSync(fd, buffer, read, buffer.length - read, location.offset + read); + if (count === 0) + throw new Error(`Claude transcript was truncated during replay: ${location.file}`); + read += count; + } + const entry = parseRecord(buffer.toString("utf8")); + if (entry) yield entry; + } + } } function isToolResult(entry: TranscriptEntry): boolean {