refactor(agents)!: replace agent.event replay with canonical recovery

apps/server/src/application/bootstrap/server-bootstrap.ts

@@ -866,13 +866,8 @@ async function bootstrapServer(): Promise<void> {
       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);

apps/server/src/application/transport/__tests__/port-push.test.ts

@@ -19,18 +19,18 @@ 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", () => {
     const port = createMockPort();
     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
   });
 });

apps/server/src/application/transport/__tests__/push.test.ts

@@ -11,17 +11,39 @@ 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,
   resetTransportPayloadValidatorForTest,
   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<WebSocket> = {
     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,125 +223,21 @@ 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);
     addClient(ws);
     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", () => {

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<string, never>;
-  canonicalRecoveries: unknown[];
-} {
-  return { hydrationRequiredThreadIds: [], replayedThrough: {}, canonicalRecoveries: [] };
-}