fix(agent): preserve Codex goals across cloud resumes

Generated-By: PostHog Code
Task-Id: f8939190-58a1-493b-904c-457c7859b3d6

Author: tatoalo

packages/agent/src/acp-extensions.ts

@@ -86,8 +86,22 @@ export const POSTHOG_NOTIFICATIONS = {
 
   /** RTK output-compression token savings tallied at the end of a run */
   RTK_SAVINGS: "_posthog/rtk_savings",
+
+  /** Latest native Codex goal state, persisted so cold cloud resumes can restore it. */
+  CODEX_GOAL: "_posthog/codex_goal",
 } as const;
 
+export type CodexGoalState = {
+  objective: string;
+  status:
+    | "active"
+    | "paused"
+    | "blocked"
+    | "usageLimited"
+    | "budgetLimited"
+    | "complete";
+};
+
 /**
  * Custom request methods for PostHog-specific operations that need a response
  * (request/response, not fire-and-forget). Used with

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

@@ -306,6 +306,7 @@ describe("CodexAppServerAgent", () => {
       response: { goal: null },
       expectedParams: { threadId: "thr_1" },
       expectedText: "No goal set. Usage: `/goal <objective>`",
+      expectedGoal: undefined,
     },
     {
       label: "reads an active goal",
@@ -314,6 +315,7 @@ describe("CodexAppServerAgent", () => {
       response: { goal: { objective: "Ship the fix", status: "active" } },
       expectedParams: { threadId: "thr_1" },
       expectedText: "Goal active: Ship the fix",
+      expectedGoal: undefined,
     },
     {
       label: "sets a goal",
@@ -322,6 +324,7 @@ describe("CodexAppServerAgent", () => {
       response: { goal: { objective: "Ship the fix", status: "active" } },
       expectedParams: { threadId: "thr_1", objective: "Ship the fix" },
       expectedText: "Goal set: Ship the fix",
+      expectedGoal: { objective: "Ship the fix", status: "active" },
     },
     {
       label: "clears a goal",
@@ -330,6 +333,7 @@ describe("CodexAppServerAgent", () => {
       response: { cleared: true },
       expectedParams: { threadId: "thr_1" },
       expectedText: "Goal cleared.",
+      expectedGoal: null,
     },
     {
       label: "pauses a goal",
@@ -338,6 +342,7 @@ describe("CodexAppServerAgent", () => {
       response: { goal: { objective: "Ship the fix", status: "paused" } },
       expectedParams: { threadId: "thr_1", status: "paused" },
       expectedText: "Goal paused: Ship the fix",
+      expectedGoal: { objective: "Ship the fix", status: "paused" },
     },
     {
       label: "resumes a goal",
@@ -346,13 +351,14 @@ describe("CodexAppServerAgent", () => {
       response: { goal: { objective: "Ship the fix", status: "active" } },
       expectedParams: { threadId: "thr_1", status: "active" },
       expectedText: "Goal resumed: Ship the fix",
+      expectedGoal: { objective: "Ship the fix", status: "active" },
     },
   ])("$label without starting a model turn", async (testCase) => {
     const stub = makeStubRpc({
       "thread/start": { thread: { id: "thr_1" } },
       [testCase.method]: testCase.response,
     });
-    const { client, sessionUpdates } = makeFakeClient();
+    const { client, sessionUpdates, extNotifications } = makeFakeClient();
     const agent = new CodexAppServerAgent(client, {
       processOptions: { binaryPath: "/bundle/codex" },
       rpcFactory: stub.factory,
@@ -379,6 +385,113 @@ describe("CodexAppServerAgent", () => {
         content: { type: "text", text: testCase.expectedText },
       },
     });
+    if (testCase.expectedGoal !== undefined) {
+      expect(extNotifications).toContainEqual({
+        method: "_posthog/codex_goal",
+        params: { goal: testCase.expectedGoal },
+      });
+    }
+  });
+
+  it("handles a goal command wrapped in hidden cold-resume context", async () => {
+    const stub = makeStubRpc({
+      "thread/start": { thread: { id: "thr_1" } },
+      "thread/goal/get": {
+        goal: { objective: "Ship the fix", status: "paused" },
+      },
+    });
+    const { client, sessionUpdates } = makeFakeClient();
+    const agent = new CodexAppServerAgent(client, {
+      processOptions: { binaryPath: "/bundle/codex" },
+      rpcFactory: stub.factory,
+    });
+    await agent.newSession({ cwd: "/repo" } as unknown as NewSessionRequest);
+
+    await agent.prompt({
+      sessionId: "thr_1",
+      prompt: [
+        {
+          type: "text",
+          text: "Previous conversation context",
+          _meta: { ui: { hidden: true } },
+        },
+        { type: "text", text: "/goal" },
+        {
+          type: "text",
+          text: "Respond to the user above",
+          _meta: { ui: { hidden: true } },
+        },
+      ],
+    } as unknown as PromptRequest);
+
+    expect(stub.requests).toContainEqual({
+      method: "thread/goal/get",
+      params: { threadId: "thr_1" },
+    });
+    expect(
+      stub.requests.some((request) => request.method === "turn/start"),
+    ).toBe(false);
+    expect(sessionUpdates).toContainEqual({
+      sessionId: "thr_1",
+      update: {
+        sessionUpdate: "user_message_chunk",
+        content: { type: "text", text: "/goal" },
+      },
+    });
+  });
+
+  it("restores a persisted goal when starting a replacement thread", async () => {
+    const restoredGoal = { objective: "Ship the fix", status: "paused" };
+    const stub = makeStubRpc({
+      "thread/start": { thread: { id: "thr_1" } },
+      "thread/goal/set": { goal: restoredGoal },
+    });
+    const { client, extNotifications } = makeFakeClient();
+    const agent = new CodexAppServerAgent(client, {
+      processOptions: { binaryPath: "/bundle/codex" },
+      rpcFactory: stub.factory,
+    });
+
+    await agent.newSession({
+      cwd: "/repo",
+      _meta: { codexGoal: restoredGoal },
+    } as unknown as NewSessionRequest);
+
+    expect(stub.requests).toContainEqual({
+      method: "thread/goal/set",
+      params: { threadId: "thr_1", ...restoredGoal },
+    });
+    expect(extNotifications).toContainEqual({
+      method: "_posthog/codex_goal",
+      params: { goal: restoredGoal },
+    });
+  });
+
+  it("interrupts a native goal turn that was already queued when paused", async () => {
+    const stub = makeStubRpc({
+      "thread/start": { thread: { id: "thr_1" } },
+      "thread/goal/set": {
+        goal: { objective: "Ship the fix", status: "paused" },
+      },
+    });
+    const { client } = makeFakeClient();
+    const agent = new CodexAppServerAgent(client, {
+      processOptions: { binaryPath: "/bundle/codex" },
+      rpcFactory: stub.factory,
+    });
+    await agent.newSession({ cwd: "/repo" } as unknown as NewSessionRequest);
+
+    await agent.prompt({
+      sessionId: "thr_1",
+      prompt: [{ type: "text", text: "/goal pause" }],
+    } as unknown as PromptRequest);
+    stub.emit("turn/started", { turn: { id: "goal_tick_1" } });
+    await Promise.resolve();
+
+    expect(stub.requests).toContainEqual({
+      method: "turn/interrupt",
+      params: { threadId: "thr_1", turnId: "goal_tick_1" },
+    });
   });
 
   it("includes buffered command output when completion omits aggregatedOutput", async () => {

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

@@ -20,7 +20,10 @@ import type {
   StopReason,
 } from "@agentclientprotocol/sdk";
 import { mcpToolKey, posthogToolMeta } from "@posthog/shared";
-import { POSTHOG_NOTIFICATIONS } from "../../acp-extensions";
+import {
+  type CodexGoalState,
+  POSTHOG_NOTIFICATIONS,
+} from "../../acp-extensions";
 import { DEFAULT_CODEX_MODEL } from "../../gateway-models";
 import type { ProcessSpawnedCallback } from "../../types";
 import { ALLOW_BYPASS } from "../../utils/common";
@@ -87,6 +90,7 @@ type AppServerSessionMeta = {
   channelMode?: boolean;
   spokenNarration?: boolean;
   baseBranch?: string;
+  codexGoal?: CodexGoalState;
 };
 
 /** The subset of codex's `Thread` the adapter reads: id + persisted `turns` for history replay. */
