feat(channels): add turn-adoption ack to channel inbound dispatch

src/auto-reply/get-reply-options.types.ts

@@ -91,6 +91,12 @@ export type GetReplyOptions = {
   imageOrder?: PromptImageOrderEntry[];
   /** Notifies when an agent run actually starts (useful for webchat command handling). */
   onAgentRunStart?: (runId: string) => void;
+  /**
+   * Called after the restart-recovery delivery-context persist attempt
+   * completes (context may be absent when source delivery is suppressed).
+   * Channels may complete ingress ownership here without waiting for settle.
+   */
+  onTurnAdopted?: () => void | Promise<void>;
   /** Shared lifecycle owner for the current user-turn transcript append. */
   userTurnTranscriptRecorder?: UserTurnTranscriptRecorder;
   onReplyStart?: () => Promise<void> | void;

src/auto-reply/reply/agent-runner.runreplyagent.e2e.test.ts

@@ -743,6 +743,105 @@ describe("runReplyAgent pending final delivery capture", () => {
     expect(stored.restartRecoveryDeliveryRunId).toBeUndefined();
   });
 
+  it("fires onTurnAdopted after restart recovery delivery context persist completes", async () => {
+    const sessionEntry: SessionEntry = {
+      sessionId: "session",
+      updatedAt: Date.now(),
+    };
+    const sessionStore = { main: sessionEntry };
+    const storePath = await createSessionStoreFile(sessionEntry);
+    const events: string[] = [];
+    const onTurnAdopted = vi.fn(async () => {
+      const storedAtAdoption = await readStoredMainSession(storePath);
+      expect(storedAtAdoption.restartRecoveryDeliveryContext).toEqual({
+        channel: "discord",
+        to: "channel:24680",
+        accountId: "work",
+        threadId: "1503645939964055592",
+      });
+      expect(typeof storedAtAdoption.restartRecoveryDeliveryRunId).toBe("string");
+      events.push("adopted");
+    });
+    state.runEmbeddedAgentMock.mockImplementationOnce(async () => {
+      events.push("agent-run");
+      return {
+        payloads: [{ text: "visible final" }],
+        meta: {},
+      };
+    });
+
+    const { run } = createMinimalRun({
+      opts: { onTurnAdopted },
+      sessionCtx: {
+        Provider: "discord",
+        OriginatingChannel: "discord",
+        OriginatingTo: "channel:24680",
+        AccountId: "work",
+        MessageSid: "1503645939964055592",
+        MessageThreadId: "1503645939964055592",
+      },
+      runOverrides: { messageProvider: "discord" },
+      sessionEntry,
+      sessionStore,
+      sessionKey: "main",
+      storePath,
+    });
+
+    await run();
+
+    expect(onTurnAdopted).toHaveBeenCalledOnce();
+    expect(events).toEqual(["adopted", "agent-run"]);
+  });
+
+  it("fires onTurnAdopted for suppressed-delivery runs before the agent turn", async () => {
+    const sessionEntry: SessionEntry = {
+      sessionId: "session",
+      updatedAt: Date.now(),
+    };
+    const sessionStore = { main: sessionEntry };
+    const storePath = await createSessionStoreFile(sessionEntry);
+    const events: string[] = [];
+    const onTurnAdopted = vi.fn(async () => {
+      const storedAtAdoption = await readStoredMainSession(storePath);
+      expect(storedAtAdoption.restartRecoveryDeliveryContext).toBeUndefined();
+      expect(storedAtAdoption.restartRecoveryDeliveryRunId).toBeUndefined();
+      events.push("adopted");
+    });
+    state.runEmbeddedAgentMock.mockImplementationOnce(async () => {
+      events.push("agent-run");
+      return {
+        payloads: [{ text: "ambient final" }],
+        meta: {},
+      };
+    });
+
+    const { run } = createMinimalRun({
+      opts: {
+        onTurnAdopted,
+        sourceReplyDeliveryMode: "message_tool_only",
+      },
+      sessionCtx: {
+        Provider: "telegram",
+        OriginatingChannel: "telegram",
+        OriginatingTo: "telegram:123",
+        AccountId: "default",
+        MessageSid: "42",
+        InboundEventKind: "room_event",
+      },
+      runOverrides: { messageProvider: "telegram" },
+      sessionEntry,
+      sessionStore,
+      sessionKey: "main",
+      storePath,
+      currentInboundEventKind: "room_event",
+    });
+
+    await run();
+
+    expect(onTurnAdopted).toHaveBeenCalledOnce();
+    expect(events).toEqual(["adopted", "agent-run"]);
+  });
+
   it("keeps heartbeat replies with real content in pending final delivery", async () => {
     const sessionEntry: SessionEntry = {
       sessionId: "session",

src/auto-reply/reply/agent-runner.ts

@@ -1728,6 +1728,10 @@ export async function runReplyAgent(params: {
     replyOperation.setPhase("running");
     const runStartedAt = Date.now();
     await persistRestartRecoveryDeliveryContext();
+    // Adoption marks run start and must never be spool-replayed (would re-run tools).
+    // Suppressed delivery has no recovery state to persist; crashed suppressed runs die
+    // silently. When a delivery context is resolvable, this still runs after its persist.
+    await opts?.onTurnAdopted?.();
     const runOutcome = await traceAgentPhase("reply.run_agent_turn", () =>
       runAgentTurnWithFallback({
         commandBody,

src/channels/turn/kernel.test.ts

@@ -1087,6 +1087,49 @@ describe("channel turn kernel", () => {
     expect(events).toEqual(["record", "afterRecord", "dispatch"]);
   });
 
+  it("threads onTurnAdopted into assembled reply options and fires after recovery persist attempt", async () => {
+    const events: string[] = [];
+    const onTurnAdopted = vi.fn(async () => {
+      events.push("adopted");
+    });
+    const dispatchReplyWithBufferedBlockDispatcher = vi.fn(
+      async (params: Parameters<DispatchReplyWithBufferedBlockDispatcher>[0]) => {
+        events.push("dispatch-start");
+        // Persist attempt completes before adoption (agent-runner contract).
+        events.push("recovery-persist");
+        await params.replyOptions?.onTurnAdopted?.();
+        events.push("settle");
+        return {
+          queuedFinal: true,
+          counts: { tool: 0, block: 0, final: 1 },
+        };
+      },
+    ) as DispatchReplyWithBufferedBlockDispatcher;
+
+    await dispatchAssembledChannelTurn({
+      cfg,
+      channel: "test",
+      agentId: "main",
+      routeSessionKey: "agent:main:test:peer",
+      storePath: "/tmp/sessions.json",
+      ctxPayload: createCtx(),
+      recordInboundSession: createRecordInboundSession(events),
+      dispatchReplyWithBufferedBlockDispatcher,
+      delivery: {
+        deliver: vi.fn(async () => undefined),
+      },
+      onTurnAdopted,
+    });
+
+    expect(onTurnAdopted).toHaveBeenCalledOnce();
+    expect(events).toEqual(["record", "dispatch-start", "recovery-persist", "adopted", "settle"]);
+    expect(dispatchReplyWithBufferedBlockDispatcher).toHaveBeenCalledWith(
+      expect.objectContaining({
+        replyOptions: expect.objectContaining({ onTurnAdopted }),
+      }),
+    );
+  });
+
   it("does not run afterRecord when session recording fails", async () => {
     const recordError = new Error("session store failed");
     const afterRecord = vi.fn();

src/channels/turn/kernel.ts

@@ -241,10 +241,11 @@ export const recordDroppedChannelInboundHistory = recordDroppedChannelTurnHistor
 function resolveAssembledReplyPipeline(
   params: AssembledChannelTurn,
 ): Pick<AssembledChannelTurn, "dispatcherOptions" | "replyOptions"> {
+  const onTurnAdopted = params.onTurnAdopted ?? params.replyOptions?.onTurnAdopted;
   if (!params.replyPipeline) {
     return {
       dispatcherOptions: params.dispatcherOptions,
-      replyOptions: params.replyOptions,
+      replyOptions: onTurnAdopted ? { ...params.replyOptions, onTurnAdopted } : params.replyOptions,
     };
   }
   const { onModelSelected, ...replyPipeline } = createChannelReplyPipeline({
@@ -262,6 +263,7 @@ function resolveAssembledReplyPipeline(
     replyOptions: {
       onModelSelected,
       ...params.replyOptions,
+      ...(onTurnAdopted ? { onTurnAdopted } : {}),
     },
   };
 }
@@ -762,20 +764,27 @@ export async function runChannelTurn<
   const admission = resolved.admission ?? preflightAdmission ?? ({ kind: "dispatch" } as const);
   let result: ChannelTurnResult<TDispatchResult>;
   try {
+    // Prepared runDispatch was assembled earlier and ignores late options (including onTurnAdopted).
     const dispatchResult = await dispatchResolvedChannelTurn(
-      admission.kind === "observeOnly"
+      "runDispatch" in resolved
         ? {
             ...resolved,
-            delivery: createNoopChannelEventDeliveryAdapter(),
+            ...(admission.kind === "observeOnly"
+              ? { delivery: createNoopChannelEventDeliveryAdapter() }
+              : {}),
             admission,
             log: params.log,
             messageId: input.id,
           }
         : {
             ...resolved,
+            ...(admission.kind === "observeOnly"
+              ? { delivery: createNoopChannelEventDeliveryAdapter() }
+              : {}),
             admission,
             log: params.log,
             messageId: input.id,
+            ...(params.onTurnAdopted ? { onTurnAdopted: params.onTurnAdopted } : {}),
           },
     );
     result = dispatchResult.dispatched ? { ...dispatchResult, admission } : dispatchResult;

src/channels/turn/types.ts

@@ -370,6 +370,11 @@ export type AssembledChannelTurn = {
   botLoopProtection?: ChannelBotLoopProtectionFacts;
   log?: (event: ChannelTurnLogEvent) => void;
   messageId?: string;
+  /**
+   * Observes turn adoption without waiting for settle. Threaded into
+   * replyOptions for the agent runner (after recovery persist attempt).
+   */
+  onTurnAdopted?: () => void | Promise<void>;
 };
 
 /** Channel turn with dispatch runner already prepared. */
@@ -473,4 +478,10 @@ export type RunChannelTurnParams<TRaw, TDispatchResult = DispatchFromConfigResul
   raw: TRaw;
   adapter: ChannelTurnAdapter<TRaw, TDispatchResult>;
   log?: (event: ChannelTurnLogEvent) => void;
+  /**
+   * Observes turn adoption without waiting for settle. Fired after the
+   * recovery-context persist attempt (context may be absent when source
+   * delivery is suppressed). Default callers still await full settle.
+   */
+  onTurnAdopted?: () => void | Promise<void>;
 };