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 b2a7526f8..f0e4066ba 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 @@ -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[]) => + events.map((event, index) => ({ + payload: { + type: "publication.recorded", + publicationId: String(index + 1), + event: { ...event, publicationId: String(index + 1) }, + }, + })), } as unknown as CanonicalAgentBoundary; } diff --git a/apps/server/src/features/agents/canonical/__tests__/canonical-agent-boundary.test.ts b/apps/server/src/features/agents/canonical/__tests__/canonical-agent-boundary.test.ts index 54620ab03..bd844edcc 100644 --- a/apps/server/src/features/agents/canonical/__tests__/canonical-agent-boundary.test.ts +++ b/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 }).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(); + }); }); diff --git a/apps/server/src/features/agents/canonical/canonical-agent-boundary.ts b/apps/server/src/features/agents/canonical/canonical-agent-boundary.ts index 461220cfa..9605367c1 100644 --- a/apps/server/src/features/agents/canonical/canonical-agent-boundary.ts +++ b/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; private readonly loadCommitSequenceCollisionStatement: ReturnType; private readonly loadLastAcceptedSequenceStatement: ReturnType; + private readonly advanceLivePublicationHead: ReturnType; private readonly diagnostics = new CanonicalAgentDiagnostics(); private readonly turnIdByExecution = new Map(); 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[], + ): 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 diff --git a/apps/server/src/features/agents/canonical/canonical-agent-event-store.ts b/apps/server/src/features/agents/canonical/canonical-agent-event-store.ts index 6577b8595..8e4cec7e0 100644 --- a/apps/server/src/features/agents/canonical/canonical-agent-event-store.ts +++ b/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); 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 new file mode 100644 index 000000000..829aa4c7c --- /dev/null +++ b/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, + 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); + } + } +} diff --git a/apps/server/src/features/agents/commands/__tests__/goal-command.test.ts b/apps/server/src/features/agents/commands/__tests__/goal-command.test.ts index 034b5dea3..ca827459a 100644 --- a/apps/server/src/features/agents/commands/__tests__/goal-command.test.ts +++ b/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)), ); } diff --git a/apps/server/src/features/agents/commands/goal-command.ts b/apps/server/src/features/agents/commands/goal-command.ts index d816a7101..228c2bea7 100644 --- a/apps/server/src/features/agents/commands/goal-command.ts +++ b/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); + }]); } } diff --git a/apps/server/src/features/agents/goals/goal-lifecycle-service.ts b/apps/server/src/features/agents/goals/goal-lifecycle-service.ts index d19b30cfd..60147f23b 100644 --- a/apps/server/src/features/agents/goals/goal-lifecycle-service.ts +++ b/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, ) { - 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) { diff --git a/apps/server/src/features/agents/orchestration/__tests__/agent-service-gate.test.ts b/apps/server/src/features/agents/orchestration/__tests__/agent-service-gate.test.ts index 86817107c..acd80ad51 100644 --- a/apps/server/src/features/agents/orchestration/__tests__/agent-service-gate.test.ts +++ b/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) => { diff --git a/apps/server/src/features/agents/orchestration/__tests__/agent-service-goal-command.test.ts b/apps/server/src/features/agents/orchestration/__tests__/agent-service-goal-command.test.ts index 9e84ccb08..1536e90d0 100644 --- a/apps/server/src/features/agents/orchestration/__tests__/agent-service-goal-command.test.ts +++ b/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 diff --git a/apps/server/src/features/agents/orchestration/__tests__/agent-service-test-harness.ts b/apps/server/src/features/agents/orchestration/__tests__/agent-service-test-harness.ts index b32a5cd6f..0b3d4624e 100644 --- a/apps/server/src/features/agents/orchestration/__tests__/agent-service-test-harness.ts +++ b/apps/server/src/features/agents/orchestration/__tests__/agent-service-test-harness.ts @@ -245,6 +245,7 @@ export function createAgentServiceForTest( messageRepo, db, runtimeCommands, + parentDurability, ); const featureEffects = new TurnFeatureEffects( resolvedPlans, @@ -291,7 +292,7 @@ export function createAgentServiceForTest( testContainer.registerInstance(TURN_FILE_EFFECTS, fileEffects); testContainer.registerInstance(TURN_ADMISSION_DISPATCH_COORDINATOR, admissions); testContainer.registerInstance(TurnConversationProjectionService, conversationProjection); - testContainer.registerInstance(PostTerminalHookCompletionEffect, new PostTerminalHookCompletionEffect(hookExecutionRepo, finalizer)); + testContainer.registerInstance(PostTerminalHookCompletionEffect, new PostTerminalHookCompletionEffect(hookExecutionRepo, finalizer, parentDurability)); testContainer.registerInstance(ProviderSessionCursorPersistence, new ProviderSessionCursorPersistence(runtimePersistence, parentDurability)); testContainer.registerInstance(ThreadCreationCoordinator, new ThreadCreationCoordinator( threadRepo, diff --git a/apps/server/src/features/agents/turns/parent-turn-durability.ts b/apps/server/src/features/agents/turns/parent-turn-durability.ts index e7eb51453..2d6acb35f 100644 --- a/apps/server/src/features/agents/turns/parent-turn-durability.ts +++ b/apps/server/src/features/agents/turns/parent-turn-durability.ts @@ -2,6 +2,7 @@ import type { AgentEvent, AgentThread, AgentTurn, + CanonicalAgentEventEnvelope, Message, NarrativeEntry, ParentNarrativeRecoveryItem, @@ -108,6 +109,10 @@ export interface ParentTurnDurability { event: AgentEvent; terminal: boolean; }): void; + recordSynthesizedPublications( + threadId: string, + events: readonly Record[], + ): readonly CanonicalAgentEventEnvelope[]; } /** Injection token for parent-turn durability. */ diff --git a/apps/server/src/features/agents/turns/post-terminal-hook-completion-effect.ts b/apps/server/src/features/agents/turns/post-terminal-hook-completion-effect.ts index 180732689..c1e80526d 100644 --- a/apps/server/src/features/agents/turns/post-terminal-hook-completion-effect.ts +++ b/apps/server/src/features/agents/turns/post-terminal-hook-completion-effect.ts @@ -1,12 +1,12 @@ import { inject, injectable } from "tsyringe"; import { AgentEventType } from "@mcode/contracts"; -import type { AgentEvent } from "@mcode/contracts"; -import { broadcast } from "../../../application/transport/push.js"; import { HookExecutionRepo, type CreateHookExecutionInput, } from "../events/persistence/hook-execution-repo.js"; +import { CanonicalAgentBoundary } from "../canonical/canonical-agent-boundary.js"; +import { publishSynthesizedAgentEvents } from "../canonical/synthesized-agent-event-publication.js"; import { TURN_FINALIZER, TurnFinalizer } from "./turn-finalizer.js"; /** Schedules durable hook completion records after their parent turn has materialized. */ @@ -15,6 +15,7 @@ export class PostTerminalHookCompletionEffect { constructor( @inject(HookExecutionRepo) private readonly hooks: HookExecutionRepo, @inject(TURN_FINALIZER) private readonly finalizer: TurnFinalizer, + @inject(CanonicalAgentBoundary) private readonly canonical: Pick, ) {} /** Persist and publish a late hook only after the terminal turn projection verifies. */ @@ -37,7 +38,7 @@ export class PostTerminalHookCompletionEffect { const messageId = this.finalizer.getLastPersistedMessageId(threadId); if (!messageId) return; this.hooks.bulkCreate([{ ...hook, messageId }]); - broadcast("agent.event", { + publishSynthesizedAgentEvents(this.canonical, threadId, [{ type: AgentEventType.HookCompleted, threadId, hookName: hook.hookName, @@ -46,6 +47,6 @@ export class PostTerminalHookCompletionEffect { didBlock: hook.didBlock, persistedMessageId: messageId, persistedHookId: hook.id, - } satisfies AgentEvent); + }]); } } diff --git a/apps/server/src/features/agents/turns/turn-admission-dispatch-coordinator.ts b/apps/server/src/features/agents/turns/turn-admission-dispatch-coordinator.ts index d695c8aec..f41d2053b 100644 --- a/apps/server/src/features/agents/turns/turn-admission-dispatch-coordinator.ts +++ b/apps/server/src/features/agents/turns/turn-admission-dispatch-coordinator.ts @@ -22,6 +22,7 @@ import { import { logger, validateBranchName } from "@mcode/shared"; import { isExplicitMcodeThreadRequest } from "@mcode/thread-orchestration"; import { broadcast } from "../../../application/transport/push.js"; +import { publishSynthesizedAgentEvents } from "../canonical/synthesized-agent-event-publication.js"; import { AttachmentService } from "../../attachments/storage/attachment-service.js"; import { FileService } from "../../projects/files/file-service.js"; import { GitWorktreeService } from "../../projects/git/git-worktree-service.js"; @@ -493,13 +494,13 @@ export class TurnAdmissionDispatchCoordinator { private publishProviderUnavailable(threadId: string, providerId: ProviderId, error: unknown): void { if (!(error instanceof ProviderDisabledError) && !(error instanceof ProviderCliMissingError)) return; - broadcast("agent.event", { + publishSynthesizedAgentEvents(this.parentTurns, threadId, [{ type: AgentEventType.ProviderUnavailable, threadId, providerId, reason: error instanceof ProviderDisabledError ? "disabled" : "cli_missing", configuredPath: error instanceof ProviderCliMissingError ? error.configuredPath : undefined, - }); + }]); } private requireWorkspace(workspaceId: string): Exclude, null> { diff --git a/apps/server/src/features/agents/turns/turn-finalizer.ts b/apps/server/src/features/agents/turns/turn-finalizer.ts index 2bba7bf1b..81bcd0f08 100644 --- a/apps/server/src/features/agents/turns/turn-finalizer.ts +++ b/apps/server/src/features/agents/turns/turn-finalizer.ts @@ -34,6 +34,7 @@ import type { TurnOutcome } from "./turn-outcome.js"; import type { TurnFileTracker } from "./turn-file-tracker.js"; import type { TurnFileEffectSummary } from "@mcode/contracts"; import type { ParentTurnDurability } from "./parent-turn-durability.js"; +import { publishSynthesizedAgentEvents } from "../canonical/synthesized-agent-event-publication.js"; import type { ParentAssistantTextCheckpointService } from "./parent-assistant-text-checkpoint-service.js"; import { deriveTurnAssistantMessageId } from "./turn-assistant-message-id.js"; import type { TurnDiffService, SettleTurnDiff } from "./turn-diff-service.js"; @@ -704,14 +705,20 @@ export class TurnFinalizer { private broadcastMaterializedAssistant(threadId: string, materialized: MaterializedAssistantRow): void { if (!materialized.shouldBroadcast) return; - broadcast("agent.event", { + const event = { type: AgentEventType.Message, threadId, content: materialized.content, tokens: null, messageId: materialized.id, ...(materialized.attachments.length > 0 ? { attachments: materialized.attachments } : {}), - } satisfies AgentEvent); + } 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; + } + publishSynthesizedAgentEvents(this.canonicalSink, threadId, [event]); } private commitAssistantMaterialization(threadId: string): void { diff --git a/apps/web/src/stores/thread-store/stable-agent-event-publications.ts b/apps/web/src/stores/thread-store/stable-agent-event-publications.ts index 16b518063..c33410017 100644 --- a/apps/web/src/stores/thread-store/stable-agent-event-publications.ts +++ b/apps/web/src/stores/thread-store/stable-agent-event-publications.ts @@ -21,7 +21,8 @@ export class StableAgentEventPublications { /** Reserve a publication before applying its UI effects, retaining failed writes in memory. */ accept(event: AgentEvent): boolean { if (!event.publicationId) return true; - if (!event.turnExecutionId || publicationSequence(event.publicationId) === null) return false; + // Synthesized publications belong to no execution; the thread-scoped sequence is the dedup key. + if (publicationSequence(event.publicationId) === null) return false; try { return this.reserve(event.threadId, event.publicationId); } catch {