feat(agents): publish synthesized agent events through canonical

apps/server/src/features/agents/canonical/__tests__/canonical-agent-boundary-stub.ts

@@ -26,5 +26,15 @@ 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.
+    recordSynthesizedPublications: (_threadId: string, events: readonly Record<string, unknown>[]) =>
+      events.map((event, index) => ({
+        payload: {
+          type: "publication.recorded",
+          publicationId: String(index + 1),
+          event: { ...event, publicationId: String(index + 1) },
+        },
+      })),
   } as unknown as CanonicalAgentBoundary;
 }

apps/server/src/features/agents/canonical/__tests__/canonical-agent-boundary.test.ts

@@ -29,6 +29,7 @@ import { ACTIVE_TURN_WRITE_BATCH_LIMITS } from "../../../../runtime/persistence/
 import { PARENT_ASSISTANT_TEXT_RETAINED_LIMITS } from "../../turns/parent-assistant-text-checkpoint-service.js";
 import {
   CANONICAL_AGENT_CONTROL_EVENT_RESERVE,
+  CANONICAL_SYNTHESIZED_EXECUTION_ID,
   CanonicalAgentBoundary,
   type CanonicalAgentEventDraft,
 } from "../canonical-agent-boundary.js";
@@ -3317,4 +3318,39 @@ describe("CanonicalAgentBoundary", () => {
     expect(db.prepare("SELECT deleted_at FROM threads WHERE id = ?")
       .get(delegation.childThread.id)).toMatchObject({ deleted_at: null });
   });
+
+  it("records synthesized publications on the shared per-thread sequence and replays them", () => {
+    startCanonicalParent(sink, db);
+
+    const first = sink.recordSynthesizedPublications(THREAD_ID, [
+      { type: "goalUpdated", threadId: THREAD_ID },
+    ]);
+    const second = sink.recordSynthesizedPublications(THREAD_ID, [
+      { type: "goalCleared", threadId: THREAD_ID, reason: "cleared" },
+    ]);
+
+    // Writer and synthesized publications share the per-thread sequence head, so a
+    // synthesized event can never collide with a writer-allocated publicationId.
+    expect(first.map((envelope) => envelope.payload)).toEqual([
+      expect.objectContaining({ type: "publication.recorded", publicationId: "1" }),
+    ]);
+    expect(second.map((envelope) => envelope.payload)).toEqual([
+      expect.objectContaining({ type: "publication.recorded", publicationId: "2" }),
+    ]);
+
+    // Execution-correlation preflight drops events carrying a foreign execution id, so
+    // thread-scoped synthesized events must ship with a publicationId only.
+    const event = (first[0]!.payload as { event: Record<string, unknown> }).event;
+    expect(event).toMatchObject({ publicationId: "1", type: "goalUpdated" });
+    expect(event).not.toHaveProperty("turnExecutionId");
+
+    // Synthesized commits hold no execution checkpoint; their events still replay through
+    // the thread-scoped delta channel a reconnecting client consumes.
+    expect(sink.loadCheckpoint(CANONICAL_SYNTHESIZED_EXECUTION_ID)).toBeNull();
+    const recovery = sink.recoverThread(THREAD_ID, { conversationRevision: 0, rosterRevision: 0 });
+    const recovered = recovery.mode === "delta" ? recovery.events : [];
+    expect(recovered.filter((envelope) => envelope.payload.type === "publication.recorded"))
+      .toHaveLength(2);
+    expect(published).toHaveBeenCalled();
+  });
 });

apps/server/src/features/agents/canonical/canonical-agent-boundary.ts

@@ -110,6 +110,13 @@ import { ConversationDisplayMaterializer } from "../conversation/migrations/conv
 
 /** Capacity held back so volatile input cannot consume every semantic batch slot. */
 export const CANONICAL_AGENT_CONTROL_EVENT_RESERVE = 16;