@@ -97,7 +101,7 @@ type AppServerThread = {
 
 type ThreadGoal = {
   objective: string;
-  status: string;
+  status: CodexGoalState["status"];
 };
 
 type GoalCommand =
@@ -119,9 +123,21 @@ const GOAL_COMMAND = {
   input: { hint: "[<objective>|clear|pause|resume]" },
 };
 
+function isHiddenPromptBlock(block: PromptRequest["prompt"][number]): boolean {
+  const meta = block._meta as { ui?: { hidden?: boolean } } | undefined;
+  return meta?.ui?.hidden === true;
+}
+
+function visiblePromptBlocks(
+  prompt: PromptRequest["prompt"],
+): PromptRequest["prompt"] {
+  return prompt.filter((block) => !isHiddenPromptBlock(block));
+}
+
 function parseGoalCommand(prompt: PromptRequest["prompt"]): GoalCommand | null {
-  if (prompt.some((block) => block.type !== "text")) return null;
-  const text = prompt
+  const visible = visiblePromptBlocks(prompt);
+  if (visible.some((block) => block.type !== "text")) return null;
+  const text = visible
     .map((block) => (block.type === "text" ? block.text : ""))
     .join("\n")
     .trim();
@@ -204,6 +220,8 @@ export class CodexAppServerAgent extends BaseAcpAgent {
   private readonly mcp = new McpManager();
   private readonly turns = new TurnController();
   private readonly usage = new UsageTracker();
+  /** Pause/clear can race a goal continuation already queued by app-server. */
+  private cancelNextGoalTurn = false;
 
   constructor(
     client: AgentSideConnection,
@@ -464,6 +482,9 @@ export class CodexAppServerAgent extends BaseAcpAgent {
     }
     this.threadId = threadId;
     this.sessionId = threadId;
+    if (method === APP_SERVER_METHODS.THREAD_START && params.meta?.codexGoal) {
+      await this.restoreGoal(params.meta.codexGoal);
+    }
     await this.loadModelConfig();
     this.emitConfigOptions();
     await this.emitAvailableCommands();
@@ -608,10 +629,11 @@ export class CodexAppServerAgent extends BaseAcpAgent {
     }
     const goalCommand = parseGoalCommand(params.prompt);
     if (goalCommand) {
-      this.broadcastUserInput(params.prompt);
+      this.broadcastUserInput(visiblePromptBlocks(params.prompt));
       await this.handleGoalCommand(goalCommand);
       return { stopReason: "end_turn" };
     }
+    this.cancelNextGoalTurn = false;
     // Reopen the notification gate (a prior interrupt may have left session.cancelled set).
     this.session.cancelled = false;
     // A new prompt while the plan handoff awaits approval implicitly declines it:
@@ -710,10 +732,13 @@ export class CodexAppServerAgent extends BaseAcpAgent {
   private async handleGoalCommand(command: GoalCommand): Promise<void> {
     if (!this.threadId) return;
     if (command.kind === "clear") {
+      this.cancelNextGoalTurn = true;
       const result = await this.rpc.request<{ cleared?: boolean }>(
         APP_SERVER_METHODS.THREAD_GOAL_CLEAR,
         { threadId: this.threadId },
       );
+      await this.emitGoalState(null);
+      await this.cancelRunningGoalTurn();
       this.broadcastAgentText(
         result.cleared ? "Goal cleared." : "No goal was set.",
       );
@@ -733,6 +758,9 @@ export class CodexAppServerAgent extends BaseAcpAgent {
       return;
     }
 
+    if (command.kind === "pause") {
+      this.cancelNextGoalTurn = true;
+    }
     const params =
       command.kind === "set"
         ? { threadId: this.threadId, objective: command.objective }
@@ -744,6 +772,10 @@ export class CodexAppServerAgent extends BaseAcpAgent {
       APP_SERVER_METHODS.THREAD_GOAL_SET,
       params,
     );
+    await this.emitGoalState(result.goal);
+    if (command.kind === "pause") {
+      await this.cancelRunningGoalTurn();
+    }
     const prefix =
       command.kind === "set"
         ? "Goal set"
@@ -753,6 +785,46 @@ export class CodexAppServerAgent extends BaseAcpAgent {
     this.broadcastAgentText(`${prefix}: ${result.goal.objective}`);
   }
 
+  private async restoreGoal(goal: CodexGoalState): Promise<void> {
+    if (!this.threadId) return;
+    const result = await this.rpc.request<{ goal: ThreadGoal }>(
+      APP_SERVER_METHODS.THREAD_GOAL_SET,
+      {
+        threadId: this.threadId,
+        objective: goal.objective,
+        status: goal.status,
+      },
+    );
+    await this.emitGoalState(result.goal);
+  }
+
+  private async emitGoalState(goal: CodexGoalState | null): Promise<void> {
+    await this.client
+      .extNotification(POSTHOG_NOTIFICATIONS.CODEX_GOAL, { goal })
+      .catch((error) =>
+        this.logger.warn("Failed to persist Codex goal state", error),
+      );
+  }
+
+  private async cancelRunningGoalTurn(): Promise<void> {
+    if (!this.turns.isRunning) return;
+    this.cancelNextGoalTurn = false;
+    await this.interrupt();
+  }
+
+  private interruptQueuedGoalTurn(turnId: string | undefined): void {
+    if (!this.cancelNextGoalTurn || !this.threadId || !turnId) return;
+    this.cancelNextGoalTurn = false;
+    void this.rpc
+      .request(APP_SERVER_METHODS.TURN_INTERRUPT, {
+        threadId: this.threadId,
+        turnId,
+      })
+      .catch((error) =>
+        this.logger.warn("Queued goal turn interrupt failed", error),
+      );
+  }
+
   /** Start one codex turn and await its completion. */
   private async runTurn(input: CodexUserInput[]): Promise<StopReason> {
     this.lastAgentMessage = "";
@@ -1081,7 +1153,9 @@ export class CodexAppServerAgent extends BaseAcpAgent {
 
     if (method === APP_SERVER_NOTIFICATIONS.TURN_STARTED) {
       // Capture the active turn id (steer precondition / interrupt target).
-      this.turns.onStarted((params as { turn?: { id?: string } })?.turn?.id);
+      const turnId = (params as { turn?: { id?: string } })?.turn?.id;
+      this.turns.onStarted(turnId);
+      this.interruptQueuedGoalTurn(turnId);
     }
 
     // codex auto-compaction surfaces as a contextCompaction item: item/started → in progress,

packages/agent/src/resume.ts

@@ -16,6 +16,7 @@
  */
 
 import type { ContentBlock } from "@agentclientprotocol/sdk";
+import type { CodexGoalState } from "./acp-extensions";
 import { selectRecentTurns } from "./adapters/claude/session/jsonl-hydration";
 import type { PostHogAPIClient } from "./posthog-api";
 import { ResumeSaga } from "./sagas/resume-saga";
@@ -29,6 +30,7 @@ export interface ResumeState {
   lastDevice?: DeviceInfo;
   logEntryCount: number;
   sessionId: string | null;
+  codexGoal?: CodexGoalState | null;
 }
 
 export interface ConversationTurn {
@@ -95,6 +97,7 @@ export async function resumeFromLog(
     lastDevice: result.data.lastDevice,
     logEntryCount: result.data.logEntryCount,
     sessionId: result.data.sessionId,
+    codexGoal: result.data.codexGoal,
   };
 }