From 6e47f28603e0b6919d73030b26c880c3742a03bb Mon Sep 17 00:00:00 2001 From: root Date: Sun, 12 Jul 2026 09:15:54 +0200 Subject: [PATCH] =?UTF-8?q?fix:=20=E5=B0=81=E5=8F=A3=E5=B7=B2=E5=AE=8C?= =?UTF-8?q?=E6=88=90=E5=8A=A9=E6=89=8B=E6=B6=88=E6=81=AF=E7=9A=84=E8=BF=9B?= =?UTF-8?q?=E5=BA=A6=E6=B5=81?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- src/backend/codex-stdio.ts | 30 ++++++++-- src/common/events.ts | 17 ++++-- src/mgr/result.ts | 4 +- src/runner/run-once.ts | 2 +- .../cases/36-stdio-event-reconstruction.ts | 9 +-- src/selftest/cases/80-assistant-final-seal.ts | 56 ++++++++++++++++++- src/selftest/fake-codex-app-server.ts | 40 ++++++++++--- 7 files changed, 132 insertions(+), 26 deletions(-) diff --git a/src/backend/codex-stdio.ts b/src/backend/codex-stdio.ts index b8a330c..86b0d24 100644 --- a/src/backend/codex-stdio.ts +++ b/src/backend/codex-stdio.ts @@ -158,19 +158,33 @@ export class CodexStdioEventReducer { private readonly assistantDeltaProgress = createAssistantDeltaProgressState(); private readonly completedAssistantMessages: CompletedAssistantMessage[] = []; private readonly completedAssistantItemIds = new Set(); + private anonymousAssistantDeltaOpen = true; private readonly deferredTerminalEvents: BackendEvent[] = []; private readonly suppressedNotifications = createSuppressedNotificationSummary(); reduce(message: JsonRecord): CodexStdioReducedNotification { const normalized = normalizeCodexNotification(message, this.suppressedNotifications); const events = [...normalized.events]; + if (normalized.startedAssistantMessage?.itemId === null) { + closeAssistantDeltaProgress(this.assistantDeltaProgress, null); + this.anonymousAssistantDeltaOpen = true; + } if (normalized.assistantDelta) { - this.assistantText += normalized.assistantDelta.text; - const progress = recordAssistantDeltaProgress(this.assistantDeltaProgress, normalized.assistantDelta); - if (progress) events.push(progress); + const itemId = normalized.assistantDelta.itemId; + if (itemId && this.completedAssistantItemIds.has(itemId)) { + recordSuppressedNotification(this.suppressedNotifications, "item/agentMessage/delta:completed-item-replay", "agentMessage"); + } else if (itemId === null && !this.anonymousAssistantDeltaOpen) { + recordSuppressedNotification(this.suppressedNotifications, "item/agentMessage/delta:completed-anonymous-replay", "agentMessage"); + } else { + this.assistantText += normalized.assistantDelta.text; + const progress = recordAssistantDeltaProgress(this.assistantDeltaProgress, normalized.assistantDelta); + if (progress) events.push(progress); + } } if (normalized.completedAssistantMessage) { const itemId = normalized.completedAssistantMessage.itemId; + closeAssistantDeltaProgress(this.assistantDeltaProgress, itemId); + if (itemId === null) this.anonymousAssistantDeltaOpen = false; if (itemId && this.completedAssistantItemIds.has(itemId)) { recordSuppressedNotification(this.suppressedNotifications, "item/completed:duplicate-agentMessage", "agentMessage"); } else { @@ -1192,7 +1206,7 @@ function escapeRegExp(value: string): string { return value.replace(/[.*+?^${}()|[\]\\]/gu, "\\$&"); } -function normalizeCodexNotification(message: JsonRecord, suppressed: SuppressedNotificationSummary): { events: BackendEvent[]; terminalEvents?: BackendEvent[]; assistantDelta?: { itemId: string | null; text: string }; completedAssistantMessage?: CompletedAssistantMessage; threadId?: string; turnId?: string; terminal?: { status: TerminalStatus; failureKind: FailureKind | null; message: string | null } } { +function normalizeCodexNotification(message: JsonRecord, suppressed: SuppressedNotificationSummary): { events: BackendEvent[]; terminalEvents?: BackendEvent[]; assistantDelta?: { itemId: string | null; text: string }; startedAssistantMessage?: { itemId: string | null }; completedAssistantMessage?: CompletedAssistantMessage; threadId?: string; turnId?: string; terminal?: { status: TerminalStatus; failureKind: FailureKind | null; message: string | null } } { const method = typeof message.method === "string" ? message.method : "unknown"; const params = asRecordAt(message, "params"); if (method === "thread/started") { @@ -1216,6 +1230,7 @@ function normalizeCodexNotification(message: JsonRecord, suppressed: SuppressedN if ((method === "item/started" || method === "item/completed") && asRecordAt(params, "item").type === "agentMessage") { const item = asRecordAt(params, "item"); const itemId = stringAt(item, "id") ?? stringAt(params, "itemId"); + if (method === "item/started") return { events: [], startedAssistantMessage: { itemId: itemId ?? null } }; const text = method === "item/completed" ? agentMessageText(item) : ""; const completedAssistantMessage = text.trim().length > 0 ? { itemId: itemId ?? null, text } : undefined; return { @@ -1349,8 +1364,9 @@ function finalAssistantMessageForTurn(completedMessages: CompletedAssistantMessa function assistantFinalSealEvent(message: FinalAssistantMessage): BackendEvent { return { - type: "assistant_message", + type: "backend_status", payload: { + phase: "assistant-message-final-sealed", itemId: message.itemId, source: "assistant-message-final-seal", messageIndex: message.messageIndex, @@ -1378,6 +1394,10 @@ function createAssistantDeltaProgressState(): AssistantDeltaProgressState { return new Map(); } +function closeAssistantDeltaProgress(state: AssistantDeltaProgressState, itemId: string | null): void { + state.delete(itemId ?? "default"); +} + function recordAssistantDeltaProgress(state: AssistantDeltaProgressState, delta: { itemId: string | null; text: string }): BackendEvent | null { if (!delta.text) return null; const key = delta.itemId ?? "default"; diff --git a/src/common/events.ts b/src/common/events.ts index 6147db7..c66c870 100644 --- a/src/common/events.ts +++ b/src/common/events.ts @@ -32,6 +32,12 @@ export function requireExternallyAppendableEventType(value: unknown): EventType } export function normalizeRunEventPayload(type: EventType, payload: JsonRecord): JsonRecord { + if (payload.finalSeal === true) { + if (type !== "backend_status" && type !== "assistant_message") { + throw new AgentRunError("schema-invalid", "assistant final seal must use backend_status or legacy assistant_message event type", { httpStatus: 400 }); + } + return normalizeAssistantFinalSeal(payload); + } if (type === "terminal_status") return normalizeTerminalStatusPayload(payload); if (type === "command_output") return normalizeCommandOutputPayload(payload); if (type === "user_message") return normalizeUserMessagePayload(payload); @@ -123,7 +129,6 @@ function nonEmptyText(value: unknown): string | null { function normalizeTextPayload(payload: JsonRecord): JsonRecord { const { text: _text, delta: _delta, content: _content, summary: _summary, textBytes: _textBytes, textTruncated: _textTruncated, outputBytes: _outputBytes, outputTruncated: _outputTruncated, ...rest } = payload; const value = typeof payload.text === "string" ? payload.text : typeof payload.delta === "string" ? payload.delta : typeof payload.content === "string" ? payload.content : ""; - if (payload.finalSeal === true) return normalizeAssistantFinalSeal(rest, value); if (payload.durableText === true) return normalizeDurableAssistantText(rest, value); if (payload.replyAuthority === true || payload.final === true) { const text = redactText(value); @@ -153,14 +158,16 @@ function normalizeDurableAssistantText(payload: JsonRecord, body: string): JsonR }; } -function normalizeAssistantFinalSeal(payload: JsonRecord, body: string): JsonRecord { - const replyRef = typeof payload.replyRef === "object" && payload.replyRef !== null && !Array.isArray(payload.replyRef) ? payload.replyRef as JsonRecord : null; +function normalizeAssistantFinalSeal(payload: JsonRecord): JsonRecord { + const { text: _text, delta: _delta, content: _content, summary: _summary, textBytes: _textBytes, textTruncated: _textTruncated, outputBytes: _outputBytes, outputTruncated: _outputTruncated, ...rest } = payload; + const body = typeof payload.text === "string" ? payload.text : typeof payload.delta === "string" ? payload.delta : typeof payload.content === "string" ? payload.content : ""; + const replyRef = typeof rest.replyRef === "object" && rest.replyRef !== null && !Array.isArray(rest.replyRef) ? rest.replyRef as JsonRecord : null; const textSha256 = replyRef && typeof replyRef.textSha256 === "string" ? replyRef.textSha256 : ""; if (body.length > 0) throw new AgentRunError("schema-invalid", "assistant final seal must not carry text, content, or delta", { httpStatus: 400 }); - if (payload.replyAuthority !== true || payload.final !== true || replyRef?.eventType !== "assistant_message" || typeof replyRef.source !== "string" || !/^[0-9a-f]{64}$/u.test(textSha256)) { + if (rest.replyAuthority !== true || rest.final !== true || replyRef?.eventType !== "assistant_message" || typeof replyRef.source !== "string" || !/^[0-9a-f]{64}$/u.test(textSha256)) { throw new AgentRunError("schema-invalid", "assistant final seal requires final replyAuthority and an assistant_message replyRef with source and textSha256", { httpStatus: 400 }); } - return { ...payload, replyRef: redactJson(replyRef), textBytes: 0, textTruncated: false, outputBytes: 0, outputTruncated: false }; + return { ...rest, replyRef: redactJson(replyRef), textBytes: 0, textTruncated: false, outputBytes: 0, outputTruncated: false }; } function normalizeToolCallPayload(payload: JsonRecord): JsonRecord { diff --git a/src/mgr/result.ts b/src/mgr/result.ts index 4f24ad1..dc20229 100644 --- a/src/mgr/result.ts +++ b/src/mgr/result.ts @@ -795,7 +795,7 @@ function resultFailureDetails(events: RunEvent[], terminal: TerminalStatus | nul function assistantReply(events: RunEvent[]): AssistantReplySummary { const assistantEvents = events.filter((event) => event.type === "assistant_message"); - const sealedReply = assistantReplyFromFinalSeal(assistantEvents); + const sealedReply = assistantReplyFromFinalSeal(events); if (sealedReply) return sealedReply; const latestAuthoritative = [...assistantEvents].reverse().find((event) => (event.payload.replyAuthority === true || event.payload.final === true) && textPayload(event.payload).length > 0); if (latestAuthoritative) return assistantReplySummary(latestAuthoritative); @@ -824,7 +824,7 @@ function assistantReplyFromFinalSeal(events: RunEvent[]): AssistantReplySummary } function assistantReplyRefMatches(event: RunEvent, replyRef: JsonRecord): boolean { - if (event.payload.finalSeal === true || textPayload(event.payload).length === 0) return false; + if (event.type !== "assistant_message" || event.payload.finalSeal === true || textPayload(event.payload).length === 0) return false; if (stringJsonValue(event.payload.source) !== stringJsonValue(replyRef.source)) return false; if ((stringJsonValue(event.payload.itemId) ?? null) !== (stringJsonValue(replyRef.itemId) ?? null)) return false; if ((numberJsonValue(event.payload.messageIndex) ?? null) !== (numberJsonValue(replyRef.messageIndex) ?? null)) return false; diff --git a/src/runner/run-once.ts b/src/runner/run-once.ts index a0372ce..145529e 100644 --- a/src/runner/run-once.ts +++ b/src/runner/run-once.ts @@ -716,7 +716,7 @@ async function writeTerminalOutbox(runnerLog: RunnerLogSink, payload: JsonRecord function shouldKeepTerminalOutboxEvent(event: BackendEvent): boolean { if (event.type === "terminal_status") return true; - if (event.type === "backend_status") return event.payload.phase === "turn/completed"; + if (event.type === "backend_status") return event.payload.phase === "turn/completed" || event.payload.finalSeal === true; if (event.type !== "assistant_message") return false; return event.payload.final === true || event.payload.replyAuthority === true; } diff --git a/src/selftest/cases/36-stdio-event-reconstruction.ts b/src/selftest/cases/36-stdio-event-reconstruction.ts index 9b71133..a11922f 100644 --- a/src/selftest/cases/36-stdio-event-reconstruction.ts +++ b/src/selftest/cases/36-stdio-event-reconstruction.ts @@ -36,10 +36,10 @@ const selfTest: SelfTestCase = async (context) => { const reconstructed = reconstructAgentRunEventsFromStdioFrames(frames, { sessionId, traceId }); assert.equal(reconstructed.selectedFrameCount, 8); assert.equal(reconstructed.reducerGroupCount, 1); - assert.equal(reconstructed.records.length, 10); + assert.equal(reconstructed.records.length, 9); assert.equal((reconstructed.ignoredFrames as JsonRecord)["direction-not-stdout"], 1); assert.equal((reconstructed.ignoredFrames as JsonRecord)["identity-mismatch"], 1); - assert.deepEqual(reconstructed.records.map((record) => record.value.sourceSeq), Array.from({ length: 10 }, (_, index) => index + 1)); + assert.deepEqual(reconstructed.records.map((record) => record.value.sourceSeq), Array.from({ length: 9 }, (_, index) => index + 1)); for (const record of reconstructed.records) { assert.equal(record.topic, DEFAULT_STDIO_RECONSTRUCTION_TOPIC); assert.equal(record.key, sessionId); @@ -57,10 +57,11 @@ const selfTest: SelfTestCase = async (context) => { const finalSeal = reconstructed.records.find((record) => { const event = record.value.event as JsonRecord; const payload = event.payload as JsonRecord; - return event.type === "assistant_message" && payload.replyAuthority === true; + return event.type === "backend_status" && payload.finalSeal === true; }); const finalSealPayload = (finalSeal?.value.event as JsonRecord).payload as JsonRecord; assert.equal(finalSealPayload.finalSeal, true); + assert.equal(finalSealPayload.phase, "assistant-message-final-sealed"); assert.equal(Object.hasOwn(finalSealPayload, "text"), false); const replyRef = finalSealPayload.replyRef as JsonRecord; const finalAssistant = reconstructed.records.find((record) => { @@ -221,7 +222,7 @@ async function assertOneShotPublisherOwnership(frames: ReturnType ({ records: frames, scanned: frames.length, invalidJson: 0, timedOut: false, elapsedMs: 0 }), }); assert.equal(producers.length, 2); - assert.deepEqual(producers[1], { connectCalls: 1, sendCalls: 10, disconnectCalls: 1 }); + assert.deepEqual(producers[1], { connectCalls: 1, sendCalls: 9, disconnectCalls: 1 }); assert.equal(producers[0]?.disconnectCalls, 0); await publishAgentRunKafkaMessage(managerConfig, managerMessage); diff --git a/src/selftest/cases/80-assistant-final-seal.ts b/src/selftest/cases/80-assistant-final-seal.ts index 0cfae8c..6daff94 100644 --- a/src/selftest/cases/80-assistant-final-seal.ts +++ b/src/selftest/cases/80-assistant-final-seal.ts @@ -31,6 +31,7 @@ const selfTest: SelfTestCase = async (context) => { try { const client = new ManagerClient(server.baseUrl); await assertHistoryReplaySealedOnce(client, context, server.baseUrl); + await assertAnonymousGenerationFence(client, context, server.baseUrl); await assertDeltaFallbackUsesBodylessSeal(client, context, server.baseUrl); return { name: "assistant-final-seal", @@ -40,6 +41,9 @@ const selfTest: SelfTestCase = async (context) => { "assistant-final-seal-late-final", "assistant-final-seal-manager-result-pointer", "assistant-final-seal-delta-fallback", + "assistant-progress-completed-item-fence", + "assistant-progress-anonymous-generation-fence", + "assistant-final-seal-control-event", ], }; } finally { @@ -47,6 +51,42 @@ const selfTest: SelfTestCase = async (context) => { } }; +async function assertAnonymousGenerationFence(client: ManagerClient, context: SelfTestContext, managerUrl: string): Promise { + const command = await createRunWithCommand(client, context, "anonymous assistant generation fence", "selftest-assistant-anonymous-generation-fence", 15_000); + const run = await runOnce({ + managerUrl, + runId: command.runId, + codexCommand: context.fakeCodexCommand, + codexArgs: context.fakeCodexArgs, + codexHome: context.codexHome, + env: { + CODEX_HOME: context.codexHome, + AGENTRUN_FAKE_CODEX_MODE: "assistant-anonymous-generation-fence", + }, + oneShot: true, + }) as JsonRecord; + assert.equal(run.terminalStatus, "completed"); + + const events = await listEvents(client, command.runId); + const completedMessages = events.filter((event) => event.type === "assistant_message" && event.payload.source === "completed-agent-message"); + assert.deepEqual(completedMessages.map((event) => event.payload.text), ["Anonymous first.", "Anonymous second final."], "new anonymous item/started must open a distinct legal generation"); + assert.ok(completedMessages.every((event) => event.payload.itemId === null)); + const progressEvents = events.filter((event) => event.type === "assistant_message" && event.payload.source === "agent-message-delta-progress"); + assert.equal(progressEvents.length, 1, "only the reopened anonymous generation's live progress must remain visible"); + assert.equal(progressEvents[0]?.payload.progressFlush, false); + assert.ok(progressEvents[0]!.seq < completedMessages[1]!.seq); + const suppression = events.find((event) => event.type === "backend_status" && event.payload.phase === "codex-app-server-notifications-suppressed"); + const replayDeltaMethod = asRecords(suppression?.payload.methods).find((item) => item.method === "item/agentMessage/delta:completed-anonymous-replay"); + assert.equal(replayDeltaMethod?.count, 1, "late anonymous delta must stay closed until the next anonymous item/started"); + const seal = events.find((event) => event.type === "backend_status" && event.payload.finalSeal === true); + assert.ok(seal); + assert.equal(asRecord(seal.payload.replyRef).itemId, null); + assert.equal(asRecord(seal.payload.replyRef).messageIndex, 2); + const result = await client.get(`/api/v1/runs/${command.runId}/commands/${command.commandId}/result`) as JsonRecord; + assert.equal(result.reply, "Anonymous second final."); + assert.equal(result.finalAssistantSeq, seal.seq); +} + async function assertHistoryReplaySealedOnce(client: ManagerClient, context: SelfTestContext, managerUrl: string): Promise { const lateFinalText = `Late unique final answer:${"x".repeat(5_000)} Authorization: Bearer REDACTED`; const rawLateFinalSecret = "final-secret-token-value"; @@ -78,14 +118,23 @@ async function assertHistoryReplaySealedOnce(client: ManagerClient, context: Sel for (const itemId of ["msg_replay_progress", "msg_same_text_a", "msg_same_text_b", "msg_late_final_seal"]) { assert.equal(completedMessages.filter((event) => event.payload.itemId === itemId).length, 1, `item ${itemId} must be durable exactly once`); } + const progressEvents = events.filter((event) => event.type === "assistant_message" && event.payload.source === "agent-message-delta-progress"); + assert.ok(progressEvents.some((event) => event.payload.itemId === "msg_late_final_seal" && event.payload.progressFlush === false), "long assistant deltas must remain visible before completion"); + assert.equal(progressEvents.some((event) => event.payload.progressFlush === true), false, "completed item delta buffers must not be replayed by finalize()"); + for (const progress of progressEvents) { + const completed = completedMessages.find((event) => event.payload.itemId === progress.payload.itemId); + assert.ok(completed && progress.seq < completed.seq, `item ${String(progress.payload.itemId)} progress must precede its durable completion`); + } + assert.equal(events.some((event) => event.type === "assistant_message" && typeof event.payload.text !== "string"), false, "bodyless control events must not be encoded as assistant messages"); const lateFinal = completedMessages.find((event) => event.payload.itemId === "msg_late_final_seal"); assert.equal(lateFinal?.payload.text, lateFinalText, "the unique durable message body must remain untruncated for seal dereferencing"); assert.equal(lateFinal?.payload.outputTruncated, false); assert.equal(JSON.stringify(events).includes(rawLateFinalSecret), false, "secret-like final text must be redacted before becoming durable"); - const seals = events.filter((event) => event.type === "assistant_message" && event.payload.finalSeal === true); + const seals = events.filter((event) => event.type === "backend_status" && event.payload.finalSeal === true); assert.equal(seals.length, 1, "one completed turn must have exactly one final seal"); const seal = seals[0]!; + assert.equal(seal.payload.phase, "assistant-message-final-sealed"); assert.equal(Object.hasOwn(seal.payload, "text"), false, "final seal must not duplicate the assistant body"); assert.equal(Object.hasOwn(seal.payload, "content"), false, "final seal must not duplicate the assistant body through content"); assert.equal(Object.hasOwn(seal.payload, "delta"), false, "final seal must not duplicate the assistant body through delta"); @@ -101,6 +150,8 @@ async function assertHistoryReplaySealedOnce(client: ManagerClient, context: Sel assert.ok(suppression, "history replay suppression must remain visible without persisting duplicate bodies"); const replayMethod = asRecords(suppression?.payload.methods).find((item) => item.method === "item/completed:duplicate-agentMessage"); assert.equal(replayMethod?.count, 5); + const replayDeltaMethod = asRecords(suppression?.payload.methods).find((item) => item.method === "item/agentMessage/delta:completed-item-replay"); + assert.equal(replayDeltaMethod?.count, 4); const toolEvents = events.filter((event) => event.type === "tool_call" && event.payload.itemId === "tool_before_history_replay"); assert.equal(toolEvents.length, 2, "assistant replay suppression must neither replay nor drop the tool start/completion pair"); @@ -137,8 +188,9 @@ async function assertDeltaFallbackUsesBodylessSeal(client: ManagerClient, contex const fallback = events.filter((event) => event.type === "assistant_message" && event.payload.source === "agent-message-delta-fallback"); assert.equal(fallback.length, 1, "delta-only turns must persist one full fallback body before sealing it"); assert.equal(fallback[0]?.payload.text, "fake codex stdio reply"); - const seal = events.find((event) => event.type === "assistant_message" && event.payload.finalSeal === true); + const seal = events.find((event) => event.type === "backend_status" && event.payload.finalSeal === true); assert.ok(seal); + assert.equal(seal.payload.phase, "assistant-message-final-sealed"); assert.equal(Object.hasOwn(seal.payload, "text"), false); assert.equal(asRecord(seal.payload.replyRef).source, "agent-message-delta-fallback"); diff --git a/src/selftest/fake-codex-app-server.ts b/src/selftest/fake-codex-app-server.ts index f1a582c..20d295e 100644 --- a/src/selftest/fake-codex-app-server.ts +++ b/src/selftest/fake-codex-app-server.ts @@ -242,29 +242,55 @@ for await (const line of rl) { if (mode === "assistant-history-replay-final-seal") { turnCounter += 1; const turn = { id: `turn_selftest_${turnCounter}`, status: "completed" }; + const replayText = "Progress before history replay."; + const sameText = "Legitimate identical assistant text."; const lateFinalText = `Late unique final answer:${"x".repeat(5_000)} Authorization: Bearer final-secret-token-value`; respond(message.id, { turn }); notify("turn/started", { turn }); notify("item/started", { item: { id: "tool_before_history_replay", type: "commandExecution", command: "echo durable-once", status: "running" } }); notify("item/completed", { item: { id: "tool_before_history_replay", type: "commandExecution", command: "echo durable-once", status: "completed", exitCode: 0 } }); - notify("item/completed", { item: { id: "msg_replay_progress", type: "agentMessage", text: "Progress before history replay." } }); - notify("item/completed", { item: { id: "msg_same_text_a", type: "agentMessage", text: "Legitimate identical assistant text." } }); - notify("item/completed", { item: { id: "msg_same_text_b", type: "agentMessage", text: "Legitimate identical assistant text." } }); - notify("item/completed", { item: { id: "msg_replay_progress", type: "agentMessage", text: "Progress before history replay." } }); + notify("item/agentMessage/delta", { itemId: "msg_replay_progress", delta: replayText }); + notify("item/completed", { item: { id: "msg_replay_progress", type: "agentMessage", text: replayText } }); + notify("item/agentMessage/delta", { itemId: "msg_same_text_a", delta: sameText }); + notify("item/completed", { item: { id: "msg_same_text_a", type: "agentMessage", text: sameText } }); + notify("item/agentMessage/delta", { itemId: "msg_same_text_b", delta: sameText }); + notify("item/completed", { item: { id: "msg_same_text_b", type: "agentMessage", text: sameText } }); + notify("item/completed", { item: { id: "msg_replay_progress", type: "agentMessage", text: replayText } }); notify("turn/completed", { turn }); setTimeout(() => { - notify("item/completed", { item: { id: "msg_replay_progress", type: "agentMessage", text: "Progress before history replay." } }); - notify("item/completed", { item: { id: "msg_same_text_a", type: "agentMessage", text: "Legitimate identical assistant text." } }); - notify("item/completed", { item: { id: "msg_same_text_b", type: "agentMessage", text: "Legitimate identical assistant text." } }); + notify("item/agentMessage/delta", { itemId: "msg_replay_progress", delta: replayText }); + notify("item/completed", { item: { id: "msg_replay_progress", type: "agentMessage", text: replayText } }); + notify("item/agentMessage/delta", { itemId: "msg_same_text_a", delta: sameText }); + notify("item/completed", { item: { id: "msg_same_text_a", type: "agentMessage", text: sameText } }); + notify("item/agentMessage/delta", { itemId: "msg_same_text_b", delta: sameText }); + notify("item/completed", { item: { id: "msg_same_text_b", type: "agentMessage", text: sameText } }); }, 20).unref?.(); setTimeout(() => { + notify("item/agentMessage/delta", { itemId: "msg_late_final_seal", delta: lateFinalText }); notify("item/completed", { item: { id: "msg_late_final_seal", type: "agentMessage", text: lateFinalText } }); }, 40).unref?.(); setTimeout(() => { + notify("item/agentMessage/delta", { itemId: "msg_late_final_seal", delta: lateFinalText }); notify("item/completed", { item: { id: "msg_late_final_seal", type: "agentMessage", text: lateFinalText } }); }, 60).unref?.(); continue; } + if (mode === "assistant-anonymous-generation-fence") { + turnCounter += 1; + const turn = { id: `turn_selftest_${turnCounter}`, status: "completed" }; + const secondProgress = `Anonymous second progress:${"y".repeat(600)}`; + respond(message.id, { turn }); + notify("turn/started", { turn }); + notify("item/started", { item: { type: "agentMessage" } }); + notify("item/agentMessage/delta", { delta: "Anonymous first." }); + notify("item/completed", { item: { type: "agentMessage", text: "Anonymous first." } }); + notify("item/agentMessage/delta", { delta: "Anonymous first." }); + notify("item/started", { item: { type: "agentMessage" } }); + notify("item/agentMessage/delta", { delta: secondProgress }); + notify("item/completed", { item: { type: "agentMessage", text: "Anonymous second final." } }); + notify("turn/completed", { turn }); + continue; + } if (mode === "web-search-progress") { turnCounter += 1; const turn = { id: `turn_selftest_${turnCounter}`, status: "completed" };