+
+/**
+ * Reserved execution identity for `publication.recorded` envelopes synthesized outside a provider
+ * execution, such as goal broadcasts and admission failures. The sequence space is shared across
+ * all threads because each envelope is keyed by its thread-scoped publicationId.
+ */
+export const CANONICAL_SYNTHESIZED_EXECUTION_ID = "00000000-0000-4000-8000-000000000000";
 const CANONICAL_EXECUTION_DIAGNOSTIC_INDEX_CAPACITY = 128;
 const CANONICAL_DIAGNOSTIC_EXPORT_EVENT_CAPACITY = 1_024;
 
@@ -387,6 +394,7 @@ export class CanonicalAgentBoundary implements ParentTurnDurability, CodexCollab
   private readonly loadCommitTurnStatement: ReturnType<CanonicalAgentBoundary["buildLoadCommitTurnStatement"]>;
   private readonly loadCommitSequenceCollisionStatement: ReturnType<CanonicalAgentBoundary["buildLoadCommitSequenceCollisionStatement"]>;
   private readonly loadLastAcceptedSequenceStatement: ReturnType<CanonicalAgentBoundary["buildLoadLastAcceptedSequenceStatement"]>;
+  private readonly advanceLivePublicationHead: ReturnType<Database["prepare"]>;
   private readonly diagnostics = new CanonicalAgentDiagnostics();
   private readonly turnIdByExecution = new Map<string, string>();
   private readonly eventStore: CanonicalAgentEventStore;
@@ -410,6 +418,10 @@ export class CanonicalAgentBoundary implements ParentTurnDurability, CodexCollab
     this.loadCommitTurnStatement = this.buildLoadCommitTurnStatement();
     this.loadCommitSequenceCollisionStatement = this.buildLoadCommitSequenceCollisionStatement();
     this.loadLastAcceptedSequenceStatement = this.buildLoadLastAcceptedSequenceStatement();
