diff --git a/apps/server/src/application/bootstrap/server-bootstrap.ts b/apps/server/src/application/bootstrap/server-bootstrap.ts index f9855b707..39ca1f8b4 100644 --- a/apps/server/src/application/bootstrap/server-bootstrap.ts +++ b/apps/server/src/application/bootstrap/server-bootstrap.ts @@ -866,13 +866,8 @@ async function bootstrapServer(): Promise { runtime: container.resolve(AgentEventPublicationRuntimePort), publicationRegistry: container.resolve(AgentEventPublicationRegistry), threadRepo, - narrativeStore, pullRequestCompletionEffect, providerRegistry, - publishAgentEvent: (event) => { - const sequencedEvent = broadcast("agent.event", event) ?? event; - portPush.send("agent.event", sequencedEvent); - }, publishThreadStatus: (status) => { broadcast("thread.status", status); portPush.send("thread.status", status); diff --git a/apps/server/src/application/transport/__tests__/port-push.test.ts b/apps/server/src/application/transport/__tests__/port-push.test.ts index c0879bbdd..6b4e4b3a2 100644 --- a/apps/server/src/application/transport/__tests__/port-push.test.ts +++ b/apps/server/src/application/transport/__tests__/port-push.test.ts @@ -19,7 +19,7 @@ describe("PortPush", () => { it("does nothing when no port is attached", () => { // Should not throw - portPush.send("agent.event", { type: "delta", threadId: "t1" }); + portPush.send("agent.canonical", { type: "delta", threadId: "t1" }); }); it("sends structured clone message via attached port", () => { @@ -27,10 +27,10 @@ describe("PortPush", () => { portPush.attach(port); const data = { type: "delta", threadId: "t1", delta: "hello" }; - portPush.send("agent.event", data); + portPush.send("agent.canonical", data); expect(port.postMessage).toHaveBeenCalledWith({ - channel: "agent.event", + channel: "agent.canonical", data, }); }); @@ -40,7 +40,7 @@ describe("PortPush", () => { portPush.attach(port); portPush.detach(); - portPush.send("agent.event", { type: "delta", threadId: "t1" }); + portPush.send("agent.canonical", { type: "delta", threadId: "t1" }); expect(port.postMessage).toHaveBeenCalledTimes(0); }); @@ -75,7 +75,7 @@ describe("PortPush", () => { expect(portPush.isActive).toBe(true); // New port receives messages, old one does not - portPush.send("agent.event", { type: "delta" }); + portPush.send("agent.canonical", { type: "delta" }); expect(port2.postMessage).toHaveBeenCalledOnce(); expect(port1.postMessage).not.toHaveBeenCalled(); }); @@ -97,14 +97,14 @@ describe("PortPush", () => { }); portPush.attach(port); - portPush.send("agent.event", { type: "delta" }); + portPush.send("agent.canonical", { type: "delta" }); // Port should have been closed and detached expect(port.close).toHaveBeenCalled(); expect(portPush.isActive).toBe(false); // Subsequent sends are no-ops - portPush.send("agent.event", { type: "delta" }); + portPush.send("agent.canonical", { type: "delta" }); expect(port.postMessage).toHaveBeenCalledOnce(); // only the failed call }); }); diff --git a/apps/server/src/application/transport/__tests__/push.test.ts b/apps/server/src/application/transport/__tests__/push.test.ts index ee37655eb..0349efd24 100644 --- a/apps/server/src/application/transport/__tests__/push.test.ts +++ b/apps/server/src/application/transport/__tests__/push.test.ts @@ -11,10 +11,9 @@ import { setClientThreadSubscriptions, subscribeClientToThread, unsubscribeClientFromThread, - MAX_AGENT_EVENT_JOURNAL_THREADS, _resetForTest, } from "../push.js"; -import { AgentEventSchema, decodeTerminalDataFrame } from "@mcode/contracts"; +import { decodeTerminalDataFrame } from "@mcode/contracts"; import { createPassThroughTransportPayloadValidator, createValidatingTransportPayloadValidator, @@ -22,6 +21,29 @@ import { setTransportPayloadValidatorForTest, } from "../payload-validation.js"; +function canonicalBatch(threadId: string) { + return { + threadId, + events: [{ + eventId: `event-${threadId}`, + routing: { + threadId, + turnId: `turn-${threadId}`, + executionId: "00000000-0000-4000-8000-000000000001", + }, + sourceProviderId: "codex", + sourceIdentities: [], + acceptedSequence: 1, + durableRevision: 1, + serverTimestamps: { + acceptedAt: "2026-08-09T20:00:00.000Z", + persistedAt: "2026-08-09T20:00:00.000Z", + }, + payload: { type: "turn.started", startedAt: "2026-08-09T20:00:00.000Z" }, + }], + }; +} + function fakeOpenSocket(received: Array<{ buf: Buffer; binary: boolean }>, completeImmediately = true): WebSocket { const ws: Partial = { readyState: 1, // OPEN @@ -128,22 +150,6 @@ describe("broadcast", () => { resetTransportPayloadValidatorForTest(); }); - it("keeps the durable publication identity while assigning a process sequence", () => { - const received: Array<{ buf: Buffer; binary: boolean }> = []; - const socket = fakeOpenSocket(received); - addClient(socket); - subscribeClientToThread(socket, "thread-a"); - const publicationId = "3"; - const first = broadcast("agent.event", { type: "turnStarted", threadId: "thread-a", - turnExecutionId: "00000000-0000-4000-8000-000000000001", publicationId }); - const replay = broadcast("agent.event", { type: "turnStarted", threadId: "thread-a", - turnExecutionId: "00000000-0000-4000-8000-000000000001", publicationId }); - expect(first).toMatchObject({ publicationId, sequence: 1 }); - expect(replay).toMatchObject({ publicationId, sequence: 2 }); - expect(received.map(({ buf }) => JSON.parse(buf.toString("utf-8")).data.publicationId)) - .toEqual([publicationId, publicationId]); - }); - it("routes thread-scoped events only to clients subscribed to that thread", () => { const a: Array<{ buf: Buffer; binary: boolean }> = []; const b: Array<{ buf: Buffer; binary: boolean }> = []; @@ -154,11 +160,7 @@ describe("broadcast", () => { subscribeClientToThread(wsA, "thread-a"); subscribeClientToThread(wsB, "thread-b"); - broadcast("agent.event", { - type: "textDelta", - threadId: "thread-a", - delta: "hello", - }); + broadcast("agent.canonical", canonicalBatch("thread-a")); expect(a).toHaveLength(1); expect(b).toHaveLength(0); @@ -175,26 +177,7 @@ describe("broadcast", () => { subscribeClientToThread(wsA, "thread-a"); subscribeClientToThread(wsB, "thread-b"); - broadcast("agent.canonical", { - threadId: "thread-a", - events: [{ - eventId: "event-a", - routing: { - threadId: "thread-a", - turnId: "turn-a", - executionId: "00000000-0000-4000-8000-000000000001", - }, - sourceProviderId: "codex", - sourceIdentities: [], - acceptedSequence: 1, - durableRevision: 1, - serverTimestamps: { - acceptedAt: "2026-08-09T20:00:00.000Z", - persistedAt: "2026-08-09T20:00:00.000Z", - }, - payload: { type: "turn.started", startedAt: "2026-08-09T20:00:00.000Z" }, - }], - }); + broadcast("agent.canonical", canonicalBatch("thread-a")); expect(a).toHaveLength(1); expect(b).toHaveLength(0); @@ -227,11 +210,7 @@ describe("broadcast", () => { subscribeClientToThread(ws, "thread-a"); unsubscribeClientFromThread(ws, "thread-a"); - broadcast("agent.event", { - type: "textDelta", - threadId: "thread-a", - delta: "hello", - }); + broadcast("agent.canonical", canonicalBatch("thread-a")); expect(received).toHaveLength(0); }); @@ -244,113 +223,13 @@ describe("broadcast", () => { setClientThreadSubscriptions(ws, ["thread-new"]); - broadcast("agent.event", { - type: "textDelta", - threadId: "thread-old", - delta: "old", - }); - broadcast("agent.event", { - type: "textDelta", - threadId: "thread-new", - delta: "new", - }); + broadcast("agent.canonical", canonicalBatch("thread-old")); + broadcast("agent.canonical", canonicalBatch("thread-new")); expect(received).toHaveLength(1); expect(JSON.parse(received[0].buf.toString("utf-8")).data.threadId).toBe("thread-new"); }); - it("assigns ordered sequences and replays retained events before live delivery", () => { - const received: Array<{ buf: Buffer; binary: boolean }> = []; - broadcast("agent.event", { type: "textDelta", threadId: "thread-replay", delta: "one" }); - broadcast("agent.event", { type: "textDelta", threadId: "thread-replay", delta: "two" }); - const ws = fakeOpenSocket(received); - addClient(ws); - - const result = setClientThreadSubscriptions(ws, ["thread-replay"], { "thread-replay": 0 }); - expect(result).toEqual({ hydrationRequiredThreadIds: [], replayedThrough: { "thread-replay": 2 } }); - broadcast("agent.event", { type: "textDelta", threadId: "thread-replay", delta: "three" }); - - expect(received.map((entry) => JSON.parse(entry.buf.toString("utf-8")).data.sequence)).toEqual([1, 2, 3]); - }); - - it("keeps an evicted thread's next event distinguishable from its old cursor", () => { - const first = AgentEventSchema().parse(broadcast("agent.event", { type: "turnStarted", threadId: "evicted-thread" })); - if (!first.epoch) throw new Error("Agent event has no journal epoch"); - for (let index = 0; index < MAX_AGENT_EVENT_JOURNAL_THREADS; index++) { - broadcast("agent.event", { type: "turnStarted", threadId: `other-${index}` }); - } - const next = broadcast("agent.event", { type: "turnStarted", threadId: "evicted-thread" }); - expect(next).not.toEqual(first); - const received: Array<{ buf: Buffer; binary: boolean }> = []; - const ws = fakeOpenSocket(received); - addClient(ws); - const result = setClientThreadSubscriptions(ws, ["evicted-thread"], { - "evicted-thread": { epoch: first.epoch, sequence: 1 }, - }); - expect(result.hydrationRequiredThreadIds).toEqual(["evicted-thread"]); - expect(received).toHaveLength(0); - }); - - it("requires hydration for an ambiguous legacy cursor or a cursor ahead of its journal", () => { - const first = AgentEventSchema().parse(broadcast("agent.event", { type: "turnStarted", threadId: "cursor-thread" })); - if (!first.epoch) throw new Error("Agent event has no journal epoch"); - const received: Array<{ buf: Buffer; binary: boolean }> = []; - const ws = fakeOpenSocket(received); - addClient(ws); - for (const cursor of [1, { epoch: first.epoch, sequence: 2 }]) { - const result = setClientThreadSubscriptions(ws, ["cursor-thread"], { "cursor-thread": cursor }); - expect(result.hydrationRequiredThreadIds).toEqual(["cursor-thread"]); - } - expect(received).toHaveLength(0); - }); - - it("replays only newer events for a cursor in the current journal generation", () => { - const first = AgentEventSchema().parse(broadcast("agent.event", { type: "turnStarted", threadId: "current-thread" })); - if (!first.epoch) throw new Error("Agent event has no journal epoch"); - broadcast("agent.event", { type: "textDelta", threadId: "current-thread", delta: "new text" }); - const received: Array<{ buf: Buffer; binary: boolean }> = []; - const ws = fakeOpenSocket(received); - addClient(ws); - const result = setClientThreadSubscriptions(ws, ["current-thread"], { - "current-thread": { epoch: first.epoch, sequence: 1 }, - }); - expect(result).toEqual({ hydrationRequiredThreadIds: [], replayedThrough: { "current-thread": 2 } }); - expect(received).toHaveLength(1); - }); - - it("does not replay history for legacy subscription calls without cursors", () => { - const received: Array<{ buf: Buffer; binary: boolean }> = []; - broadcast("agent.event", { type: "textDelta", threadId: "thread-legacy", delta: "old" }); - const ws = fakeOpenSocket(received); - addClient(ws); - setClientThreadSubscriptions(ws, ["thread-legacy"]); - expect(received).toHaveLength(0); - }); - - it("reports hydration when a cursor falls behind the bounded journal", () => { - const received: Array<{ buf: Buffer; binary: boolean }> = []; - for (let i = 0; i < 257; i++) { - broadcast("agent.event", { type: "textDelta", threadId: "thread-gap", delta: String(i) }); - } - const ws = fakeOpenSocket(received); - addClient(ws); - const result = setClientThreadSubscriptions(ws, ["thread-gap"], { "thread-gap": 0 }); - expect(result.hydrationRequiredThreadIds).toEqual(["thread-gap"]); - expect(received).toHaveLength(0); - }); - - it("reports hydration for a cursor from another server epoch without replay", () => { - const received: Array<{ buf: Buffer; binary: boolean }> = []; - broadcast("agent.event", { type: "textDelta", threadId: "thread-epoch", delta: "one" }); - const ws = fakeOpenSocket(received); - addClient(ws); - const result = setClientThreadSubscriptions(ws, ["thread-epoch"], { - "thread-epoch": { epoch: "00000000-0000-4000-8000-000000000001", sequence: 0 }, - }); - expect(result.hydrationRequiredThreadIds).toEqual(["thread-epoch"]); - expect(received).toHaveLength(0); - }); - it("clears all thread subscriptions when replacing with an empty set", () => { const received: Array<{ buf: Buffer; binary: boolean }> = []; const ws = fakeOpenSocket(received); @@ -358,11 +237,7 @@ describe("broadcast", () => { subscribeClientToThread(ws, "thread-a"); setClientThreadSubscriptions(ws, []); - broadcast("agent.event", { - type: "textDelta", - threadId: "thread-a", - delta: "ignored", - }); + broadcast("agent.canonical", canonicalBatch("thread-a")); expect(received).toHaveLength(0); }); @@ -418,7 +293,7 @@ describe("broadcast", () => { expect(good).toHaveLength(1); }); - it("terminates a client with unsent pushes and replays its retained agent gap", () => { + it("terminates a client with unsent pushes", () => { const slowReceived: Array<{ buf: Buffer; binary: boolean }> = []; const healthyReceived: Array<{ buf: Buffer; binary: boolean }> = []; const slow = Object.assign(new NodeEvents.EventEmitter(), fakeOpenSocket(slowReceived, false)); @@ -437,22 +312,12 @@ describe("broadcast", () => { broadcastTerminalData("pty-1", index, new Uint8Array(1_024 * 1_024)); } - broadcast("agent.event", { type: "textDelta", threadId: "thread-recovery", delta: "retained" }); + broadcast("agent.canonical", canonicalBatch("thread-recovery")); expect(terminate).toHaveBeenCalledOnce(); expect(clientCount()).toBe(1); expect(slowReceived).toHaveLength(15); expect(healthyReceived).toHaveLength(17); - - const recoveredReceived: Array<{ buf: Buffer; binary: boolean }> = []; - const recovered = fakeOpenSocket(recoveredReceived); - addClient(recovered); - const replay = setClientThreadSubscriptions(recovered, ["thread-recovery"], { "thread-recovery": 0 }); - - expect(replay).toEqual({ hydrationRequiredThreadIds: [], replayedThrough: { "thread-recovery": 1 } }); - expect(recoveredReceived.map((entry) => JSON.parse(entry.buf.toString("utf-8")).data)).toEqual([ - expect.objectContaining({ sequence: 1, delta: "retained" }), - ]); }); it("lets tests swap validating and pass-through payload adapters", () => { diff --git a/apps/server/src/application/transport/push-subscription-rpc.ts b/apps/server/src/application/transport/push-subscription-rpc.ts index f68554451..c6068c437 100644 --- a/apps/server/src/application/transport/push-subscription-rpc.ts +++ b/apps/server/src/application/transport/push-subscription-rpc.ts @@ -28,15 +28,15 @@ const pushSubscriptionHandlers: Record< if (client) unsubscribeClientFromThread(client, params.threadId); }, "push.setThreadSubscriptions": (deps, params, client) => { - if (!client) return emptySubscriptionReplay(); - const replay = setClientThreadSubscriptions(client, params.threadIds, params.cursors); + if (!client) return { canonicalRecoveries: [] }; + setClientThreadSubscriptions(client, params.threadIds); const canonicalRecoveries = params.revisions ? params.threadIds.flatMap((threadId: string) => { const revision = params.revisions?.[threadId]; return revision ? [deps.canonicalSink.recoverThread(threadId, revision)] : []; }) : []; - return { ...replay, canonicalRecoveries }; + return { canonicalRecoveries }; }, }; @@ -57,11 +57,3 @@ export function routePushSubscriptionRpc( } return pushSubscriptionHandlers[method](deps, params, client); } - -function emptySubscriptionReplay(): { - hydrationRequiredThreadIds: string[]; - replayedThrough: Record; - canonicalRecoveries: unknown[]; -} { - return { hydrationRequiredThreadIds: [], replayedThrough: {}, canonicalRecoveries: [] }; -} diff --git a/apps/server/src/application/transport/push.ts b/apps/server/src/application/transport/push.ts index a3b187c25..b512bccd6 100644 --- a/apps/server/src/application/transport/push.ts +++ b/apps/server/src/application/transport/push.ts @@ -4,30 +4,17 @@ */ import type { WebSocket } from "ws"; -import * as NodeCrypto from "node:crypto"; -import { WS_CHANNELS, type WsChannelName, type SetThreadSubscriptionsInput, encodeTerminalDataFrame } from "@mcode/contracts"; +import { WS_CHANNELS, type WsChannelName, encodeTerminalDataFrame } from "@mcode/contracts"; import { logger } from "@mcode/shared"; import { getTransportPayloadValidator } from "./payload-validation.js"; -/** Maximum transient agent events retained for one thread. */ -export const MAX_AGENT_EVENT_JOURNAL_EVENTS_PER_THREAD = 256; -/** Maximum thread journals retained by the process. */ -export const MAX_AGENT_EVENT_JOURNAL_THREADS = 100; /** Maximum unsent push bytes for one connected client, excluding RPC response bytes. */ export const MAX_QUEUED_PUSH_BYTES = 16 * 1_024 * 1_024; const clients = new Set(); const queuedPushBytes = new Map(); const threadSubscriptions = new Map>(); -interface AgentEventJournal { - epoch: string; - sequence: number; - events: unknown[]; -} - -const threadJournals = new Map(); const SUBSCRIPTION_SCOPED_CHANNELS = new Set([ - "agent.event", "agent.canonical", "turn.fileEffectsUpdated", "turn.diffChanged", @@ -96,65 +83,9 @@ export function unsubscribeClientFromThread(ws: WebSocket, threadId: string): vo export function setClientThreadSubscriptions( ws: WebSocket, threadIds: readonly string[], - cursors?: NonNullable, -): { hydrationRequiredThreadIds: string[]; replayedThrough: Record } { - if (!clients.has(ws)) return { hydrationRequiredThreadIds: [], replayedThrough: {} }; - threadSubscriptions.set(ws, new Set(threadIds)); - const hydrationRequiredThreadIds: string[] = []; - const replayedThrough: Record = {}; - if (!cursors) return { hydrationRequiredThreadIds, replayedThrough }; - - for (const threadId of threadIds) { - replayThreadSubscription( - ws, - threadId, - cursors[threadId], - hydrationRequiredThreadIds, - replayedThrough, - ); - } - return { hydrationRequiredThreadIds, replayedThrough }; -} - -function replayThreadSubscription( - ws: WebSocket, - threadId: string, - rawCursor: NonNullable[string] | undefined, - hydrationRequiredThreadIds: string[], - replayedThrough: Record, ): void { - if (rawCursor === undefined) return; - const cursor = typeof rawCursor === "number" ? rawCursor : rawCursor.sequence; - const journal = threadJournals.get(threadId); - if (requiresThreadHydration(rawCursor, cursor, journal)) { - hydrationRequiredThreadIds.push(threadId); - return; - } - if (!journal) return; - const deliveredThrough = replayJournalEvents(ws, cursor, journal.events); - if (deliveredThrough !== cursor) replayedThrough[threadId] = deliveredThrough; -} - -function requiresThreadHydration( - rawCursor: NonNullable[string], - cursor: number, - journal: AgentEventJournal | undefined, -): boolean { - if (typeof rawCursor === "number" && cursor > 0) return true; - if (!journal) return cursor > 0; - if (typeof rawCursor !== "number" && rawCursor.epoch !== journal.epoch) return true; - if (cursor > journal.sequence) return true; - return journal.events.length > 0 && cursor < (journal.events[0] as { sequence: number }).sequence - 1; -} - -function replayJournalEvents(ws: WebSocket, cursor: number, events: readonly unknown[]): number { - let deliveredThrough = cursor; - for (const event of events) { - if ((event as { sequence: number }).sequence <= cursor) continue; - if (!sendToClient(ws, "agent.event", event)) break; - deliveredThrough = (event as { sequence: number }).sequence; - } - return deliveredThrough; + if (!clients.has(ws)) return; + threadSubscriptions.set(ws, new Set(threadIds)); } /** Get the current number of connected clients. */ @@ -197,10 +128,8 @@ export function broadcast( } const threadId = payloadThreadId(data); - const candidate = decorateAgentEvent(channel, data, threadId); - const validation = getTransportPayloadValidator().validatePush(channel, candidate, schema); + const validation = getTransportPayloadValidator().validatePush(channel, data, schema); if (!validation.ok) return undefined; - retainAgentEvent(channel, threadId, validation.data); const payload = JSON.stringify({ type: "push" as const, channel, @@ -210,32 +139,6 @@ export function broadcast( return validation.data; } -function decorateAgentEvent(channel: WsChannelName, data: unknown, threadId: string | undefined): unknown { - if (channel !== "agent.event" || !threadId || !data || typeof data !== "object") return data; - const journal = threadJournals.get(threadId); - const sequence = (journal?.sequence ?? 0) + 1; - const epoch = journal?.epoch ?? NodeCrypto.randomUUID(); - return { ...(data as Record), sequence, epoch }; -} - -function retainAgentEvent(channel: WsChannelName, threadId: string | undefined, event: unknown): void { - if (channel !== "agent.event" || !threadId) return; - if (!event || typeof event !== "object" || !("sequence" in event) || !("epoch" in event)) return; - if (typeof event.sequence !== "number" || typeof event.epoch !== "string") return; - const existing = threadJournals.get(threadId); - const events = existing ? [...existing.events, event] : [event]; - events.splice(0, Math.max(0, events.length - MAX_AGENT_EVENT_JOURNAL_EVENTS_PER_THREAD)); - threadJournals.delete(threadId); - threadJournals.set(threadId, { epoch: event.epoch, sequence: event.sequence, events }); - trimJournalMap(threadJournals); -} - -function trimJournalMap(map: Map): void { - while (map.size > MAX_AGENT_EVENT_JOURNAL_THREADS) { - map.delete(map.keys().next().value as string); - } -} - function sendBroadcastPayload(channel: WsChannelName, threadId: string | undefined, payload: string): void { const requiresThreadSubscription = SUBSCRIPTION_SCOPED_CHANNELS.has(channel); const payloadBytes = Buffer.byteLength(payload, "utf8"); @@ -342,5 +245,4 @@ export function _resetForTest(): void { clients.clear(); queuedPushBytes.clear(); threadSubscriptions.clear(); - threadJournals.clear(); } diff --git a/apps/server/src/features/agents/canonical/__tests__/canonical-agent-boundary-stub.ts b/apps/server/src/features/agents/canonical/__tests__/canonical-agent-boundary-stub.ts index f0e4066ba..56b2574a0 100644 --- a/apps/server/src/features/agents/canonical/__tests__/canonical-agent-boundary-stub.ts +++ b/apps/server/src/features/agents/canonical/__tests__/canonical-agent-boundary-stub.ts @@ -1,4 +1,5 @@ import type { CanonicalAgentBoundary } from "../../index.js"; +import { broadcast } from "../../../../application/transport/push.js"; import type { Database } from "bun:sqlite"; /** Creates an AgentService test seam that runs compatibility writes without canonical persistence. */ @@ -26,15 +27,19 @@ export function createCanonicalAgentBoundaryStub( finishCanonicalChildTurn: () => null, recordProviderDiagnostic: () => undefined, recordCodexChildRoutingDiagnostic: () => false, - // Synthesized publications bypass persistence but keep the publication.recorded shape so the - // legacy broadcast side effect survives in tests without a canonical store. + // Synthesized publications bypass persistence but keep the publication-stamped wire shape + // so tests can observe the same event payload the renderer would project. recordSynthesizedPublications: (_threadId: string, events: readonly Record[]) => - events.map((event, index) => ({ - payload: { - type: "publication.recorded", - publicationId: String(index + 1), - event: { ...event, publicationId: String(index + 1) }, - }, - })), + events.map((event, index) => { + const stamped = { ...event, publicationId: String(index + 1) }; + broadcast("agent.event", stamped); + return { + payload: { + type: "publication.recorded", + publicationId: String(index + 1), + event: stamped, + }, + }; + }), } as unknown as CanonicalAgentBoundary; } diff --git a/apps/server/src/features/agents/canonical/synthesized-agent-event-publication.ts b/apps/server/src/features/agents/canonical/synthesized-agent-event-publication.ts index 829aa4c7c..8132c8202 100644 --- a/apps/server/src/features/agents/canonical/synthesized-agent-event-publication.ts +++ b/apps/server/src/features/agents/canonical/synthesized-agent-event-publication.ts @@ -1,19 +1,14 @@ import type { AgentEvent } from "@mcode/contracts"; -import { broadcast } from "../../../application/transport/push.js"; import type { CanonicalAgentBoundary } from "./canonical-agent-boundary.js"; /** - * Delivers server-synthesized agent events through the canonical stream and the legacy broadcast. - * Both copies share the allocated publicationId, so the client applies whichever copy arrives first. + * Delivers server-synthesized agent events through the canonical stream. The + * committed publication envelopes are returned so callers can observe them. */ export function publishSynthesizedAgentEvents( canonical: Pick, threadId: string, events: readonly AgentEvent[], ): void { - for (const envelope of canonical.recordSynthesizedPublications(threadId, events)) { - if (envelope.payload.type === "publication.recorded") { - broadcast("agent.event", envelope.payload.event); - } - } + canonical.recordSynthesizedPublications(threadId, events); } diff --git a/apps/server/src/features/agents/commands/goal-command.ts b/apps/server/src/features/agents/commands/goal-command.ts index 228c2bea7..8456aa05e 100644 --- a/apps/server/src/features/agents/commands/goal-command.ts +++ b/apps/server/src/features/agents/commands/goal-command.ts @@ -11,7 +11,7 @@ import { import type { MessageRepo } from "../conversation/persistence/message-repo.js"; import type { CommandContext, CommandOutcome, McodeCommand } from "./command-router.js"; -/** Publishes synthesized agent events through the canonical stream and the legacy broadcast. */ +/** Publishes synthesized agent events through the canonical stream. */ type PublishSynthesizedFn = (threadId: string, events: readonly AgentEvent[]) => void; /** Repositories and database handle the command needs to persist its rows. */ diff --git a/apps/server/src/features/agents/events/__tests__/provider-event-publication.test.ts b/apps/server/src/features/agents/events/__tests__/provider-event-publication.test.ts index 087b3a7c7..d3760638c 100644 --- a/apps/server/src/features/agents/events/__tests__/provider-event-publication.test.ts +++ b/apps/server/src/features/agents/events/__tests__/provider-event-publication.test.ts @@ -4,7 +4,6 @@ import { publishParentProviderEvent } from "../provider-event-publication.js"; function buildPublicationDeps() { return { - publishAgentEvent: vi.fn(), updateThreadStatus: vi.fn(), publishThreadStatus: vi.fn(), }; @@ -12,7 +11,7 @@ function buildPublicationDeps() { describe("provider event publication ownership", () => { - it("does not publish generated attachments to the parent UI", () => { + it("does not apply a status for generated attachments", () => { const deps = buildPublicationDeps(); const event: AgentEvent = { type: AgentEventType.GeneratedAttachment, @@ -25,8 +24,9 @@ describe("provider event publication ownership", () => { }, }; - expect(publishParentProviderEvent(event, event, deps)).toBe(false); - expect(deps.publishAgentEvent).not.toHaveBeenCalled(); + expect(publishParentProviderEvent(event, deps)).toBe(false); + expect(deps.updateThreadStatus).not.toHaveBeenCalled(); + expect(deps.publishThreadStatus).not.toHaveBeenCalled(); }); it("publishes a parent completion and updates its status", () => { @@ -40,8 +40,7 @@ describe("provider event publication ownership", () => { tokensOut: 1, }; - expect(publishParentProviderEvent(parentEvent, parentEvent, deps)).toBe(true); - expect(deps.publishAgentEvent).toHaveBeenCalledWith(parentEvent); + expect(publishParentProviderEvent(parentEvent, deps)).toBe(true); expect(deps.updateThreadStatus).toHaveBeenCalledWith("parent-thread", "completed"); expect(deps.publishThreadStatus).toHaveBeenCalledWith({ threadId: "parent-thread", @@ -49,7 +48,7 @@ describe("provider event publication ownership", () => { }); }); - it("publishes a committed terminal replay without changing a newer thread status", () => { + it("skips the status transition for a committed terminal replay", () => { const deps = buildPublicationDeps(); const event: AgentEvent = { type: AgentEventType.TurnComplete, @@ -62,8 +61,7 @@ describe("provider event publication ownership", () => { tokensOut: 1, }; - expect(publishParentProviderEvent(event, event, deps)).toBe(true); - expect(deps.publishAgentEvent).toHaveBeenCalledWith(event); + expect(publishParentProviderEvent(event, deps)).toBe(true); expect(deps.updateThreadStatus).not.toHaveBeenCalled(); expect(deps.publishThreadStatus).not.toHaveBeenCalled(); }); @@ -76,8 +74,7 @@ describe("provider event publication ownership", () => { error: "parent provider disconnected", }; - expect(publishParentProviderEvent(parentEvent, parentEvent, deps)).toBe(true); - expect(deps.publishAgentEvent).toHaveBeenCalledWith(parentEvent); + expect(publishParentProviderEvent(parentEvent, deps)).toBe(true); expect(deps.updateThreadStatus).toHaveBeenCalledWith("parent-thread", "errored"); expect(deps.publishThreadStatus).toHaveBeenCalledWith({ threadId: "parent-thread", @@ -99,7 +96,7 @@ describe("provider event publication ownership", () => { outcome, }; - expect(publishParentProviderEvent(event, event, deps)).toBe(true); + expect(publishParentProviderEvent(event, deps)).toBe(true); expect(deps.updateThreadStatus).toHaveBeenCalledWith("parent-thread", status); expect(deps.publishThreadStatus).toHaveBeenCalledWith({ threadId: "parent-thread", @@ -107,7 +104,7 @@ describe("provider event publication ownership", () => { }); }); - it("does not publish a terminal status for an outcome-less Ended", () => { + it("does not apply a terminal status for an outcome-less Ended", () => { const deps = buildPublicationDeps(); const event: AgentEvent = { type: AgentEventType.Ended, @@ -115,8 +112,7 @@ describe("provider event publication ownership", () => { turnExecutionId: "00000000-0000-4000-8000-000000000001", }; - expect(publishParentProviderEvent(event, event, deps)).toBe(false); - expect(deps.publishAgentEvent).not.toHaveBeenCalled(); + expect(publishParentProviderEvent(event, deps)).toBe(false); expect(deps.updateThreadStatus).not.toHaveBeenCalled(); expect(deps.publishThreadStatus).not.toHaveBeenCalled(); }); @@ -130,8 +126,7 @@ describe("provider event publication ownership", () => { reason: "provider_lost", }; - expect(publishParentProviderEvent(event, event, deps)).toBe(true); - expect(deps.publishAgentEvent).toHaveBeenCalledWith(event); + expect(publishParentProviderEvent(event, deps)).toBe(true); expect(deps.updateThreadStatus).not.toHaveBeenCalled(); expect(deps.publishThreadStatus).not.toHaveBeenCalled(); }); diff --git a/apps/server/src/features/agents/events/provider-event-publication.ts b/apps/server/src/features/agents/events/provider-event-publication.ts index 036b6e271..942e78e1a 100644 --- a/apps/server/src/features/agents/events/provider-event-publication.ts +++ b/apps/server/src/features/agents/events/provider-event-publication.ts @@ -2,20 +2,16 @@ import type { AgentEvent } from "@mcode/contracts"; /** Dependencies used to publish one normalized provider event to the parent UI. */ export interface ParentProviderEventPublicationDeps { - publishAgentEvent: (event: AgentEvent) => void; updateThreadStatus: (threadId: string, status: "completed" | "errored" | "interrupted") => void; publishThreadStatus: (payload: { threadId: string; status: "completed" | "errored" | "interrupted" }) => void; } -/** Publish a normalized parent event and apply its legacy parent status transition. */ +/** Apply the status transition of a publishable normalized provider event. */ export function publishParentProviderEvent( event: AgentEvent, - enrichedEvent: AgentEvent, deps: ParentProviderEventPublicationDeps, ): boolean { if (!shouldPublishParentEvent(event)) return false; - - deps.publishAgentEvent(enrichedEvent); // A semantic writer committed this status before releasing its terminal event. // A late replay must not overwrite the status of a newer execution. if (event.publicationId !== undefined) return true; diff --git a/apps/server/src/features/agents/orchestration/__tests__/agent-service-transient-retry.test.ts b/apps/server/src/features/agents/orchestration/__tests__/agent-service-transient-retry.test.ts index fb83c7ffe..93055ac82 100644 --- a/apps/server/src/features/agents/orchestration/__tests__/agent-service-transient-retry.test.ts +++ b/apps/server/src/features/agents/orchestration/__tests__/agent-service-transient-retry.test.ts @@ -343,8 +343,7 @@ describe("AgentService transient-failure auto-retry", () => { it("does not publish an interrupted status for suppressed retry teardown", async () => { const { service, sendTurn, providerEmitter, threadRepo } = buildService(); startAgentServiceIngressForTest(service, (event) => { - publishParentProviderEvent(event, event, { - publishAgentEvent: vi.fn(), + publishParentProviderEvent(event, { updateThreadStatus: threadRepo.updateStatus, publishThreadStatus: vi.fn(), }); diff --git a/apps/server/src/features/agents/orchestration/__tests__/agent-service-turn-cleanup.test.ts b/apps/server/src/features/agents/orchestration/__tests__/agent-service-turn-cleanup.test.ts index 1a65b16ef..5d4a8911e 100644 --- a/apps/server/src/features/agents/orchestration/__tests__/agent-service-turn-cleanup.test.ts +++ b/apps/server/src/features/agents/orchestration/__tests__/agent-service-turn-cleanup.test.ts @@ -1573,8 +1573,7 @@ describe("AgentService Ended finalization", () => { new SubagentLifecycleService(canonicalSink, providerRegistry), ); startAgentServiceIngressForTest(service, (event) => { - publishParentProviderEvent(event, event, { - publishAgentEvent: vi.fn(), + publishParentProviderEvent(event, { updateThreadStatus: (threadId, status) => threadRepo.updateStatus(threadId, status), publishThreadStatus: (payload) => broadcast("thread.status", payload), }); diff --git a/apps/server/src/features/agents/orchestration/__tests__/start-agent-orchestration.test.ts b/apps/server/src/features/agents/orchestration/__tests__/start-agent-orchestration.test.ts index 90740477b..2f48cb9c1 100644 --- a/apps/server/src/features/agents/orchestration/__tests__/start-agent-orchestration.test.ts +++ b/apps/server/src/features/agents/orchestration/__tests__/start-agent-orchestration.test.ts @@ -26,7 +26,6 @@ function buildOrchestration() { })), updateStatus: vi.fn(), }; - const publishedEvents: AgentEvent[] = []; const publishThreadStatus = vi.fn(); const pullRequestCompletionEffect = { schedule: vi.fn() }; @@ -34,10 +33,8 @@ function buildOrchestration() { runtime, publicationRegistry, threadRepo, - narrativeStore: { getCurrentParentToolCallId: vi.fn(() => "agent-parent") }, pullRequestCompletionEffect, providerRegistry: { resolveAll: vi.fn(() => []) }, - publishAgentEvent: (event: AgentEvent) => publishedEvents.push(event), publishPermissionRequest: vi.fn(), publishPermissionResolved: vi.fn(), publishThreadStatus, @@ -48,52 +45,12 @@ function buildOrchestration() { publish, runtime, threadRepo, - publishedEvents, publishThreadStatus, pullRequestCompletionEffect, }; } describe("agent orchestration", () => { - it("publishes enriched parent events once while keeping attachments private", () => { - const orchestration = buildOrchestration(); - - orchestration.publish({ - type: AgentEventType.ToolUse, - threadId: "thread-1", - toolCallId: "tool-1", - toolName: "Edit", - toolInput: { - file_path: "src/app.ts", - old_string: "secret", - new_string: "replacement", - }, - }); - - expect(orchestration.publishedEvents).toHaveLength(1); - expect(orchestration.publishedEvents[0]).toMatchObject({ - type: AgentEventType.ToolUse, - parentToolCallId: "agent-parent", - toolInput: { file_path: "src/app.ts" }, - }); - expect(orchestration.publishedEvents[0]).not.toMatchObject({ - toolInput: { old_string: expect.anything(), new_string: expect.anything() }, - }); - - orchestration.publish({ - type: AgentEventType.GeneratedAttachment, - threadId: "thread-1", - attachment: { - id: "attachment-1", - name: "capture.png", - mimeType: "image/png", - sizeBytes: 1, - }, - }); - - expect(orchestration.publishedEvents).toHaveLength(1); - }); - it("keeps terminal status publication on the parent event path", () => { const orchestration = buildOrchestration(); const event: AgentEvent = { @@ -107,7 +64,6 @@ describe("agent orchestration", () => { orchestration.publish(event); - expect(orchestration.publishedEvents).toEqual([event]); expect(orchestration.threadRepo.updateStatus).toHaveBeenCalledWith("thread-1", "completed"); expect(orchestration.publishThreadStatus).toHaveBeenCalledWith({ threadId: "thread-1", @@ -116,18 +72,17 @@ describe("agent orchestration", () => { expect(orchestration.pullRequestCompletionEffect.schedule).toHaveBeenCalledWith("thread-1"); }); - it("releases a committed terminal receipt even when the legacy runtime suppression is set", () => { + it("suppresses a replayed terminal status but releases a committed terminal receipt", () => { const orchestration = buildOrchestration(); orchestration.runtime.shouldSuppressTurnComplete.mockReturnValue(true); const legacy: AgentEvent = { type: AgentEventType.TurnComplete, threadId: "thread-1", reason: "completed", costUsd: null, tokensIn: 1, tokensOut: 1 }; orchestration.publish(legacy); - expect(orchestration.publishedEvents).toEqual([]); + expect(orchestration.threadRepo.updateStatus).not.toHaveBeenCalled(); const committed: AgentEvent = { ...legacy, turnExecutionId: "00000000-0000-4000-8000-000000000001", publicationId: "1" }; orchestration.publish(committed); - expect(orchestration.publishedEvents).toEqual([committed]); expect(orchestration.threadRepo.updateStatus).not.toHaveBeenCalled(); expect(orchestration.publishThreadStatus).not.toHaveBeenCalled(); expect(orchestration.pullRequestCompletionEffect.schedule).toHaveBeenCalledWith("thread-1"); diff --git a/apps/server/src/features/agents/orchestration/agent-event-publication-service.ts b/apps/server/src/features/agents/orchestration/agent-event-publication-service.ts index e7c74be7b..395fe1e60 100644 --- a/apps/server/src/features/agents/orchestration/agent-event-publication-service.ts +++ b/apps/server/src/features/agents/orchestration/agent-event-publication-service.ts @@ -1,13 +1,11 @@ import { AgentEventType, type AgentEvent, type IProviderRegistry, type PermissionRequest } from "@mcode/contracts"; import type { TurnPullRequestCompletionEffect } from "../../pull-requests/index.js"; import type { ThreadRepo } from "../../thread-control/persistence/thread-repo.js"; -import type { NarrativeStore } from "../conversation/narrative/narrative-store.js"; import { publishParentProviderEvent } from "../events/provider-event-publication.js"; import { publishAgentPermissionEvents } from "../permissions/permission-publication.js"; -import { sanitizePublicToolInput } from "../tools/input/public-tool-input.js"; import { serverWorkTrace } from "../diagnostics/server-work-trace.js"; -/** The runtime read surface needed to prepare renderer-facing provider events. */ +/** The runtime read surface needed to gate provider event status transitions. */ export interface AgentEventPublicationRuntime { getCurrentFileEffectTurnId(threadId: string): string | undefined; shouldSuppressTurnEnded(threadId: string): boolean; @@ -19,10 +17,8 @@ export interface AgentEventPublicationRuntime { export interface AgentEventPublicationDependencies { runtime: AgentEventPublicationRuntime; threads: ThreadRepo; - narrative: NarrativeStore; pullRequests: TurnPullRequestCompletionEffect; providers: IProviderRegistry; - publishAgentEvent(event: AgentEvent): void; publishPermissionRequest(request: PermissionRequest): void; publishPermissionResolved(payload: { requestId: string; decision: "allow" | "allow-session" | "deny" | "cancelled"; optionLabel?: string }): void; publishThreadStatus(payload: { threadId: string; status: "completed" | "errored" | "interrupted" }): void; @@ -52,9 +48,7 @@ export class AgentEventPublicationService { private publishCore(event: AgentEvent): void { if (this.shouldSuppress(event)) return; - const enriched = this.sanitize(this.enrich(event)); - const published = publishParentProviderEvent(event, enriched, { - publishAgentEvent: this.dependencies.publishAgentEvent, + const published = publishParentProviderEvent(event, { updateThreadStatus: (threadId, status) => this.dependencies.threads.updateStatus(threadId, status), publishThreadStatus: this.dependencies.publishThreadStatus, }); @@ -69,32 +63,4 @@ export class AgentEventPublicationService { return event.type === AgentEventType.Error && this.dependencies.runtime.shouldSuppressTransientTurnError(event.threadId, event.error ?? ""); } - - private enrich(event: AgentEvent): AgentEvent { - const withFileEffect = this.enrichTurnStart(event); - const withParent = this.enrichToolParent(withFileEffect); - return withParent; - } - - private enrichTurnStart(event: AgentEvent): AgentEvent { - if (event.type !== AgentEventType.TurnStarted) return event; - const fileEffectTurnId = this.dependencies.runtime.getCurrentFileEffectTurnId(event.threadId); - return fileEffectTurnId ? { ...event, fileEffectTurnId } : event; - } - - private enrichToolParent(event: AgentEvent): AgentEvent { - if (event.type !== AgentEventType.ToolUse || event.toolName === "Agent" || event.parentToolCallId) return event; - const parentToolCallId = this.dependencies.narrative.getCurrentParentToolCallId(event.threadId); - return parentToolCallId ? { ...event, parentToolCallId } : event; - } - - private sanitize(event: AgentEvent): AgentEvent { - if (event.type === AgentEventType.ToolUse) { - return { ...event, toolInput: sanitizePublicToolInput(event.toolInput, event.toolName) }; - } - if (event.type === AgentEventType.ToolResult && event.toolInput) { - return { ...event, toolInput: sanitizePublicToolInput(event.toolInput) }; - } - return event; - } } diff --git a/apps/server/src/features/agents/orchestration/start-agent-orchestration.ts b/apps/server/src/features/agents/orchestration/start-agent-orchestration.ts index 2c935f166..54e1d96ae 100644 --- a/apps/server/src/features/agents/orchestration/start-agent-orchestration.ts +++ b/apps/server/src/features/agents/orchestration/start-agent-orchestration.ts @@ -1,8 +1,7 @@ -import type { AgentEvent, IProviderRegistry, PermissionRequest } from "@mcode/contracts"; +import type { IProviderRegistry, PermissionRequest } from "@mcode/contracts"; import { AgentEventPublicationRegistry } from "./agent-event-publication-registry.js"; import type { AgentEventPublicationRuntime } from "./agent-event-publication-service.js"; import type { TurnPullRequestCompletionEffect } from "../../pull-requests/index.js"; -import type { NarrativeStore } from "../conversation/narrative/narrative-store.js"; import type { ThreadRepo } from "../../thread-control/persistence/thread-repo.js"; import { AgentEventPublicationService } from "./agent-event-publication-service.js"; @@ -10,10 +9,8 @@ interface AgentOrchestrationDependencies { runtime: AgentEventPublicationRuntime; publicationRegistry: AgentEventPublicationRegistry; threadRepo: ThreadRepo; - narrativeStore: NarrativeStore; pullRequestCompletionEffect: TurnPullRequestCompletionEffect; providerRegistry: IProviderRegistry; - publishAgentEvent: (event: AgentEvent) => void; publishPermissionRequest: (request: PermissionRequest) => void; publishPermissionResolved: (payload: { requestId: string; decision: "allow" | "allow-session" | "deny" | "cancelled"; optionLabel?: string }) => void; publishThreadStatus: (payload: { threadId: string; status: "completed" | "errored" | "interrupted" }) => void; @@ -27,10 +24,8 @@ export function startAgentOrchestration({ runtime, publicationRegistry, threadRepo, - narrativeStore, pullRequestCompletionEffect, providerRegistry, - publishAgentEvent, publishPermissionRequest, publishPermissionResolved, publishThreadStatus, @@ -38,10 +33,8 @@ export function startAgentOrchestration({ const publication = new AgentEventPublicationService({ runtime, threads: threadRepo, - narrative: narrativeStore, pullRequests: pullRequestCompletionEffect, providers: providerRegistry, - publishAgentEvent, publishPermissionRequest, publishPermissionResolved, publishThreadStatus, diff --git a/apps/server/src/features/agents/orchestration/turn-runtime-controller.ts b/apps/server/src/features/agents/orchestration/turn-runtime-controller.ts index b6c6edb17..fe4f44567 100644 --- a/apps/server/src/features/agents/orchestration/turn-runtime-controller.ts +++ b/apps/server/src/features/agents/orchestration/turn-runtime-controller.ts @@ -574,12 +574,15 @@ export class TurnRuntimeController implements TurnLifecycleControl, TurnRuntimeE } this.eventApplication.beginPreparedTurn(lease.threadId, lease.turnExecutionId, request.turnId); this.turnAdmissions.markDispatchActive(lease.threadId); + // The tracker generation must exist before turnStarted is ingested so its + // canonical publication embed carries the same fileEffectTurnId the wire path enriched. + await this.ensureTurnFileTracking(lease.threadId, prepared.cwd); this.emitProviderEvent(prepared.provider, { type: "turnStarted", threadId: lease.threadId, turnExecutionId: lease.turnExecutionId, + fileEffectTurnId: this.turnFileEffects.currentTurnId(lease.threadId), } satisfies AgentEvent); - await this.ensureTurnFileTracking(lease.threadId, prepared.cwd); await this.turnFileEffects.get(lease.threadId); this.eventApplication.recordContextSeed(lease.threadId, prepared.contextSeed, prepared.contextWindow ?? undefined); } diff --git a/apps/server/src/features/agents/transport/agent-rpc.ts b/apps/server/src/features/agents/transport/agent-rpc.ts index f3b7719b4..89d09a983 100644 --- a/apps/server/src/features/agents/transport/agent-rpc.ts +++ b/apps/server/src/features/agents/transport/agent-rpc.ts @@ -144,6 +144,9 @@ const agentHandlers: AgentRpcHandlerMap = { }, "agent.stop": (deps, params) => deps.agentService.stopSession(params.threadId), "agent.activeCount": (deps) => deps.agentService.runtimeAccess().activeCount(), + // The runtime registry is the reconnect snapshot authority: startup recovery + // interrupts every unfinished canonical checkpoint, so registry entries are + // exactly the executions canonical state still considers live. "agent.listRunning": (deps) => deps.agentService.runtimeAccess().runtimeSnapshots(), "agent.answerQuestions": async (deps, params) => { await deps.planTurnService.answerQuestions( diff --git a/apps/server/src/features/agents/turns/__tests__/turn-finalizer.test.ts b/apps/server/src/features/agents/turns/__tests__/turn-finalizer.test.ts index c19a8c583..8ab735ded 100644 --- a/apps/server/src/features/agents/turns/__tests__/turn-finalizer.test.ts +++ b/apps/server/src/features/agents/turns/__tests__/turn-finalizer.test.ts @@ -17,6 +17,7 @@ import type { TurnOutcome } from "../turn-outcome.js"; import type { TurnFileTracker } from "../turn-file-tracker.js"; import { broadcast } from "../../../../application/transport/push.js"; import { CanonicalAgentBoundary } from "../../canonical/canonical-agent-boundary.js"; +import { createCanonicalAgentBoundaryStub } from "../../canonical/__tests__/canonical-agent-boundary-stub.js"; import { ParentAssistantTextCheckpointService } from "../parent-assistant-text-checkpoint-service.js"; vi.mock("../../../../application/transport/push.js", () => ({ broadcast: vi.fn() })); @@ -87,6 +88,8 @@ describe("TurnFinalizer.finalize — turn outcome → tool-call status", () => { snapshotService, turnSnapshotRepo, db, + undefined, + createCanonicalAgentBoundaryStub(db), ); }); diff --git a/apps/server/src/features/agents/turns/turn-finalizer.ts b/apps/server/src/features/agents/turns/turn-finalizer.ts index 81bcd0f08..2c40a7eda 100644 --- a/apps/server/src/features/agents/turns/turn-finalizer.ts +++ b/apps/server/src/features/agents/turns/turn-finalizer.ts @@ -631,8 +631,9 @@ export class TurnFinalizer { * emitted a `Message` — the accumulated streaming text. The deterministic * per-turn id makes a replayed write an `INSERT OR IGNORE` no-op. * - * For the interrupted streaming-text path it also broadcasts `agent.event` so - * clients align their in-memory transcript with the new DB row; the provider + * For the interrupted streaming-text path it also publishes a canonical + * Message event so clients align their in-memory transcript with the new DB + * row; the provider * body path needs no broadcast because the `Message` event already carried * this id to the client. Returns null only when the write throws. */ @@ -713,11 +714,7 @@ export class TurnFinalizer { messageId: materialized.id, ...(materialized.attachments.length > 0 ? { attachments: materialized.attachments } : {}), } satisfies AgentEvent; - // Without a canonical sink there is no durable publication, so keep the legacy-only copy. - if (!this.canonicalSink) { - broadcast("agent.event", event); - return; - } + if (!this.canonicalSink) return; publishSynthesizedAgentEvents(this.canonicalSink, threadId, [event]); } diff --git a/apps/web/src/__tests__/agent-event-branches.test.ts b/apps/web/src/__tests__/agent-event-branches.test.ts index 792e1091f..14ecb0488 100644 --- a/apps/web/src/__tests__/agent-event-branches.test.ts +++ b/apps/web/src/__tests__/agent-event-branches.test.ts @@ -466,61 +466,10 @@ describe("handleAgentEvent branches", () => { expect(calls[0].isComplete).toBe(false); }); - it("deduplicates sequenced replay events while accepting a first sequence above one", () => { - const handleAgentEvent = useThreadStore.getState().handleAgentEvent; - - handleAgentEvent({ - type: "toolUse", - threadId: "thread-1", - sequence: 17, - toolCallId: "seq-tool-1", - toolName: "Read", - toolInput: { path: "/first" }, - } as AgentEvent); - handleAgentEvent({ - type: "toolUse", - threadId: "thread-1", - sequence: 17, - toolCallId: "seq-tool-1", - toolName: "Read", - toolInput: { path: "/replay" }, - } as AgentEvent); - handleAgentEvent({ - type: "toolUse", - threadId: "thread-1", - sequence: 18, - toolCallId: "seq-tool-2", - toolName: "Write", - toolInput: { path: "/second" }, - } as AgentEvent); - - const record = readThreadField("thread-1", (thread) => thread); - expect(record.lastAgentEventSequence).toBe(18); - expect(record.toolCalls.map((call) => call.id)).toEqual(["seq-tool-1", "seq-tool-2"]); - expect(record.toolCalls[0]?.toolInput).toEqual({ path: "/first" }); - }); - - it("resets sequence authority when server event epoch changes", () => { - const handleAgentEvent = useThreadStore.getState().handleAgentEvent; - handleAgentEvent({ - type: "toolUse", threadId: "thread-1", epoch: "00000000-0000-4000-8000-000000000001", - sequence: 9, toolCallId: "old", toolName: "Read", toolInput: {}, - } as AgentEvent); - handleAgentEvent({ - type: "toolUse", threadId: "thread-1", epoch: "00000000-0000-4000-8000-000000000002", - sequence: 1, toolCallId: "new", toolName: "Write", toolInput: {}, - } as AgentEvent); - expect(readThreadField("thread-1", (thread) => thread.lastAgentEventSequence)).toBe(1); - expect(readThreadField("thread-1", (thread) => thread.lastAgentEventEpoch)) - .toBe("00000000-0000-4000-8000-000000000002"); - }); - it("does not reapply a stable publication after thread rehydration and a server restart", () => { const threadId = "thread-stable-publication"; const executionId = "00000000-0000-4000-8000-000000000001"; const publicationId = "2"; - const firstEpoch = "00000000-0000-4000-8000-000000000002"; - const nextEpoch = "00000000-0000-4000-8000-000000000003"; const key = `mcode-agent-publication-v2:${threadId}`; sessionStorage.removeItem(key); try { @@ -530,7 +479,7 @@ describe("handleAgentEvent branches", () => { turnExecutionId: executionId }]]), }); const original: Extract = { type: "toolUse", threadId, turnExecutionId: executionId, - publicationId, epoch: firstEpoch, sequence: 8, toolCallId: "stable-call", + publicationId, toolCallId: "stable-call", toolName: "Read", toolInput: { path: "/original" } }; useThreadStore.getState().handleAgentEvent(original); const persistedCalls = readThreadField(threadId, (record) => record.toolCalls); @@ -541,12 +490,11 @@ describe("handleAgentEvent branches", () => { records: new Map([[threadId, { ...createEmptyThreadRecord(), runtimePhase: "running", turnExecutionId: executionId, toolCalls: persistedCalls }]]), }); - useThreadStore.getState().handleAgentEvent({ ...original, epoch: nextEpoch, - sequence: 1, toolInput: { path: "/replayed" } }); + useThreadStore.getState().handleAgentEvent({ ...original, + toolInput: { path: "/replayed" } }); expect(readThreadField(threadId, (record) => record.toolCalls)).toEqual(persistedCalls); - expect(readThreadField(threadId, (record) => record.lastAgentEventEpoch)).toBe(nextEpoch); useThreadStore.getState().handleAgentEvent({ ...original, publicationId: undefined, - epoch: nextEpoch, sequence: 2, toolCallId: "legacy-call" }); + toolCallId: "legacy-call" }); expect(readThreadField(threadId, (record) => record.toolCalls)).toHaveLength(2); } finally { sessionStorage.removeItem(key); @@ -565,8 +513,7 @@ describe("handleAgentEvent branches", () => { turnExecutionId: executionId, streaming: "completed answer" }]]), }); const completion = { type: "turnComplete", threadId, turnExecutionId: executionId, - publicationId: "3", sequence: 3, - epoch: "00000000-0000-4000-8000-000000000012", reason: "end_turn", + publicationId: "3", reason: "end_turn", costUsd: null, tokensIn: 0, tokensOut: 0 } as AgentEvent; useThreadStore.getState().handleAgentEvent(completion); const completed = readThreadField(threadId, (record) => record); @@ -576,8 +523,7 @@ describe("handleAgentEvent branches", () => { resetThreadStoreForTests({ currentThreadId: threadId, runningThreadIds: new Set(), records: new Map([[threadId, completed]]), }); - useThreadStore.getState().handleAgentEvent({ ...completion, - epoch: "00000000-0000-4000-8000-000000000013", sequence: 1 }); + useThreadStore.getState().handleAgentEvent({ ...completion }); expect(readThreadField(threadId, (record) => record.messages)).toHaveLength(messageCount); expect(readThreadField(threadId, (record) => record.runtimePhase)).toBe(completed.runtimePhase); } finally { diff --git a/apps/web/src/__tests__/threadStore-followup-runtime.test.ts b/apps/web/src/__tests__/threadStore-followup-runtime.test.ts index 2e5640d02..ea6e60d96 100644 --- a/apps/web/src/__tests__/threadStore-followup-runtime.test.ts +++ b/apps/web/src/__tests__/threadStore-followup-runtime.test.ts @@ -57,9 +57,9 @@ describe("follow-up runtime ownership", () => { original.call(this, key, value); }); const handle = useThreadStore.getState().handleAgentEvent; - handle({ type: "turnStarted", threadId: THREAD_ID, turnExecutionId: "second", sequence: 1, publicationId: "1" }); - handle({ type: "textDelta", threadId: THREAD_ID, turnExecutionId: "second", sequence: 2, publicationId: "2", delta: "Visible reply" }); - handle({ type: "ended", threadId: THREAD_ID, turnExecutionId: "second", sequence: 3, publicationId: "3", outcome: "completed" }); + handle({ type: "turnStarted", threadId: THREAD_ID, turnExecutionId: "second", publicationId: "1" }); + handle({ type: "textDelta", threadId: THREAD_ID, turnExecutionId: "second", publicationId: "2", delta: "Visible reply" }); + handle({ type: "ended", threadId: THREAD_ID, turnExecutionId: "second", publicationId: "3", outcome: "completed" }); expect(record()?.runtimePhase).toBe("completed"); expect(record()?.messages.map((message) => message.content)).toContain("Visible reply"); expect(useThreadStore.getState().runningThreadIds.has(THREAD_ID)).toBe(false); diff --git a/apps/web/src/features/conversation/messages/__tests__/ChatView.test.tsx b/apps/web/src/features/conversation/messages/__tests__/ChatView.test.tsx index 7e71b28b2..5f7da0d48 100644 --- a/apps/web/src/features/conversation/messages/__tests__/ChatView.test.tsx +++ b/apps/web/src/features/conversation/messages/__tests__/ChatView.test.tsx @@ -358,11 +358,20 @@ function disableAtomicSubscriptionTransport() { }); } +/** One canonical delta recovery that always counts as newer than a fresh replica. */ +function deltaRecovery(threadId: string) { + return { + mode: "delta" as const, + threadId, + from: { conversationRevision: 0, rosterRevision: 0 }, + through: { conversationRevision: 1, rosterRevision: 0 }, + events: [], + }; +} + /** Enable the atomic transport path and return its caller-boundary spy. */ function enableAtomicSubscriptionTransport() { const setThreadSubscriptions = vi.fn().mockResolvedValue({ - hydrationRequiredThreadIds: [], - replayedThrough: {}, canonicalRecoveries: [], }); Object.defineProperty(chatViewTransportMock, "setThreadSubscriptions", { @@ -1687,7 +1696,7 @@ describe("ChatView - Thread Title Double-Click Rename", () => { expect(new Set(input.threadIds).size).toBe(100); }); - it("sends observed cursors without reconciling cursor-only changes", async () => { + it("sends observed revisions without reconciling unchanged records", async () => { const setThreadSubscriptions = enableAtomicSubscriptionTransport(); const runningThreadIds = new Set(); const emptyRecord = createEmptyThreadRecord(); @@ -1697,7 +1706,6 @@ describe("ChatView - Thread Title Double-Click Rename", () => { ...emptyRecord.canonicalAgent, revision: { conversationRevision: 4, rosterRevision: 2 }, }, - lastAgentEventSequence: 7, }; setupWorkspaceMock(defaultWorkspaceState({ activeThreadId: "thread-1" })); chatViewThreadMockRef.current = defaultThreadState({ @@ -1711,12 +1719,11 @@ describe("ChatView - Thread Title Double-Click Rename", () => { await waitFor(() => { expect(setThreadSubscriptions).toHaveBeenCalledWith({ threadIds: ["thread-1"], - cursors: { "thread-1": 7 }, revisions: { "thread-1": { conversationRevision: 4, rosterRevision: 2 } }, }); }); - const secondRecord = { ...firstRecord, lastAgentEventSequence: 8 }; + const secondRecord = { ...firstRecord }; chatViewThreadMockRef.current = defaultThreadState({ currentThreadId: "thread-1", runningThreadIds, @@ -1739,8 +1746,6 @@ describe("ChatView - Thread Title Double-Click Rename", () => { }, }; const setThreadSubscriptions = vi.fn().mockResolvedValue({ - hydrationRequiredThreadIds: [], - replayedThrough: {}, canonicalRecoveries: [recovery], }); Object.defineProperty(chatViewTransportMock, "setThreadSubscriptions", { @@ -1887,10 +1892,10 @@ describe("ChatView - Thread Title Double-Click Rename", () => { expect(setThreadSubscriptions).toHaveBeenCalledTimes(2); }); - it("ignores stale atomic hydration responses before mutating cursor or residency state", async () => { - const requests: Array<(result: { hydrationRequiredThreadIds: string[] }) => void> = []; + it("ignores stale atomic recovery responses before mutating canonical or residency state", async () => { + const requests: Array<(result: { canonicalRecoveries: unknown[] }) => void> = []; const setThreadSubscriptions = vi.fn( - () => new Promise<{ hydrationRequiredThreadIds: string[] }>((resolve) => { + () => new Promise<{ canonicalRecoveries: unknown[] }>((resolve) => { requests.push(resolve); }), ); @@ -1904,17 +1909,18 @@ describe("ChatView - Thread Title Double-Click Rename", () => { await waitFor(() => expect(setThreadSubscriptions).toHaveBeenCalledTimes(1)); chatViewConnectionStatusRef.current = "reconnecting"; rerender(); - requests[0]?.({ hydrationRequiredThreadIds: ["thread-1"] }); + requests[0]?.({ canonicalRecoveries: [deltaRecovery("thread-1")] }); await new Promise((resolve) => setTimeout(resolve, 0)); expect(chatViewThreadStoreSetStateMock).not.toHaveBeenCalled(); + expect(chatViewApplyCanonicalRecoveriesMock).not.toHaveBeenCalled(); expect(chatViewResidencyMock.invalidateConversation).not.toHaveBeenCalled(); expect(chatViewResidencyMock.refresh).not.toHaveBeenCalled(); }); - it("swallows rejected atomic hydration refreshes", async () => { + it("swallows rejected atomic recovery refreshes", async () => { const setThreadSubscriptions = vi.fn().mockResolvedValue({ - hydrationRequiredThreadIds: ["thread-1"], + canonicalRecoveries: [deltaRecovery("thread-1")], }); Object.defineProperty(chatViewTransportMock, "setThreadSubscriptions", { configurable: true, @@ -1937,10 +1943,10 @@ describe("ChatView - Thread Title Double-Click Rename", () => { }); }); - it("ignores same-epoch stale atomic hydration responses after the target changes", async () => { - const requests: Array<(result: { hydrationRequiredThreadIds: string[] }) => void> = []; + it("ignores same-epoch stale atomic recovery responses after the target changes", async () => { + const requests: Array<(result: { canonicalRecoveries: unknown[] }) => void> = []; const setThreadSubscriptions = vi.fn( - () => new Promise<{ hydrationRequiredThreadIds: string[] }>((resolve) => { + () => new Promise<{ canonicalRecoveries: unknown[] }>((resolve) => { requests.push(resolve); }), ); @@ -1951,11 +1957,7 @@ describe("ChatView - Thread Title Double-Click Rename", () => { }); const thread1 = makeThread({ id: "thread-1", title: "Thread 1" }); const thread2 = makeThread({ id: "thread-2", title: "Thread 2", status: "active" }); - const record = { - ...createEmptyThreadRecord(), - lastAgentEventEpoch: "epoch-a", - lastAgentEventSequence: 7, - }; + const record = createEmptyThreadRecord(); setupWorkspaceMock(defaultWorkspaceState({ activeThreadId: thread1.id, threads: [thread1, thread2], @@ -1978,15 +1980,16 @@ describe("ChatView - Thread Title Double-Click Rename", () => { rerender(); expect(setThreadSubscriptions).toHaveBeenCalledTimes(1); - requests[0]?.({ hydrationRequiredThreadIds: [thread1.id] }); + requests[0]?.({ canonicalRecoveries: [deltaRecovery(thread1.id)] }); await new Promise((resolve) => setTimeout(resolve, 0)); expect(chatViewThreadStoreSetStateMock).not.toHaveBeenCalled(); + expect(chatViewApplyCanonicalRecoveriesMock).not.toHaveBeenCalled(); expect(chatViewResidencyMock.invalidateConversation).not.toHaveBeenCalled(); expect(chatViewResidencyMock.refresh).not.toHaveBeenCalled(); await waitFor(() => expect(setThreadSubscriptions).toHaveBeenCalledTimes(2)); - requests[1]?.({ hydrationRequiredThreadIds: [] }); + requests[1]?.({ canonicalRecoveries: [] }); await new Promise((resolve) => setTimeout(resolve, 0)); expect(setThreadSubscriptions).toHaveBeenCalledTimes(2); }); diff --git a/apps/web/src/features/conversation/subscriptions/useThreadSubscriptionReconciler.ts b/apps/web/src/features/conversation/subscriptions/useThreadSubscriptionReconciler.ts index 71e6ba747..55841da8a 100644 --- a/apps/web/src/features/conversation/subscriptions/useThreadSubscriptionReconciler.ts +++ b/apps/web/src/features/conversation/subscriptions/useThreadSubscriptionReconciler.ts @@ -38,7 +38,6 @@ type SubscriptionRefs = { type AtomicSubscriptionInput = { readonly input: SetThreadSubscriptionsInput; readonly revisions: NonNullable; - readonly cursorAuthority: Map; }; type AtomicRequestContext = { @@ -47,7 +46,6 @@ type AtomicRequestContext = { readonly requestId: number; readonly sentThreadIds: Set; readonly revisions: NonNullable; - readonly cursorAuthority: Map; }; type SubscriptionReconcileContext = { @@ -104,23 +102,11 @@ function hasCanonicalRecovery(threadIds: readonly string[]): boolean { } function atomicSubscriptionInput(threadIds: string[]): AtomicSubscriptionInput { - const cursors: NonNullable = {}; const revisions: NonNullable = {}; - const cursorAuthority = new Map(); for (const threadId of threadIds) { - const record = readThreadRecord(threadId); - revisions[threadId] = record.canonicalAgent.revision; - const sequence = record.lastAgentEventSequence; - if (typeof sequence !== "number" || sequence <= 0) continue; - cursorAuthority.set(threadId, { epoch: record.lastAgentEventEpoch, sequence }); - cursors[threadId] = record.lastAgentEventEpoch - ? { epoch: record.lastAgentEventEpoch, sequence } - : sequence; + revisions[threadId] = readThreadRecord(threadId).canonicalAgent.revision; } - const input = Object.keys(cursors).length > 0 - ? { threadIds, cursors, revisions } - : { threadIds, revisions }; - return { input, revisions, cursorAuthority }; + return { input: { threadIds, revisions }, revisions }; } function createAtomicRequest( @@ -140,7 +126,6 @@ function createAtomicRequest( requestId, sentThreadIds: new Set(threadIds), revisions: subscription.revisions, - cursorAuthority: subscription.cursorAuthority, }; } @@ -185,51 +170,6 @@ function applyCanonicalRecovery( if (recovery.threadId === activeThreadId) refreshConversation(activeThreadId); } -function recordOwnsCursor( - record: ReturnType, - expected: { epoch?: string; sequence?: number } | undefined, -): boolean { - if (expected?.epoch) { - return record.lastAgentEventEpoch === expected.epoch - && record.lastAgentEventSequence === expected.sequence; - } - return record.lastAgentEventEpoch === undefined - && record.lastAgentEventSequence === expected?.sequence; -} - -function clearHydrationCursor(threadId: string, expected: { epoch?: string; sequence?: number } | undefined): void { - useThreadStore.setState((state) => { - const record = state.records.get(threadId); - if (!record || !recordOwnsCursor(record, expected)) return state; - const records = new Map(state.records); - records.set(threadId, { - ...record, - lastAgentEventEpoch: undefined, - lastAgentEventSequence: undefined, - }); - return { records }; - }); -} - -function applyHydrationRequired( - threadIds: readonly string[], - request: AtomicRequestContext, - context: SubscriptionReconcileContext, -): boolean { - const residency = getConversationResidency(); - for (const threadId of threadIds) { - if (!atomicResponseIsCurrent(context.refs, request)) return false; - clearHydrationCursor(threadId, request.cursorAuthority.get(threadId)); - if (!atomicResponseIsCurrent(context.refs, request)) return false; - residency.invalidateConversation(threadId); - if (threadId === context.activeThreadId && context.runningThreadIds.has(threadId)) { - if (!atomicResponseIsCurrent(context.refs, request)) return false; - refreshConversation(threadId); - } - } - return true; -} - function applyAtomicResponse( result: SetThreadSubscriptionsResult, request: AtomicRequestContext, @@ -243,7 +183,6 @@ function applyAtomicResponse( for (const recovery of canonicalRecoveries) { applyCanonicalRecovery(recovery, request.revisions, context.activeThreadId); } - if (!applyHydrationRequired(result.hydrationRequiredThreadIds, request, context)) return; if (!atomicResponseIsCurrent(context.refs, request)) return; context.refs.confirmedThreadIds.current = request.sentThreadIds; context.refs.retryAttempt.current = 0; diff --git a/apps/web/src/features/projects/state/__tests__/workspace-behavior.test.ts b/apps/web/src/features/projects/state/__tests__/workspace-behavior.test.ts index 509d97260..8a6fe7845 100644 --- a/apps/web/src/features/projects/state/__tests__/workspace-behavior.test.ts +++ b/apps/web/src/features/projects/state/__tests__/workspace-behavior.test.ts @@ -2311,8 +2311,6 @@ describe("Workspace Behavior", () => { currentTurnResponseKey: "authoritative-response", streaming: "authoritative response", toolCalls: [persistedTool], - lastAgentEventSequence: 4, - lastAgentEventEpoch: "server-epoch", }, ), })); @@ -2330,8 +2328,6 @@ describe("Workspace Behavior", () => { expect(record?.streaming).toBe("authoritative response"); expect(record?.toolCalls).toEqual([persistedTool]); expect(record?.toolCalls).not.toContainEqual({ id: "placeholder-tool" }); - expect(record?.lastAgentEventSequence).toBe(4); - expect(record?.lastAgentEventEpoch).toBe("server-epoch"); expect(useThreadStore.getState().runningThreadIds.has("persisted-authoritative")).toBe(true); expect(useThreadStore.getState().records.has(placeholderId)).toBe(false); }); diff --git a/apps/web/src/stores/thread-record.ts b/apps/web/src/stores/thread-record.ts index 48b88d0e8..9352fa8a4 100644 --- a/apps/web/src/stores/thread-record.ts +++ b/apps/web/src/stores/thread-record.ts @@ -132,10 +132,6 @@ export interface ThreadRecord { serverMessageIds: Record; narrativeByMessage: ThreadNarrativeByMessage; answeredPlanMessageIds: Set; - /** Highest positive server-assigned agent-event sequence observed for this thread. */ - lastAgentEventSequence?: number; - /** Server-process epoch paired with {@link lastAgentEventSequence}. */ - lastAgentEventEpoch?: string; error: string | null; streaming: string; diff --git a/apps/web/src/stores/thread-store/__tests__/stable-agent-event-publications.test.ts b/apps/web/src/stores/thread-store/__tests__/stable-agent-event-publications.test.ts index 01efb4692..517d5c022 100644 --- a/apps/web/src/stores/thread-store/__tests__/stable-agent-event-publications.test.ts +++ b/apps/web/src/stores/thread-store/__tests__/stable-agent-event-publications.test.ts @@ -20,15 +20,13 @@ function storage(): Pick { } describe("stable AgentEvent publications", () => { - it("rejects a publish-then-crash replay after recreating the client and changing server epoch", () => { + it("rejects a publish-then-crash replay after recreating the client", () => { const shared = storage(); const firstClient = new StableAgentEventPublications(() => shared); const published = event(7); - expect(firstClient.accept({ ...published, epoch: "00000000-0000-4000-8000-000000000004" })) - .toBe(true); + expect(firstClient.accept(published)).toBe(true); const reloadedClient = new StableAgentEventPublications(() => shared); - expect(reloadedClient.accept({ ...published, epoch: "00000000-0000-4000-8000-000000000005" })) - .toBe(false); + expect(reloadedClient.accept(published)).toBe(false); expect(reloadedClient.accept(event(8))).toBe(true); expect(reloadedClient.accept(event(7))).toBe(false); }); diff --git a/apps/web/src/stores/thread-store/agent-event-preflight.ts b/apps/web/src/stores/thread-store/agent-event-preflight.ts index f5df6ca2c..8c638ebde 100644 --- a/apps/web/src/stores/thread-store/agent-event-preflight.ts +++ b/apps/web/src/stores/thread-store/agent-event-preflight.ts @@ -9,7 +9,7 @@ export interface AgentEventRuntime { runtimeRecord: ThreadRecord; } -/** Handler signature for one validated and sequenced agent event type. */ +/** Handler signature for one validated agent event type. */ export type AgentEventHandler = ( event: Extract, runtime: AgentEventRuntime, @@ -20,7 +20,7 @@ export type AgentEventHandlerTable = { [Type in AgentEvent["type"]]: AgentEventHandler; }; -/** Narrow store operations needed to validate and sequence one agent event. */ +/** Narrow store operations needed to validate one agent event. */ export interface AgentEventPreflightContext { clearApiRetry: (threadId: string) => void; acceptPublication?: (event: AgentEvent) => boolean; @@ -32,9 +32,7 @@ export interface AgentEventPreflightContext { invalidateDeferredNarrativeEvents: (threadId: string) => void; invalidatePermissionSnapshots: (threadId: string) => void; isDisplayConversationLeased: (threadId: string) => boolean; - patchRecord: (threadId: string, patch: Partial) => void; promoteDeferredNarrativeEvents: (threadId: string) => void; - recordBackgroundEventDropped: (threadId: string) => void; scheduleDeferredNarrativeCleanup: (threadId: string) => void; } @@ -74,47 +72,6 @@ function acceptsExecution( ); } -function incomingSequence(event: AgentEvent): number | undefined { - if (typeof event.sequence !== "number") return undefined; - return event.sequence > 0 ? event.sequence : undefined; -} - -function isStaleSequence( - record: ThreadRecord, - sequence: number, - epoch: string | undefined, -): boolean { - if (epoch !== undefined && epoch !== record.lastAgentEventEpoch) return false; - const lastSequence = record.lastAgentEventSequence; - return lastSequence !== undefined && sequence <= lastSequence; -} - -function recordSequence( - context: AgentEventPreflightContext, - event: AgentEvent, - sequence: number, - epoch: string | undefined, -): void { - context.patchRecord(event.threadId, { - lastAgentEventSequence: sequence, - ...(epoch !== undefined ? { lastAgentEventEpoch: epoch } : {}), - }); -} - -function acceptsSequence(context: AgentEventPreflightContext, event: AgentEvent): boolean { - const sequence = incomingSequence(event); - if (sequence === undefined) return true; - const epoch = typeof event.epoch === "string" ? event.epoch : undefined; - if (isStaleSequence(context.getRecord(event.threadId), sequence, epoch)) { - if (context.getCurrentThreadId() !== event.threadId) { - context.recordBackgroundEventDropped(event.threadId); - } - return false; - } - recordSequence(context, event, sequence, epoch); - return true; -} - function isLifecycleExit(event: AgentEvent): boolean { return event.type === "turnComplete" || event.type === "ended" || event.type === "error"; } @@ -187,7 +144,6 @@ export function prepareAgentEvent( const runtimeActive = isRuntimeActive(runtimeRecord, runningThreadIds, event.threadId); const incomingExecutionId = eventExecutionId(event); if (!acceptsExecution(event, runtimeRecord, incomingExecutionId)) return null; - if (!acceptsSequence(context, event)) return null; if (context.acceptPublication && !context.acceptPublication(event)) return null; const currentThreadId = context.getCurrentThreadId(); diff --git a/apps/web/src/stores/thread-store/runtime.ts b/apps/web/src/stores/thread-store/runtime.ts index 3c9b9310a..6bcec742e 100644 --- a/apps/web/src/stores/thread-store/runtime.ts +++ b/apps/web/src/stores/thread-store/runtime.ts @@ -43,17 +43,6 @@ function mergePendingPersistMessageIds( ])]; } -function applyPersistedSequence( - patch: Partial, - persisted: ThreadRecord, - persistedExists: boolean, -): void { - const sequence = persistedExists ? (persisted.lastAgentEventSequence ?? 0) : 0; - if (sequence === 0) return; - patch.lastAgentEventSequence = sequence; - patch.lastAgentEventEpoch = persisted.lastAgentEventEpoch; -} - function runtimeIdentityPatch( placeholder: ThreadRecord, persisted: ThreadRecord, @@ -111,7 +100,6 @@ function createRuntimePatch( ...runtimeIdentityPatch(placeholder, persisted, placeholderRunning), ...runtimeNarrativePatch(placeholder, persisted, persistedId, createTurnResponseKey), }; - applyPersistedSequence(patch, persisted, persistedExists); if (!persistedExists || persisted.error === null) patch.error = placeholder.error; return patch; } diff --git a/apps/web/src/stores/threadStore.ts b/apps/web/src/stores/threadStore.ts index 3e9de01c2..b0b71c9eb 100644 --- a/apps/web/src/stores/threadStore.ts +++ b/apps/web/src/stores/threadStore.ts @@ -26,7 +26,6 @@ import { findModelById } from "@/lib/model-registry"; import { resolveContextWindow } from "@/lib/resolve-context-window"; import { useSettingsStore } from "./settingsStore"; import { createConversationResidency, registerConversationResidency } from "@/features/conversation/residency/conversation-residency"; -import { recordBackgroundEventDropped } from "@/lib/thread-switch-telemetry"; import { clearPendingTurnPersistMessage, projectTurnResponse, @@ -125,8 +124,8 @@ function batchTouchesParentNarrative(events: readonly CanonicalAgentEventEnvelop } /** - * Canonical envelopes carrying numbered renderer-facing publications replay through the same - * dispatch as the agent.event copy; the publication cursor accepts whichever copy arrives first. + * Canonical envelopes carrying numbered renderer-facing publications feed the shared + * agent-event projection; the publication cursor drops duplicates across replays. */ function dispatchCanonicalPublications(events: readonly CanonicalAgentEventEnvelope[]): void { const handle = useThreadStore.getState().handleAgentEvent; @@ -3399,8 +3398,6 @@ export const useThreadStore = create((zustandSet, get) => { latestTurnWithChanges: null, serverMessageIds: {}, narrativeByMessage: {}, - lastAgentEventEpoch: undefined, - lastAgentEventSequence: undefined, }), })); } @@ -3831,9 +3828,7 @@ export const useThreadStore = create((zustandSet, get) => { invalidateDeferredNarrativeEvents, invalidatePermissionSnapshots: (id) => threadHydrator.invalidatePermissionSnapshots(id), isDisplayConversationLeased: (id) => conversationResidency.isDisplayConversationLeased(id), - patchRecord: patchRec, promoteDeferredNarrativeEvents, - recordBackgroundEventDropped, scheduleDeferredNarrativeCleanup, }, event); if (!runtime) return; diff --git a/apps/web/src/transport/ws-events.test.ts b/apps/web/src/transport/ws-events.test.ts index c14c53a1a..8ba2bc3c3 100644 --- a/apps/web/src/transport/ws-events.test.ts +++ b/apps/web/src/transport/ws-events.test.ts @@ -1,5 +1,5 @@ import { afterEach, beforeEach, describe, expect, it, vi } from "vitest"; -import type { AgentEvent, PermissionRequest, ThreadStartup } from "@mcode/contracts"; +import type { PermissionRequest, ThreadStartup } from "@mcode/contracts"; import type { Thread } from "@/transport"; vi.mock("@/transport", () => ({ @@ -288,38 +288,6 @@ describe("ws-events provider.catalogChanged", () => { }); }); -describe("ws-events agent.event", () => { - afterEach(() => { - stopPushListeners(); - vi.restoreAllMocks(); - }); - - it("drops malformed data before it reaches the thread store", () => { - const handleAgentEvent = vi.spyOn(useThreadStore.getState(), "handleAgentEvent"); - startPushListeners(); - - pushEmitter.emit("agent.event", { type: "message", threadId: 42, content: "invalid" }); - - expect(handleAgentEvent).not.toHaveBeenCalled(); - }); - - it("forwards a valid parsed event exactly once", () => { - const event = { - type: "message", - threadId: "thread-1", - content: "valid", - tokens: null, - } satisfies AgentEvent; - const handleAgentEvent = vi.spyOn(useThreadStore.getState(), "handleAgentEvent"); - startPushListeners(); - - pushEmitter.emit("agent.event", event); - - expect(handleAgentEvent).toHaveBeenCalledOnce(); - expect(handleAgentEvent).toHaveBeenCalledWith(event); - }); -}); - describe("ws-events permission.request", () => { afterEach(() => { stopPushListeners(); @@ -335,19 +303,6 @@ describe("ws-events permission.request", () => { } satisfies PermissionRequest; startPushListeners(); - pushEmitter.emit("agent.event", { - type: "system", - threadId: request.threadId, - subtype: "approval.review.manual-required", - message: "Manual approval is required before Codex can continue.", - systemNotice: { - kind: "diagnostic", - presentation: "timeline", - scope: "turn", - sessionId: "notice-session", - noticeKey: "approval-review-manual-required", - }, - } satisfies AgentEvent); expect(buildVolatileItems([], undefined, undefined, undefined, useThreadStore.getState().records.get(request.threadId)?.permissions)) .not.toContainEqual(expect.objectContaining({ type: "permission-request" })); pushEmitter.emit("permission.request", request); diff --git a/apps/web/src/transport/ws-events.ts b/apps/web/src/transport/ws-events.ts index 2a8c4f556..3af3606d5 100644 --- a/apps/web/src/transport/ws-events.ts +++ b/apps/web/src/transport/ws-events.ts @@ -134,7 +134,7 @@ function handleTerminalData(data: unknown): void { * appropriate Zustand stores. Call once at app startup. * * Push channels handled: - * - `agent.event` -- agent stream events forwarded to threadStore + * - `agent.canonical` -- canonical semantic batches forwarded to threadStore * - `terminal.data` -- PTY output forwarded to xterm via ptyDataRegistry * - `terminal.exit` -- PTY exit forwarded via ptyDataRegistry * - Reconnect-gap banners are emitted from ws-transport after `terminal.reattach` RPC @@ -170,17 +170,6 @@ export function startPushListeners(): void { // Guard against double-init stopPushListeners(); - const handleAgentEvent = useThreadStore.getState().handleAgentEvent; - - // agent.event: the server wraps each sidecar event with { threadId, type, ... } - unsubs.push( - pushEmitter.on("agent.event", (data) => { - const parsed = WS_CHANNELS["agent.event"].safeParse(data); - if (!parsed.success) return; - handleAgentEvent(parsed.data); - }), - ); - unsubs.push( pushEmitter.on("agent.canonical", (data) => { const parsed = WS_CHANNELS["agent.canonical"].safeParse(data); diff --git a/packages/contracts/src/events/agent-event.ts b/packages/contracts/src/events/agent-event.ts index bd2a1829c..fce18d701 100644 --- a/packages/contracts/src/events/agent-event.ts +++ b/packages/contracts/src/events/agent-event.ts @@ -374,21 +374,15 @@ const AgentEventPayloadSchema = z.discriminatedUnion("type", [ }), ]); -/** Cryptographic identity for one server process event stream. */ -export const AgentEventEpochSchema = z.string().uuid(); - -/** Optional server-assigned per-thread ordering metadata. */ +/** Canonical publication metadata stamped by the server before broadcast. */ const AgentEventSequenceSchema = z.object({ - sequence: z.number().int().positive().optional(), - /** Server-process epoch paired with `sequence`; absent for legacy events. */ - epoch: AgentEventEpochSchema.optional(), /** Mcode-owned execution identity for the turn that produced this event. */ turnExecutionId: z.string().uuid().optional(), /** Durable semantic receipt identity; survives a server event-epoch change. */ publicationId: z.string().regex(/^[1-9]\d*$/).max(16).optional(), }); -/** Validated agent event payload, including an optional server ordering sequence. */ +/** Validated agent event payload, including canonical publication identity. */ export const AgentEventSchema = lazySchema(() => z.preprocess( rejectProviderRuntimeFields, diff --git a/packages/contracts/src/index.ts b/packages/contracts/src/index.ts index 7cdd5bdec..e16939e64 100644 --- a/packages/contracts/src/index.ts +++ b/packages/contracts/src/index.ts @@ -752,7 +752,7 @@ export type { } from "./models/browser-narrative.js"; // Events -export { AgentEventSchema, AgentEventType, AgentEventEpochSchema } from "./events/agent-event.js"; +export { AgentEventSchema, AgentEventType } from "./events/agent-event.js"; export type { AgentEvent } from "./events/agent-event.js"; export { CodexChildEvidenceSchema, diff --git a/packages/contracts/src/ws/__tests__/methods.test.ts b/packages/contracts/src/ws/__tests__/methods.test.ts index 27be9aea6..266047e8a 100644 --- a/packages/contracts/src/ws/__tests__/methods.test.ts +++ b/packages/contracts/src/ws/__tests__/methods.test.ts @@ -91,22 +91,6 @@ describe("thread switching WebSocket contracts", () => { expect(method.params.safeParse({ threadIds: ["thread-1", "thread-1"] }).success).toBe(false); expect(method.params.safeParse({ threadIds: Array.from({ length: MAX_THREAD_SUBSCRIPTIONS + 1 }, (_, index) => `thread-${index}`) }).success).toBe(false); expect(method.params.safeParse({ threadIds: [""] }).success).toBe(false); - expect(method.params.safeParse({ - threadIds: ["thread-1"], - cursors: { "thread-1": { epoch: "00000000-0000-4000-8000-000000000001", sequence: 4 } }, - }).success).toBe(true); - expect(method.params.safeParse({ - threadIds: ["thread-1"], - cursors: { "thread-1": 4 }, - }).success).toBe(true); - expect(method.params.safeParse({ - threadIds: ["thread-1"], - cursors: { "thread-1": -1 }, - }).success).toBe(false); - expect(method.params.safeParse({ - threadIds: ["thread-1"], - cursors: { "thread-1": 1.5 }, - }).success).toBe(false); expect(method.params.safeParse({ threadIds: ["thread-1"], revisions: { @@ -121,19 +105,18 @@ describe("thread switching WebSocket contracts", () => { }).success).toBe(false); expect(method.params.safeParse({ threadIds: ["thread-1"], - cursors: Object.fromEntries( - Array.from({ length: MAX_THREAD_SUBSCRIPTIONS + 1 }, (_, index) => [`thread-${index}`, index]), + revisions: Object.fromEntries( + Array.from({ length: MAX_THREAD_SUBSCRIPTIONS + 1 }, (_, index) => [ + `thread-${index}`, + { conversationRevision: 0, rosterRevision: 0 }, + ]), ), }).success).toBe(false); }); - it("parses structured hydration and replay results", () => { + it("parses structured canonical recovery results", () => { const method = WS_METHODS()["push.setThreadSubscriptions"]; - const result = { - hydrationRequiredThreadIds: ["thread-1"], - replayedThrough: { "thread-2": 12 }, - canonicalRecoveries: [], - }; + const result = { canonicalRecoveries: [] }; const parsed = method.result.safeParse(result); @@ -148,9 +131,7 @@ describe("thread switching WebSocket contracts", () => { const method = WS_METHODS()["push.setThreadSubscriptions"]; expect(method.result.safeParse({ - hydrationRequiredThreadIds: ["thread-1"], - replayedThrough: { "thread-2": 0 }, - canonicalRecoveries: [], + canonicalRecoveries: [{ threadId: "thread-1" }], }).success).toBe(false); }); diff --git a/packages/contracts/src/ws/channels.ts b/packages/contracts/src/ws/channels.ts index 1a7ad67cc..369058aee 100644 --- a/packages/contracts/src/ws/channels.ts +++ b/packages/contracts/src/ws/channels.ts @@ -1,5 +1,4 @@ import { z } from "zod"; -import { AgentEventSchema } from "../events/agent-event.js"; import { CanonicalAgentEventEnvelopeSchema } from "../compat/agent-model.js"; import { ThreadStatusSchema } from "../models/enums.js"; import { ThreadSchema } from "../models/thread.js"; @@ -69,7 +68,6 @@ export const WS_CHANNELS = { reason: z.enum(["credential-revoked", "provider-session-ended"]), }) .strict(), - "agent.event": AgentEventSchema(), /** Canonical semantic batches published only after their durable transaction commits. */ "agent.canonical": z.object({ threadId: z.string().min(1), diff --git a/packages/contracts/src/ws/methods.ts b/packages/contracts/src/ws/methods.ts index aa53ef3a5..c11b156e9 100644 --- a/packages/contracts/src/ws/methods.ts +++ b/packages/contracts/src/ws/methods.ts @@ -235,12 +235,6 @@ const ThreadSubscriptionIdSchema = z.string().trim().min(1).max(CONVERSATION_TAI /** Maximum identity length for a settled UUID or native Live comparison. */ const TURN_DIFF_COMPARISON_ID_MAX_LENGTH = 512; -/** Cursor identity for one server-process event stream. */ -const AgentEventCursorSchema = z.object({ - epoch: z.string().uuid(), - sequence: z.number().int().nonnegative(), -}); - /** Complete desired push subscription set for one WebSocket connection. */ export const SetThreadSubscriptionsSchema = lazySchema(() => z.object({ @@ -249,18 +243,6 @@ export const SetThreadSubscriptionsSchema = lazySchema(() => context.addIssue({ code: z.ZodIssueCode.custom, message: "threadIds must be unique" }); } }), - /** Last applied agent-event sequence per thread; omitted for legacy subscribe calls. */ - cursors: z.record( - ThreadSubscriptionIdSchema, - z.union([z.number().int().nonnegative(), AgentEventCursorSchema]), - ).superRefine((cursors, context) => { - if (Object.keys(cursors).length > MAX_THREAD_SUBSCRIPTIONS) { - context.addIssue({ - code: z.ZodIssueCode.custom, - message: `cursors must contain at most ${MAX_THREAD_SUBSCRIPTIONS} entries`, - }); - } - }).optional(), /** Last installed canonical revisions per desired thread. */ revisions: z.record( ThreadSubscriptionIdSchema, @@ -282,8 +264,6 @@ export type SetThreadSubscriptionsInput = z.infer z.object({ - hydrationRequiredThreadIds: z.array(ThreadSubscriptionIdSchema), - replayedThrough: z.record(ThreadSubscriptionIdSchema, z.number().int().positive()), canonicalRecoveries: z.array(CanonicalAgentReconnectRecoverySchema()) .max(MAX_THREAD_SUBSCRIPTIONS), }),