fix(cloud-agent): broadcast user prompt on codex cloud runs

packages/agent/src/adapters/codex/codex-agent.test.ts

@@ -60,20 +60,32 @@ describe("CodexAcpAgent", () => {
     vi.clearAllMocks();
   });
 
-  function createAgent(): CodexAcpAgent {
+  function createAgent(overrides: Partial<AgentSideConnection> = {}): {
+    agent: CodexAcpAgent;
+    client: AgentSideConnection & {
+      extNotification: ReturnType<typeof vi.fn>;
+      sessionUpdate: ReturnType<typeof vi.fn>;
+    };
+  } {
     const client = {
       extNotification: vi.fn(),
-    } as unknown as AgentSideConnection;
-
-    return new CodexAcpAgent(client, {
+      sessionUpdate: vi.fn(),
+      ...overrides,
+    } as unknown as AgentSideConnection & {
+      extNotification: ReturnType<typeof vi.fn>;
+      sessionUpdate: ReturnType<typeof vi.fn>;
+    };
+
+    const agent = new CodexAcpAgent(client, {
       codexProcessOptions: {
         cwd: process.cwd(),
       },
     });
+    return { agent, client };
   }
 
   it("applies the requested initial mode for a new session", async () => {
-    const agent = createAgent();
+    const { agent } = createAgent();
     mockCodexConnection.newSession.mockResolvedValue({
       sessionId: "session-1",
       modes: { currentModeId: "auto", availableModes: [] },
@@ -96,7 +108,7 @@ describe("CodexAcpAgent", () => {
   });
 
   it("preserves the live session mode when loading an existing session", async () => {
-    const agent = createAgent();
+    const { agent } = createAgent();
     mockCodexConnection.loadSession.mockResolvedValue({
       modes: { currentModeId: "read-only", availableModes: [] },
       configOptions: [],
@@ -114,4 +126,53 @@ describe("CodexAcpAgent", () => {
         .sessionState.permissionMode,
     ).toBe("read-only");
   });
+
+  it("broadcasts user prompt as user_message_chunk before delegating to codex-acp", async () => {
+    const { agent, client } = createAgent();
+    // Seed an active session so prompt() has the state it expects.
+    mockCodexConnection.newSession.mockResolvedValue({
+      sessionId: "session-1",
+      modes: { currentModeId: "auto", availableModes: [] },
+      configOptions: [],
+    } satisfies Partial<NewSessionResponse>);
+    await agent.newSession({
+      cwd: process.cwd(),
+    } as never);
+
+    const callOrder: string[] = [];
+    client.sessionUpdate.mockImplementation(async () => {
+      callOrder.push("sessionUpdate");
+    });
+    mockCodexConnection.prompt.mockImplementation(async () => {
+      callOrder.push("prompt");
+      return { stopReason: "end_turn" };
+    });
+
+    await agent.prompt({
+      sessionId: "session-1",
+      prompt: [
+        { type: "text", text: "first chunk" },
+        { type: "text", text: "second chunk" },
+      ],
+    } as never);
+
+    expect(client.sessionUpdate).toHaveBeenCalledTimes(2);
+    expect(client.sessionUpdate).toHaveBeenNthCalledWith(1, {
+      sessionId: "session-1",
+      update: {
+        sessionUpdate: "user_message_chunk",
+        content: { type: "text", text: "first chunk" },
+      },
+    });
+    expect(client.sessionUpdate).toHaveBeenNthCalledWith(2, {
+      sessionId: "session-1",
+      update: {
+        sessionUpdate: "user_message_chunk",
+        content: { type: "text", text: "second chunk" },
+      },
+    });
+    // Broadcast must land before the prompt reaches codex-acp so the user
+    // turn is persisted even if the underlying prompt fails.
+    expect(callOrder).toEqual(["sessionUpdate", "sessionUpdate", "prompt"]);
+  });
 });

packages/agent/src/adapters/codex/codex-agent.ts

@@ -369,6 +369,12 @@ export class CodexAcpAgent extends BaseAcpAgent {
     this.session.interruptReason = undefined;
     resetUsage(this.sessionState);
 
+    // codex-acp does not echo the user prompt back on the agent→client
+    // channel, so without this broadcast the tapped stream (persisted to S3
+    // and rendered by the PostHog web UI) never sees a user turn and only
+    // the assistant reply shows up. Mirrors ClaudeAcpAgent.broadcastUserMessage.
+    await this.broadcastUserMessage(params);
+
     const response = await this.codexConnection.prompt(params);
 
     // Usage is already accumulated via sessionUpdate notifications in
@@ -417,6 +423,20 @@ export class CodexAcpAgent extends BaseAcpAgent {
     });
   }
 
+  private async broadcastUserMessage(params: PromptRequest): Promise<void> {
+    for (const chunk of params.prompt) {
+      const notification = {
+        sessionId: params.sessionId,
+        update: {
+          sessionUpdate: "user_message_chunk" as const,
+          content: chunk,
+        },
+      };
+      await this.client.sessionUpdate(notification);
+      this.appendNotification(params.sessionId, notification);
+    }
+  }
+
   async setSessionMode(
     params: SetSessionModeRequest,
   ): Promise<SetSessionModeResponse> {