From a49f322b326430338ce640f2d3a071af2c1b0c9f Mon Sep 17 00:00:00 2001 From: "(CJ) Chukwudi Nwobodo" <142016413+chuks-qua@users.noreply.github.com> Date: Tue, 6 Oct 2026 10:56:42 +0100 Subject: [PATCH 1/2] fix(verify-mcode): read live turn events from agent.canonical Since #1779 the server publishes live turn events only on the agent.canonical push channel. The runtime live harness and the composer-queue proof still listened for agent.event, so every runtime live scenario timed out waiting for a terminal event even when the assistant reply was saved. canonicalFrameToLiveEvents reads the publication.recorded payloads of accepted, saved, and recovery frames and applies the web client's publication route. Callers record each event ID once because saved and recovery frames repeat accepted events. A stopped turn now publishes only ended with an interrupted outcome, so ended counts as completion only with a completed outcome and counts as a stop otherwise. --- .../references/features/composer-queue.md | 2 +- .../provider-events-and-durability.md | 2 +- .../verify-mcode/scripts/composer-queue.mjs | 32 +++-- .../scripts/composer-queue.test.mjs | 14 ++- .../skills/verify-mcode/scripts/runtime.mjs | 112 ++++++++++++++---- .../verify-mcode/scripts/runtime.test.mjs | 81 ++++++++++++- 6 files changed, 207 insertions(+), 36 deletions(-) diff --git a/.agents/skills/verify-mcode/references/features/composer-queue.md b/.agents/skills/verify-mcode/references/features/composer-queue.md index 77c8ec06a..fdf37ee32 100644 --- a/.agents/skills/verify-mcode/references/features/composer-queue.md +++ b/.agents/skills/verify-mcode/references/features/composer-queue.md @@ -33,7 +33,7 @@ For each provider, the Electron proof selects and rechecks the provider and mode The receipt records safe predicate counts and booleans. Live admission requires exact durable root, A, B, and C prompt identities, exact visible queue rows, and a visible Stop control. It does not require provider start events. Duplicate durable A or B prompts, incorrect queue rows, or an idle composer fail the proof. The receipt retains provider event counts as diagnostics. The deterministic verifier tests retain Codex and Cursor terminal-order checks as supporting adapter coverage. -The proof opens its Electron-local socket before it reads Cursor settings or checks provider readiness. It uses that socket for the temporary Cursor setting, provider and model checks, queue RPCs, and event replay. Once the UI returns the new thread ID, it replays that thread from cursor zero on the same socket. `turnComplete` is a completed terminal. `ended` is completed only when its outcome is `completed` or `success`. The receipt omits event payloads and execution IDs. When the verifier temporarily enables Cursor, the receipt records only that the original state was disabled and that it restored it before owned resource cleanup. +The proof opens its Electron-local socket before it reads Cursor settings or checks provider readiness. It uses that socket for the temporary Cursor setting, provider and model checks, queue RPCs, and event replay. Once the UI returns the new thread ID, it replays that thread from cursor zero on the same socket and reads provider events from `agent.canonical` publications. `turnComplete` is a completed terminal. `ended` is completed only when its outcome is `completed` or `success`. The receipt omits event payloads and execution IDs. When the verifier temporarily enables Cursor, the receipt records only that the original state was disabled and that it restored it before owned resource cleanup. Run the verifier's own deterministic check before proof collection: diff --git a/.agents/skills/verify-mcode/references/features/provider-events-and-durability.md b/.agents/skills/verify-mcode/references/features/provider-events-and-durability.md index 8345b9284..8a031ba30 100644 --- a/.agents/skills/verify-mcode/references/features/provider-events-and-durability.md +++ b/.agents/skills/verify-mcode/references/features/provider-events-and-durability.md @@ -186,7 +186,7 @@ Run `runtime health`, then run: bun .agents/skills/verify-mcode/scripts/verify-mcode.mjs runtime live --provider codex --model --scenario completion --confirm-provider-call ``` -The harness requires `turnComplete` or `ended`. It fails immediately when the target thread emits `error`, `errored`, `cancelled`, `interrupted`, or `paused` first. It then reads `conversation.page` and `message.list` until both return a durable assistant message. The receipt omits assistant text and provider-private payloads. +The harness reads provider events from the `publication.recorded` payloads of `agent.canonical` frames. The `agent.event` channel no longer exists. Saved and recovery frames repeat accepted events, so it records each event ID once. It requires `turnComplete`, or `ended` with a completed outcome. It fails immediately when the target thread first emits `error`, an `ended` outcome other than completed, or an `errored`, `cancelled`, `interrupted`, or `paused` status. It then reads `conversation.page` and `message.list` until both return a durable assistant message. The receipt omits assistant text and provider-private payloads. ### Codex notice triggers and desktop proof diff --git a/.agents/skills/verify-mcode/scripts/composer-queue.mjs b/.agents/skills/verify-mcode/scripts/composer-queue.mjs index fd2c4787a..107047bd1 100644 --- a/.agents/skills/verify-mcode/scripts/composer-queue.mjs +++ b/.agents/skills/verify-mcode/scripts/composer-queue.mjs @@ -13,7 +13,7 @@ import { openDesktopSocket, verifyThreadLifecycleHealth, } from "./thread-lifecycle.mjs"; -import { openRuntimeVerificationSocket } from "./runtime.mjs"; +import { canonicalFrameToLiveEvents, openRuntimeVerificationSocket } from "./runtime.mjs"; import { stopElectron } from "../../../../.agents/skills/electorn-live-testing/scripts/stop-electron.mjs"; const EVIDENCE_DIRECTORY = ".dev/verification/composer-queue"; @@ -1087,18 +1087,26 @@ async function subscribeQueueProviderEvents(socket, threadId, evidence) { evidence.subscribed = true; } +// Evidence keeps the redacted `agent.event` record shape that the queue checks +// read; the wire source is now the `agent.canonical` publication stream. function recordQueueProviderPush(evidence, record, push) { - if (push?.channel !== "agent.event" || push.data?.threadId !== record.threadId || typeof push.data.type !== "string") return; - evidence.events.push({ - channel: "agent.event", - data: { - outcome: safeEventOutcome(push.data.outcome), - threadId: record.threadId, - turnExecutionId: safeTurnExecutionId(push.data.turnExecutionId), - type: push.data.type, - }, - }); - if (evidence.events.length > 40) evidence.events.shift(); + if (push?.channel !== "agent.canonical" || push.data?.threadId !== record.threadId) return; + evidence.seenEventIds ??= new Set(); + for (const event of canonicalFrameToLiveEvents(push.data)) { + // Saved and recovery frames repeat accepted events. + if (evidence.seenEventIds.has(event.eventId)) continue; + evidence.seenEventIds.add(event.eventId); + evidence.events.push({ + channel: "agent.event", + data: { + outcome: safeEventOutcome(event.outcome), + threadId: record.threadId, + turnExecutionId: safeTurnExecutionId(event.turnExecutionId), + type: event.type, + }, + }); + if (evidence.events.length > 40) evidence.events.shift(); + } } function safeEventOutcome(value) { diff --git a/.agents/skills/verify-mcode/scripts/composer-queue.test.mjs b/.agents/skills/verify-mcode/scripts/composer-queue.test.mjs index 1dcccde52..62dce144a 100644 --- a/.agents/skills/verify-mcode/scripts/composer-queue.test.mjs +++ b/.agents/skills/verify-mcode/scripts/composer-queue.test.mjs @@ -1378,9 +1378,21 @@ test("uses one Electron-local socket for Composer queue RPCs and provider events expect(() => requireSameQueueSocket(uiSocket, worktreeSocket)).toThrow("same Electron-local socket"); const evidence = { events: [] }; let opened = 0; + const turnStarted = { + eventId: "publication:thread-owned:1", + routing: { threadId: "thread-owned", turnId: "turn-1", executionId: "execution-1" }, + sourceProviderId: "codex", + sourceIdentities: [], + acceptedSequence: 1, + serverTimestamps: { acceptedAt: "2026-10-06T09:37:20.543Z" }, + progressPosition: { epoch: "epoch-1", sequence: 1 }, + payload: { type: "publication.recorded", publicationId: "1", event: { type: "turnStarted", threadId: "thread-owned", publicationId: "1" } }, + }; const socket = await openQueueSocket(async (onPush) => { opened += 1; - onPush({ channel: "agent.event", data: { threadId: "thread-owned", type: "turnStarted" } }); + onPush({ channel: "agent.canonical", data: { phase: "accepted", threadId: "thread-owned", epoch: "epoch-1", from: 0, through: 1, events: [turnStarted] } }); + onPush({ channel: "agent.canonical", data: { phase: "saved", threadId: "thread-owned", epoch: "epoch-1", through: 1, + revision: { conversationRevision: 1, rosterRevision: 0 }, events: [{ ...turnStarted, durableRevision: 1 }] } }); return uiSocket; }, ownedRecord(), evidence); expect(socket).toBe(uiSocket); diff --git a/.agents/skills/verify-mcode/scripts/runtime.mjs b/.agents/skills/verify-mcode/scripts/runtime.mjs index eb4979761..0e154ce8f 100644 --- a/.agents/skills/verify-mcode/scripts/runtime.mjs +++ b/.agents/skills/verify-mcode/scripts/runtime.mjs @@ -526,6 +526,7 @@ async function live(repoRoot, options) { proofDeadline: null, ownedThreadIds: [], eventsByThread: new Map(), + seenCanonicalEventIds: new Set(), }; try { await prepareLiveRun(repoRoot, options, run); @@ -1728,13 +1729,25 @@ async function requireTerminalEvent(report, deadline) { function findCompletionOutcome(events) { for (const event of events) { - if (event.kind === "agent" && ["turnComplete", "ended"].includes(event.type)) { - return { succeeded: true, label: event.type }; - } - if (event.kind === "agent" && event.type === "error") return { succeeded: false, label: "error" }; - if (event.kind === "status" && ["errored", "cancelled", "interrupted", "paused"].includes(event.status)) { - return { succeeded: false, label: event.status }; - } + const outcome = event.kind === "agent" ? agentEventOutcome(event) : statusEventOutcome(event); + if (outcome) return outcome; + } + return null; +} + +function agentEventOutcome(event) { + if (event.type === "turnComplete") return { succeeded: true, label: event.type }; + // A stopped turn publishes only `ended`, so its outcome decides success. + if (event.type === "ended") { + return isCompletedOutcome(event.outcome) ? { succeeded: true, label: event.type } : { succeeded: false, label: event.outcome }; + } + if (event.type === "error") return { succeeded: false, label: "error" }; + return null; +} + +function statusEventOutcome(event) { + if (event.kind === "status" && ["errored", "cancelled", "interrupted", "paused"].includes(event.status)) { + return { succeeded: false, label: event.status }; } return null; } @@ -1780,7 +1793,12 @@ async function requireStoppedOutcome(report, deadline) { } function hasStoppedStatus(events) { - return events.some((event) => event.kind === "status" && ["paused", "interrupted", "cancelled"].includes(event.status)); + return events.some((event) => (event.kind === "status" && ["paused", "interrupted", "cancelled"].includes(event.status)) + || (event.kind === "agent" && event.type === "ended" && ["interrupted", "cancelled"].includes(event.outcome))); +} + +function isCompletedOutcome(outcome) { + return outcome === undefined || outcome === "completed"; } function hasSharedStopResult(stopResults, threadId) { @@ -1939,23 +1957,77 @@ function recordPush(run, push) { if (typeof threadId !== "string" || !run.ownedThreadIds.includes(threadId)) return; const events = run.eventsByThread.get(threadId); if (!events) return; - const event = liveEventFromPush(push); - if (!event) return; - events.push(event); - if (!includeInReceipt(run, threadId, event)) return; - run.report.events.push(event); - if (run.report.events.length > 400) run.report.events.splice(0, run.report.events.length - 400); + for (const event of liveEventsFromPush(run, push)) { + events.push(event); + if (!includeInReceipt(run, threadId, event)) continue; + run.report.events.push(event); + if (run.report.events.length > 400) run.report.events.splice(0, run.report.events.length - 400); + } } -function liveEventFromPush(push) { - if (push.channel === "agent.event" && typeof push.data.type === "string") { - return { +function liveEventsFromPush(run, push) { + if (push.channel === "agent.canonical") { + // Saved and recovery frames repeat accepted events, and a restarted runtime + // replays them from cursor 0, so one event ID records only once per run. + const unseen = canonicalFrameToLiveEvents(push.data).filter((event) => !run.seenCanonicalEventIds.has(event.eventId)); + for (const event of unseen) run.seenCanonicalEventIds.add(event.eventId); + return unseen.map((event) => ({ kind: "agent", - type: push.data.type, - subtype: push.data.subtype === OPENCODE_SESSION_INVALIDATED_SUBTYPE ? push.data.subtype : undefined, + type: event.type, + subtype: event.subtype === OPENCODE_SESSION_INVALIDATED_SUBTYPE ? event.subtype : undefined, + outcome: event.outcome, elapsedMs: Date.now(), - }; + })); } + const event = liveEventFromPush(push); + return event ? [event] : []; +} + +/** + * Reads the renderer-facing agent events that one `agent.canonical` frame publishes. + * Each record keeps the envelope's event ID so callers can drop the repeats that + * saved and recovery frames carry. Frames mirrored to a child subscriber + * (`ownerThreadId` set) and publications routed to another thread or execution + * are skipped, matching the web client's publication route. + */ +export function canonicalFrameToLiveEvents(frame) { + if (!frame || typeof frame !== "object" || frame.ownerThreadId !== undefined) return []; + return canonicalFrameEnvelopes(frame).flatMap((envelope) => { + const event = routedPublication(envelope, frame.threadId); + if (!event) return []; + return [{ + eventId: envelope.eventId, + threadId: event.threadId, + type: event.type, + ...optionalString("subtype", event.subtype), + ...optionalString("outcome", event.outcome), + ...optionalString("turnExecutionId", event.turnExecutionId), + }]; + }); +} + +function canonicalFrameEnvelopes(frame) { + if (frame.phase !== "recovery") return frame.events ?? []; + const durable = frame.durable?.mode === "delta" ? frame.durable.events : []; + return [...durable, ...(frame.retained ?? [])]; +} + +function routedPublication(envelope, threadId) { + const event = envelope?.payload?.type === "publication.recorded" ? envelope.payload.event : null; + if (typeof event?.type !== "string" || event.threadId !== threadId) return null; + return publicationMatchesRoute(event, envelope.routing, threadId) ? event : null; +} + +function publicationMatchesRoute(event, routing, threadId) { + return routing?.threadId === threadId + && (event.turnExecutionId === undefined || event.turnExecutionId === routing.executionId); +} + +function optionalString(key, value) { + return typeof value === "string" ? { [key]: value } : {}; +} + +function liveEventFromPush(push) { if (push.channel === "thread.status" && typeof push.data.status === "string") { return { kind: "status", status: push.data.status, elapsedMs: Date.now() }; } diff --git a/.agents/skills/verify-mcode/scripts/runtime.test.mjs b/.agents/skills/verify-mcode/scripts/runtime.test.mjs index 2a6e03de0..e3fa1516f 100644 --- a/.agents/skills/verify-mcode/scripts/runtime.test.mjs +++ b/.agents/skills/verify-mcode/scripts/runtime.test.mjs @@ -6,7 +6,7 @@ import * as NodeOS from "node:os"; import * as NodePath from "node:path"; import * as NodeTest from "node:test"; -import { assertRuntimeFreshness, isOpenCodeSessionInvalidatedEvent, isRuntimeHarnessEvidenceFile, openVerificationSocketUrl, runBun } from "./runtime.mjs"; +import { assertRuntimeFreshness, canonicalFrameToLiveEvents, isOpenCodeSessionInvalidatedEvent, isRuntimeHarnessEvidenceFile, openVerificationSocketUrl, runBun } from "./runtime.mjs"; const CLI = NodePath.join(import.meta.dirname, "verify-mcode.mjs"); const BROWSER_PROOF = NodePath.join(import.meta.dirname, "browser-opencode-proof.mjs"); @@ -145,6 +145,85 @@ NodeTest.test("recognizes the provider-neutral OpenCode session invalidation sub NodeAssertStrict.equal(isOpenCodeSessionInvalidatedEvent({ type: "system", subtype: "opencode:session-recreated" }), false); }); +const THREAD = "e2487eb3-3152-4bfb-b466-1aec64ef7ec5"; +const CHILD = "7c0f2d64-1b5e-4c55-9d0b-3f5a1f0e9b21"; +const EXECUTION = "660e1a9f-f474-4e6e-8c13-ac64abd4dace"; +const EPOCH = "5d3d286f-a6d0-4cdc-b734-2d9fb26eb0d3:a9eafe97-3bdc-4701-ab55-345856cdf877"; + +// Mirrors one `AcceptedCanonicalAgentEventEnvelopeSchema` envelope captured from a live Claude turn. +function canonicalEnvelope(eventId, sequence, payload, routingThreadId = THREAD) { + return { + eventId, + routing: { threadId: routingThreadId, turnId: "1fe19a18-e38b-4696-853d-8809a04e4a9e", executionId: EXECUTION }, + sourceProviderId: "claude", + sourceIdentities: [], + acceptedSequence: sequence, + serverTimestamps: { acceptedAt: "2026-10-06T09:37:20.543Z" }, + payload, + progressPosition: { epoch: EPOCH, sequence }, + }; +} + +function publication(eventId, sequence, event, routingThreadId) { + return canonicalEnvelope(eventId, sequence, { type: "publication.recorded", publicationId: String(sequence), event: { ...event, publicationId: String(sequence) } }, routingThreadId); +} + +NodeTest.test("reads terminal publications from an accepted canonical frame and ignores semantic payloads", () => { + const frame = { + phase: "accepted", threadId: THREAD, epoch: EPOCH, from: 26, through: 28, + events: [ + canonicalEnvelope("e:27", 27, { type: "turn.completed", endedAt: "2026-10-06T09:37:20.543Z" }), + publication("e:28", 28, { type: "turnComplete", threadId: THREAD, reason: "end_turn", providerId: "claude", turnExecutionId: EXECUTION }), + ], + }; + + NodeAssertStrict.deepEqual(canonicalFrameToLiveEvents(frame), [ + { eventId: "e:28", threadId: THREAD, type: "turnComplete", turnExecutionId: EXECUTION }, + ]); +}); + +NodeTest.test("reads stop outcomes and session notices from saved frames", () => { + const frame = { + phase: "saved", threadId: THREAD, epoch: EPOCH, through: 9, + revision: { conversationRevision: 9, rosterRevision: 0 }, + events: [ + { ...publication("e:8", 8, { type: "system", threadId: THREAD, subtype: "sdk_session_invalidated" }), durableRevision: 8 }, + { ...publication("e:9", 9, { type: "ended", threadId: THREAD, turnExecutionId: EXECUTION, outcome: "interrupted" }), durableRevision: 9 }, + ], + }; + + NodeAssertStrict.deepEqual(canonicalFrameToLiveEvents(frame), [ + { eventId: "e:8", threadId: THREAD, type: "system", subtype: "sdk_session_invalidated" }, + { eventId: "e:9", threadId: THREAD, type: "ended", outcome: "interrupted", turnExecutionId: EXECUTION }, + ]); +}); + +NodeTest.test("reads a recovery frame's durable delta before its retained suffix", () => { + const frame = { + phase: "recovery", threadId: THREAD, epoch: EPOCH, acceptedThrough: 2, savedThrough: 1, loss: "none", + durable: { + mode: "delta", threadId: THREAD, + from: { conversationRevision: 0, rosterRevision: 0 }, through: { conversationRevision: 1, rosterRevision: 0 }, + events: [{ ...publication("e:1", 1, { type: "turnStarted", threadId: THREAD, turnExecutionId: EXECUTION }), durableRevision: 1 }], + }, + retained: [publication("e:2", 2, { type: "turnComplete", threadId: THREAD, turnExecutionId: EXECUTION })], + }; + + NodeAssertStrict.deepEqual(canonicalFrameToLiveEvents(frame).map((event) => event.type), ["turnStarted", "turnComplete"]); +}); + +NodeTest.test("skips mirrored child frames and publications routed to another thread or execution", () => { + const events = [ + publication("e:1", 1, { type: "turnComplete", threadId: THREAD, turnExecutionId: EXECUTION }), + publication("e:2", 2, { type: "turnComplete", threadId: CHILD, turnExecutionId: EXECUTION }), + publication("e:3", 3, { type: "turnComplete", threadId: THREAD, turnExecutionId: "0b6b2c7e-8f1d-4c3a-9e5f-2a7d4c1b8e90" }), + ]; + const owned = { phase: "accepted", threadId: THREAD, epoch: EPOCH, from: 0, through: 3, events }; + + NodeAssertStrict.deepEqual(canonicalFrameToLiveEvents(owned).map((event) => event.eventId), ["e:1"]); + NodeAssertStrict.deepEqual(canonicalFrameToLiveEvents({ ...owned, threadId: CHILD, ownerThreadId: THREAD }), []); +}); + NodeTest.test("rejects desktop verification sockets without loopback authentication", async () => { await NodeAssertStrict.rejects(openVerificationSocketUrl(process.cwd(), "ws://example.test/?token=token"), /loopback WebSocket URL/); await NodeAssertStrict.rejects(openVerificationSocketUrl(process.cwd(), "ws://localhost/"), /lacks its authentication token/); From b1bebdba75d0f0466b598a4353868a67851bba7a Mon Sep 17 00:00:00 2001 From: "(CJ) Chukwudi Nwobodo" <142016413+chuks-qua@users.noreply.github.com> Date: Tue, 6 Oct 2026 10:56:53 +0100 Subject: [PATCH 2/2] fix(verify-mcode): wait for runtime readiness after the OpenCode resume restart agent:up returns after it writes ports.json, before the server answers health. The resume proof called health with a 15-second budget right after agent:up and failed when startup took longer. On a loaded host the server took up to 15.9 seconds to open its listener, Vite once took 19.9 seconds, and one restart also failed the agent:ready CLI's fixed 30-second budget. Call agentReady in-process, bounded by the proof deadline. --- .agents/skills/verify-mcode/scripts/runtime.mjs | 5 +++++ 1 file changed, 5 insertions(+) diff --git a/.agents/skills/verify-mcode/scripts/runtime.mjs b/.agents/skills/verify-mcode/scripts/runtime.mjs index 0e154ce8f..5703f4278 100644 --- a/.agents/skills/verify-mcode/scripts/runtime.mjs +++ b/.agents/skills/verify-mcode/scripts/runtime.mjs @@ -1673,6 +1673,11 @@ async function restartOpenCodeResumeRuntime(repoRoot, run) { run.socket = null; await runWorktreeRuntimeCommand(repoRoot, "agent:down"); await runWorktreeRuntimeCommand(repoRoot, "agent:up"); + // agent:up returns once it writes ports.json, before the server answers health. + // A loaded host has measured past agent:ready's fixed 30-second CLI budget, so + // readiness shares the proof deadline instead. + const { agentReady } = await import("../../../../scripts/agent/agent-ready.mjs"); + await agentReady(repoRoot, { timeoutMs: Math.max(HEALTH_TIMEOUT_MS, run.proofDeadline - Date.now()) }); await health(repoRoot); run.socket = await openSocket(repoRoot, readRuntime(repoRoot), (push) => recordPush(run, push)); const subscription = await run.socket.rpc("push.setThreadSubscriptions", {