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..5703f4278 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); @@ -1672,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", { @@ -1728,13 +1734,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 +1798,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 +1962,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/);