fix: relay ACP sessions_spawn parent streaming (#34310) (thanks @vincentkoc) (#34310)

Co-authored-by: Onur Solmaz <2453968+osolmaz@users.noreply.github.com>

src/agents/acp-spawn.test.ts

@@ -33,6 +33,8 @@ const hoisted = vi.hoisted(() => {
   const sessionBindingListBySessionMock = vi.fn();
   const closeSessionMock = vi.fn();
   const initializeSessionMock = vi.fn();
+  const startAcpSpawnParentStreamRelayMock = vi.fn();
+  const resolveAcpSpawnStreamLogPathMock = vi.fn();
   const state = {
     cfg: createDefaultSpawnConfig(),
   };
@@ -45,6 +47,8 @@ const hoisted = vi.hoisted(() => {
     sessionBindingListBySessionMock,
     closeSessionMock,
     initializeSessionMock,
+    startAcpSpawnParentStreamRelayMock,
+    resolveAcpSpawnStreamLogPathMock,
     state,
   };
 });
@@ -100,6 +104,13 @@ vi.mock("../infra/outbound/session-binding-service.js", async (importOriginal) =
   };
 });
 
+vi.mock("./acp-spawn-parent-stream.js", () => ({
+  startAcpSpawnParentStreamRelay: (...args: unknown[]) =>
+    hoisted.startAcpSpawnParentStreamRelayMock(...args),
+  resolveAcpSpawnStreamLogPath: (...args: unknown[]) =>
+    hoisted.resolveAcpSpawnStreamLogPathMock(...args),
+}));
+
 const { spawnAcpDirect } = await import("./acp-spawn.js");
 
 function createSessionBindingCapabilities() {
@@ -132,6 +143,16 @@ function createSessionBinding(overrides?: Partial<SessionBindingRecord>): Sessio
   };
 }
 