+    // Synthesized publications share the writer's per-thread sequence so client dedup sees one order.
+    this.advanceLivePublicationHead = db.prepare(`INSERT INTO canonical_writer_live_publication_heads (thread_id, last_sequence)
+      VALUES (?, ?) ON CONFLICT(thread_id) DO UPDATE SET last_sequence = last_sequence + excluded.last_sequence
+      WHERE last_sequence <= 9007199254740991 - excluded.last_sequence RETURNING last_sequence`);
     this.displayMaterializer = new ConversationDisplayMaterializer(db);
     this.eventStore = new CanonicalAgentEventStore(db, {
       loadThread: (threadId) => this.loadThread(threadId),
@@ -427,6 +439,7 @@ export class CanonicalAgentBoundary implements ParentTurnDurability, CodexCollab
       persistState: (state, threadId, turnId, executionId, conversationChanged, events) =>
         this.persistState(state, threadId, turnId, executionId, conversationChanged, events),
       insertEvent: (event) => this.insertEvent(event),
+      lastAcceptedSequence: (threadId, executionId) => this.lastAcceptedSequence(threadId, executionId),
       persistCheckpoint: (checkpoint) => this.persistCheckpoint(this.toCanonicalCheckpoint(checkpoint)!),
       materializeItems: (events) => this.materializeRecordedItems(events),
       recover: (input, error) => this.recoverCommit(this.toCanonicalCommitInput(input), error),
@@ -506,6 +519,50 @@ export class CanonicalAgentBoundary implements ParentTurnDurability, CodexCollab
       : this.eventStore.commit(this.toEventStoreCommitInput(input));
   }
 
+  /**
+   * Record renderer-facing events that no provider execution publishes, such as goal broadcasts,
+   * as canonical publications. The drafts and their sequence advance commit in one transaction,
+   * and the committed envelopes publish through the canonical channel.
+   */
+  recordSynthesizedPublications(
+    threadId: string,
+    events: readonly Record<string, unknown>[],
+  ): readonly CanonicalAgentEventEnvelope[] {
+    if (events.length === 0) return [];
+    const thread = this.loadThread(threadId);
+    const result = this.commit({
+      threadId,
+      turnId: "",
+      executionId: CANONICAL_SYNTHESIZED_EXECUTION_ID,
+      phase: "synthesized",
+      persistCheckpoint: false,
+      events: () => {
+        const head = this.advanceLivePublicationHead.get(threadId, events.length) as
+          | { last_sequence: number }
+          | undefined;
+        if (!head) throw new Error(`Live publication sequence exhausted for thread ${threadId}`);
+        const first = head.last_sequence - events.length + 1;
+        return events.map((event, index): CanonicalAgentEventDraft => {
+          const publicationId = String(first + index);
+          return {
+            eventId: `publication:${threadId}:${publicationId}`,
+            routing: { threadId, executionId: CANONICAL_SYNTHESIZED_EXECUTION_ID },
+            sourceProviderId: thread?.providerId ?? "server",
+            sourceIdentities: [],
+            payload: {
+              type: "publication.recorded",
+              publicationId,
+              // The publicationId alone is the dedup key; stamping an execution id would break
+              // execution-correlation preflight for events that belong to no turn.
+              event: { ...event, publicationId },
+            },
+          };
+        });
+      },
+    });
+    return result.events;
+  }
+
   /** Commit inside the caller's transaction. The caller publishes result.events after that commit. */
   commitInsideTransaction(input: CanonicalAgentCommitInput): CanonicalAgentCommitResult {
     return serverWorkTrace

apps/server/src/features/agents/canonical/canonical-agent-event-store.ts

@@ -80,6 +80,7 @@ export interface CanonicalAgentEventStoreOperations {
     events: readonly CanonicalAgentEventEnvelope[],
   ): void;
   insertEvent(event: CanonicalAgentEventEnvelope): void;
+  lastAcceptedSequence(threadId: string, executionId: string): number;
   persistCheckpoint(checkpoint: CanonicalAgentEventStoreCheckpoint): void;
   materializeItems(events: readonly CanonicalAgentEventEnvelope[]): void;
   recover(input: CanonicalAgentEventStoreInput, error: unknown): CanonicalAgentCommitResult | null;
@@ -171,7 +172,9 @@ export class CanonicalAgentEventStore {
   ): CanonicalAgentCommitResult {
     const acceptedAt = new Date().toISOString();
     const candidateRevision = (context.thread?.conversationRevision ?? 0) + 1;
-    let acceptedSequence = context.checkpoint?.lastAcceptedSequence ?? 0;
+    // Checkpoint-less executions (synthesized publications) sequence from their own event rows.
+    let acceptedSequence = context.checkpoint?.lastAcceptedSequence
+      ?? this.operations.lastAcceptedSequence(input.threadId, input.executionId);
     let envelopes = newDrafts.map((draft) => {
       acceptedSequence += 1;
       return this.operations.createEnvelope(draft, acceptedSequence, candidateRevision, acceptedAt);

apps/server/src/features/agents/canonical/synthesized-agent-event-publication.ts

@@ -0,0 +1,19 @@
+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.
+ */
+export function publishSynthesizedAgentEvents(
+  canonical: Pick<CanonicalAgentBoundary, "recordSynthesizedPublications">,
+  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);
+    }
+  }
+}

apps/server/src/features/agents/commands/__tests__/goal-command.test.ts

@@ -77,7 +77,7 @@ describe("GoalCommand", () => {
   function build() {
     return new GoalCommand(
       { messageRepo, db },
-      broadcast as unknown as (channel: "agent.event", data: AgentEvent) => void,
+      (_threadId, events) => events.forEach((event) => broadcast("agent.event", event)),
     );
   }
 

apps/server/src/features/agents/commands/goal-command.ts

@@ -11,8 +11,8 @@ import {
 import type { MessageRepo } from "../conversation/persistence/message-repo.js";
 import type { CommandContext, CommandOutcome, McodeCommand } from "./command-router.js";
 
-/** Broadcast function shape used to push agent events to connected clients. */
-type BroadcastFn = (channel: "agent.event", data: AgentEvent) => void;
+/** Publishes synthesized agent events through the canonical stream and the legacy broadcast. */
+type PublishSynthesizedFn = (threadId: string, events: readonly AgentEvent[]) => void;
 
 /** Repositories and database handle the command needs to persist its rows. */
 interface GoalCommandDeps {
@@ -50,7 +50,7 @@ function asClaudeNativeGoalCommandProvider(
 export class GoalCommand implements McodeCommand {
   constructor(
     private readonly deps: GoalCommandDeps,
-    private readonly broadcast: BroadcastFn,
+    private readonly publishSynthesized: PublishSynthesizedFn,
   ) {}
 
   /** The session id the goal hook keys on, derived from the thread id. */
@@ -177,11 +177,11 @@ export class GoalCommand implements McodeCommand {
     threadId: string,
     reason: "cleared" | "rollback" | "completed",
   ): void {
-    this.broadcast("agent.event", {
+    this.publishSynthesized(threadId, [{
       type: AgentEventType.GoalCleared,
       threadId,
       reason,
-    } satisfies AgentEvent);
+    }]);
   }
 
   /**
@@ -204,12 +204,12 @@ export class GoalCommand implements McodeCommand {
       assistantMsgId = a.id;
     })();
 
-    this.broadcast("agent.event", {
+    this.publishSynthesized(threadId, [{
       type: AgentEventType.Message,
       threadId,
       content: replyText,
       tokens: null,
       messageId: assistantMsgId,
-    } satisfies AgentEvent);
+    }]);
   }
 }

apps/server/src/features/agents/goals/goal-lifecycle-service.ts

@@ -13,7 +13,8 @@ import {
   type ProviderId,
 } from "@mcode/contracts";
 import { logger } from "@mcode/shared";
-import { broadcast } from "../../../application/transport/push.js";
+import { CanonicalAgentBoundary } from "../canonical/canonical-agent-boundary.js";
+import { publishSynthesizedAgentEvents } from "../canonical/synthesized-agent-event-publication.js";
 import { ThreadRepo } from "../../thread-control/persistence/thread-repo.js";
 import { GoalCommand } from "../commands/goal-command.js";
 import type {
@@ -71,8 +72,10 @@ export class GoalLifecycleService {
     @inject(MessageRepo) messages: MessageRepo,
     @inject("Database") database: Database,
     @inject(AgentRuntimeCommandPort) private readonly runtime: AgentRuntimeCommandPort,
+    @inject(CanonicalAgentBoundary) private readonly canonical: Pick<CanonicalAgentBoundary, "recordSynthesizedPublications">,
   ) {
-    this.command = new GoalCommand({ messageRepo: messages, db: database }, broadcast);
+    this.command = new GoalCommand({ messageRepo: messages, db: database },
+      (threadId, events) => publishSynthesizedAgentEvents(canonical, threadId, events));
   }
 
   /** Route a goal command before the owning execution facade dispatches a provider turn. */
@@ -258,18 +261,17 @@ export class GoalLifecycleService {
       return;
     }
     const now = Date.now();
-    broadcast("agent.event", {
+    publishSynthesizedAgentEvents(this.canonical, event.threadId, [{
       type: AgentEventType.GoalUpdated,
       threadId: event.threadId,
       goal: { ...goal, status: "complete", timeUsedSeconds: this.elapsedSeconds(goal, now), updatedAt: now, controls: { ...goal.controls, canClear: false } },
-    } satisfies AgentEvent);
-    broadcast("agent.event", {
+    }, {
       type: AgentEventType.GoalCleared,
       threadId: event.threadId,
       providerId: goal.providerId ?? "unknown",
       reason: "completed",
       turnId: goal.turnId ?? null,
-    } satisfies AgentEvent);
+    }]);
   }
 
   private async refreshNativeGoal(
@@ -283,18 +285,17 @@ export class GoalLifecycleService {
     const result = await native.runNativeGoalCommand(sessionId, "/goal");
     if (result?.kind !== "empty") return;
     const now = Date.now();
-    broadcast("agent.event", {
+    publishSynthesizedAgentEvents(this.canonical, threadId, [{
       type: AgentEventType.GoalUpdated,
       threadId,
       goal: { ...before, status: "complete", timeUsedSeconds: this.elapsedSeconds(before, now), updatedAt: now, controls: { ...before.controls, canClear: false } },
-    } satisfies AgentEvent);
-    broadcast("agent.event", {
+    }, {
       type: AgentEventType.GoalCleared,
       threadId,
       providerId: before.providerId ?? "claude",
       reason: "completed",
       turnId: before.turnId ?? null,
-    } satisfies AgentEvent);
+    }]);
   }
 
   private async lookup(
@@ -354,19 +355,19 @@ export class GoalLifecycleService {
   }
 
   private broadcastGoalUpdated(threadId: string, goal: GoalState): void {
-    broadcast("agent.event", {
+    publishSynthesizedAgentEvents(this.canonical, threadId, [{
       type: AgentEventType.GoalUpdated,
       threadId,
       goal,
-    } satisfies AgentEvent);
+    }]);
   }
 
   private broadcastGoalCleared(threadId: string, reason: "rollback"): void {
-    broadcast("agent.event", {
+    publishSynthesizedAgentEvents(this.canonical, threadId, [{
       type: AgentEventType.GoalCleared,
       threadId,
       reason,
-    } satisfies AgentEvent);
+    }]);
   }
 
   private requireThread(threadId: string) {

apps/server/src/features/agents/orchestration/__tests__/agent-service-gate.test.ts

@@ -264,14 +264,16 @@ describe("AgentService.sendMessage — admission gates", () => {
     // Provider must NOT be resolved — no agent session started
     expect(resolveProvider).not.toHaveBeenCalled();
 
-    // A providerUnavailable event must have been broadcast on the agent.event channel
-    expect(broadcast).toHaveBeenCalledWith("agent.event", {
+    // A providerUnavailable event must have been broadcast on the agent.event channel.
+    // The canonical publication stamps a shared publicationId on the legacy copy.
+    expect(broadcast).toHaveBeenCalledWith("agent.event", expect.objectContaining({
       type: "providerUnavailable",
       threadId: THREAD_ID,
       providerId: "codex",
       reason: "disabled",
       configuredPath: undefined,
-    });
+      publicationId: expect.any(String),
+    }));
   });
 
   it.each(["failed", "stopped", "archived", "deleted"] as const)("rejects composer sends to %s threads before persistence", async (threadStatus) => {

apps/server/src/features/agents/orchestration/__tests__/agent-service-goal-command.test.ts

@@ -153,6 +153,7 @@ function buildService(db: Database) {
   } as unknown as ProviderAvailabilityService;
 
   const eventPublication = new AgentEventPublicationRegistry();
+  const canonicalStub = createCanonicalAgentBoundaryStub(db);
   const svc = createAgentServiceForTest(
     threadRepo,
     workspaceRepo,
@@ -176,7 +177,7 @@ function buildService(db: Database) {
       undefined,
       undefined,
       undefined,
-      createCanonicalAgentBoundaryStub(db),
+      canonicalStub,
       undefined,
       undefined,
       undefined,
@@ -195,6 +196,7 @@ function buildService(db: Database) {
     messageRepo,
     db,
     svc.runtimeAccess() as never,
+    canonicalStub,
   );
 
   // Provider adapters always stamp turn-scoped events with the active execution