+function createRelayHandle(overrides?: {
+  dispose?: ReturnType<typeof vi.fn>;
+  notifyStarted?: ReturnType<typeof vi.fn>;
+}) {
+  return {
+    dispose: overrides?.dispose ?? vi.fn(),
+    notifyStarted: overrides?.notifyStarted ?? vi.fn(),
+  };
+}
+
 function expectResolvedIntroTextInBindMetadata(): void {
   const callWithMetadata = hoisted.sessionBindingBindMock.mock.calls.find(
     (call: unknown[]) =>
@@ -236,6 +257,12 @@ describe("spawnAcpDirect", () => {
     hoisted.sessionBindingResolveByConversationMock.mockReset().mockReturnValue(null);
     hoisted.sessionBindingListBySessionMock.mockReset().mockReturnValue([]);
     hoisted.sessionBindingUnbindMock.mockReset().mockResolvedValue([]);
+    hoisted.startAcpSpawnParentStreamRelayMock
+      .mockReset()
+      .mockImplementation(() => createRelayHandle());
+    hoisted.resolveAcpSpawnStreamLogPathMock
+      .mockReset()
+      .mockReturnValue("/tmp/sess-main.acp-stream.jsonl");
   });
 
   it("spawns ACP session, binds a new thread, and dispatches initial task", async () => {
@@ -423,4 +450,147 @@ describe("spawnAcpDirect", () => {
     expect(hoisted.callGatewayMock).not.toHaveBeenCalled();
     expect(hoisted.initializeSessionMock).not.toHaveBeenCalled();
   });
+
+  it('streams ACP progress to parent when streamTo="parent"', async () => {
+    const firstHandle = createRelayHandle();
+    const secondHandle = createRelayHandle();
+    hoisted.startAcpSpawnParentStreamRelayMock
+      .mockReset()
+      .mockReturnValueOnce(firstHandle)
+      .mockReturnValueOnce(secondHandle);
+
+    const result = await spawnAcpDirect(
+      {
+        task: "Investigate flaky tests",
+        agentId: "codex",
+        streamTo: "parent",
+      },
+      {
+        agentSessionKey: "agent:main:main",
+        agentChannel: "discord",
+        agentAccountId: "default",
+        agentTo: "channel:parent-channel",
+      },
+    );
+
+    expect(result.status).toBe("accepted");
+    expect(result.streamLogPath).toBe("/tmp/sess-main.acp-stream.jsonl");
+    const agentCall = hoisted.callGatewayMock.mock.calls
+      .map((call: unknown[]) => call[0] as { method?: string; params?: Record<string, unknown> })
+      .find((request) => request.method === "agent");
+    const agentCallIndex = hoisted.callGatewayMock.mock.calls.findIndex(
+      (call: unknown[]) => (call[0] as { method?: string }).method === "agent",
+    );
+    const relayCallOrder = hoisted.startAcpSpawnParentStreamRelayMock.mock.invocationCallOrder[0];
+    const agentCallOrder = hoisted.callGatewayMock.mock.invocationCallOrder[agentCallIndex];
+    expect(agentCall?.params?.deliver).toBe(false);
+    expect(typeof relayCallOrder).toBe("number");
+    expect(typeof agentCallOrder).toBe("number");
+    expect(relayCallOrder < agentCallOrder).toBe(true);
+    expect(hoisted.startAcpSpawnParentStreamRelayMock).toHaveBeenCalledWith(
+      expect.objectContaining({
+        parentSessionKey: "agent:main:main",
+        agentId: "codex",
+        logPath: "/tmp/sess-main.acp-stream.jsonl",
+        emitStartNotice: false,
+      }),
+    );
+    const relayRuns = hoisted.startAcpSpawnParentStreamRelayMock.mock.calls.map(
+      (call: unknown[]) => (call[0] as { runId?: string }).runId,
+    );
+    expect(relayRuns).toContain(agentCall?.params?.idempotencyKey);
+    expect(relayRuns).toContain(result.runId);
+    expect(hoisted.resolveAcpSpawnStreamLogPathMock).toHaveBeenCalledWith({
+      childSessionKey: expect.stringMatching(/^agent:codex:acp:/),
+    });
+    expect(firstHandle.dispose).toHaveBeenCalledTimes(1);
+    expect(firstHandle.notifyStarted).not.toHaveBeenCalled();
+    expect(secondHandle.notifyStarted).toHaveBeenCalledTimes(1);
+  });
+
+  it("announces parent relay start only after successful child dispatch", async () => {
+    const firstHandle = createRelayHandle();
+    const secondHandle = createRelayHandle();
+    hoisted.startAcpSpawnParentStreamRelayMock
+      .mockReset()
+      .mockReturnValueOnce(firstHandle)
+      .mockReturnValueOnce(secondHandle);
+
+    const result = await spawnAcpDirect(
+      {
+        task: "Investigate flaky tests",
+        agentId: "codex",
+        streamTo: "parent",
+      },
+      {
+        agentSessionKey: "agent:main:main",
+      },
+    );
+
+    expect(result.status).toBe("accepted");
+    expect(firstHandle.notifyStarted).not.toHaveBeenCalled();
+    expect(secondHandle.notifyStarted).toHaveBeenCalledTimes(1);
+    const notifyOrder = secondHandle.notifyStarted.mock.invocationCallOrder;
+    const agentCallIndex = hoisted.callGatewayMock.mock.calls.findIndex(
+      (call: unknown[]) => (call[0] as { method?: string }).method === "agent",
+    );
+    const agentCallOrder = hoisted.callGatewayMock.mock.invocationCallOrder[agentCallIndex];
+    expect(typeof agentCallOrder).toBe("number");
+    expect(typeof notifyOrder[0]).toBe("number");
+    expect(notifyOrder[0] > agentCallOrder).toBe(true);
+  });
+
+  it("disposes pre-registered parent relay when initial ACP dispatch fails", async () => {
+    const relayHandle = createRelayHandle();
+    hoisted.startAcpSpawnParentStreamRelayMock.mockReturnValueOnce(relayHandle);
+    hoisted.callGatewayMock.mockImplementation(async (argsUnknown: unknown) => {
+      const args = argsUnknown as { method?: string };
+      if (args.method === "sessions.patch") {
+        return { ok: true };
+      }
+      if (args.method === "agent") {
+        throw new Error("agent dispatch failed");
+      }
+      if (args.method === "sessions.delete") {
+        return { ok: true };
+      }
+      return {};
+    });
+
+    const result = await spawnAcpDirect(
+      {
+        task: "Investigate flaky tests",
+        agentId: "codex",
+        streamTo: "parent",
+      },
+      {
+        agentSessionKey: "agent:main:main",
+      },
+    );
+
+    expect(result.status).toBe("error");
+    expect(result.error).toContain("agent dispatch failed");
+    expect(relayHandle.dispose).toHaveBeenCalledTimes(1);
+    expect(relayHandle.notifyStarted).not.toHaveBeenCalled();
+  });
+
+  it('rejects streamTo="parent" without requester session context', async () => {
+    const result = await spawnAcpDirect(
+      {
+        task: "Investigate flaky tests",
+        agentId: "codex",
+        streamTo: "parent",
+      },
+      {
+        agentChannel: "discord",
+        agentAccountId: "default",
+        agentTo: "channel:parent-channel",
+      },
+    );
+
+    expect(result.status).toBe("error");
+    expect(result.error).toContain('streamTo="parent"');
+    expect(hoisted.callGatewayMock).not.toHaveBeenCalled();
+    expect(hoisted.startAcpSpawnParentStreamRelayMock).not.toHaveBeenCalled();
+  });
 });