From a3bdd4ccfeea7a5a403aa4b54504b19309eb3170 Mon Sep 17 00:00:00 2001 From: Georges-Antoine Assi Date: Tue, 14 Jul 2026 15:21:55 -0400 Subject: [PATCH 01/10] refactor(auth): make the token cipher async IAuthTokenCipher.encrypt/decrypt become Promise-returning so a host can back them with an async primitive (e.g. the browser's Web Crypto, which is async). Thread await through the AuthService call sites; the desktop adapter wraps its existing synchronous node:crypto cipher in Promise.resolve, so desktop behavior is unchanged. Enabling change for the browser web host, but self-contained and useful on its own. Generated-By: PostHog Code Task-Id: 77d13a30-444d-4045-9d80-5e2a9f2e68ae --- .../src/main/services/auth/port-adapters.ts | 8 +-- packages/core/src/auth/auth.test.ts | 4 +- packages/core/src/auth/auth.ts | 56 +++++++++++-------- packages/core/src/auth/identifiers.ts | 7 ++- 4 files changed, 42 insertions(+), 33 deletions(-) diff --git a/apps/code/src/main/services/auth/port-adapters.ts b/apps/code/src/main/services/auth/port-adapters.ts index c354a9e5d5..784d379df1 100644 --- a/apps/code/src/main/services/auth/port-adapters.ts +++ b/apps/code/src/main/services/auth/port-adapters.ts @@ -37,12 +37,12 @@ import { @injectable() export class TokenCipherPortAdapter implements IAuthTokenCipher { - encrypt(plaintext: string): string { - return encrypt(plaintext); + encrypt(plaintext: string): Promise { + return Promise.resolve(encrypt(plaintext)); } - decrypt(encrypted: string): string | null { - return decrypt(encrypted); + decrypt(encrypted: string): Promise { + return Promise.resolve(decrypt(encrypted)); } } diff --git a/packages/core/src/auth/auth.test.ts b/packages/core/src/auth/auth.test.ts index 00bd2ee104..4b01e75b4a 100644 --- a/packages/core/src/auth/auth.test.ts +++ b/packages/core/src/auth/auth.test.ts @@ -64,8 +64,8 @@ function createPreferencePort(): IAuthPreferenceStore { } const identityCipher: IAuthTokenCipher = { - encrypt: (plaintext) => plaintext, - decrypt: (encrypted) => encrypted, + encrypt: (plaintext) => Promise.resolve(plaintext), + decrypt: (encrypted) => Promise.resolve(encrypted), }; const mockLogger: RootLogger = { diff --git a/packages/core/src/auth/auth.ts b/packages/core/src/auth/auth.ts index 18e3882e88..4253aca900 100644 --- a/packages/core/src/auth/auth.ts +++ b/packages/core/src/auth/auth.ts @@ -254,7 +254,7 @@ export class AuthService extends TypedEventEmitter { ? await this.applyOrgChange(session, newOrgId) : session.orgProjectsMap; - this.commitSessionState(session, { + await this.commitSessionState(session, { orgProjectsMap, currentOrgId: newOrgId, currentProjectId: projectId, @@ -277,7 +277,7 @@ export class AuthService extends TypedEventEmitter { orgId, ); - this.commitSessionState(session, { + await this.commitSessionState(session, { orgProjectsMap, currentOrgId: orgId, currentProjectId, @@ -326,14 +326,14 @@ export class AuthService extends TypedEventEmitter { } return orgProjects[0]?.id ?? null; } - private commitSessionState( + private async commitSessionState( prevSession: InMemorySession, next: { orgProjectsMap: OrgProjectsMap; currentOrgId: string | null; currentProjectId: number | null; }, - ): void { + ): Promise { this.session = { ...prevSession, orgProjectsMap: next.orgProjectsMap, @@ -343,7 +343,7 @@ export class AuthService extends TypedEventEmitter { }; this.persistProjectPreference(this.session); - this.persistSession({ + await this.persistSession({ refreshToken: this.session.refreshToken, cloudRegion: this.session.cloudRegion, selectedProjectId: next.currentProjectId, @@ -413,7 +413,7 @@ export class AuthService extends TypedEventEmitter { return; } - const storedSession = this.resolveStoredSession(); + const storedSession = await this.resolveStoredSession(); if (!storedSession) { this.logger.warn("Stored auth session could not be decrypted"); this.authSession.clearCurrent(); @@ -500,7 +500,7 @@ export class AuthService extends TypedEventEmitter { return this.refreshPromise; } - const sessionInput = this.getSessionInputForRefresh(); + const sessionInput = await this.getSessionInputForRefresh(); const refreshAndSync = async (): Promise => { let session: InMemorySession; @@ -532,7 +532,7 @@ export class AuthService extends TypedEventEmitter { return this.refreshPromise; } - private getSessionInputForRefresh(): StoredSessionInput { + private async getSessionInputForRefresh(): Promise { if (this.session) { return { refreshToken: this.session.refreshToken, @@ -541,7 +541,7 @@ export class AuthService extends TypedEventEmitter { }; } - const storedSession = this.resolveStoredSession(); + const storedSession = await this.resolveStoredSession(); if (!storedSession) { throw new NotAuthenticatedError(); } @@ -794,7 +794,7 @@ export class AuthService extends TypedEventEmitter { session: InMemorySession, ): Promise { this.persistProjectPreference(session); - this.persistSession({ + await this.persistSession({ refreshToken: session.refreshToken, cloudRegion: session.cloudRegion, selectedProjectId: session.currentProjectId, @@ -816,15 +816,15 @@ export class AuthService extends TypedEventEmitter { void this.refreshOrgProjects(); } } - private persistSession(input: { + private async persistSession(input: { refreshToken: string; cloudRegion: CloudRegion; selectedProjectId: number | null; - }): void { + }): Promise { const priorSelected = this.authSession.getCurrent()?.selectedProjectId ?? null; this.authSession.saveCurrent({ - refreshTokenEncrypted: this.cipher.encrypt(input.refreshToken), + refreshTokenEncrypted: await this.cipher.encrypt(input.refreshToken), cloudRegion: input.cloudRegion, selectedProjectId: input.selectedProjectId ?? priorSelected, scopeVersion: OAUTH_SCOPE_VERSION, @@ -1051,11 +1051,13 @@ export class AuthService extends TypedEventEmitter { private handleResume = (): void => { this.attemptSessionRecovery(); }; - private resolveStoredSession(): StoredSessionInput | null { + private async resolveStoredSession(): Promise { const stored = this.authSession.getCurrent(); if (!stored) return null; - const refreshToken = this.cipher.decrypt(stored.refreshTokenEncrypted); + const refreshToken = await this.cipher.decrypt( + stored.refreshTokenEncrypted, + ); if (!refreshToken) return null; return { @@ -1077,14 +1079,10 @@ export class AuthService extends TypedEventEmitter { if (!stored) return; if (stored.scopeVersion < OAUTH_SCOPE_VERSION) return; - if (!this.resolveStoredSession()) return; - - // Route through ensureValidSession so a refresh already in flight (e.g. the - // background bootstrap restore past its deadline) is shared instead of - // kicking a second concurrent token refresh that would burn the same - // rotating refresh token twice. - this.recoveryPromise = this.ensureValidSession() - .then(() => undefined) + // Claim the recovery slot synchronously so concurrent triggers (network + // regained + tab resume) don't both kick a token refresh; decryptability + // is now async (Web Crypto), so it's validated inside recoverSession. + this.recoveryPromise = this.recoverSession() .catch((error) => { this.logger.warn("Session recovery failed", { error }); }) @@ -1092,6 +1090,16 @@ export class AuthService extends TypedEventEmitter { this.recoveryPromise = null; }); } + private async recoverSession(): Promise { + // Bail before touching the network if the stored token can't be decrypted. + if (!(await this.resolveStoredSession())) return; + + // Route through ensureValidSession so a refresh already in flight (e.g. the + // background bootstrap restore past its deadline) is shared instead of + // kicking a second concurrent token refresh that would burn the same + // rotating refresh token twice. + await this.ensureValidSession(); + } private refreshOrgProjects(): Promise { if (this.orgProjectsRefreshPromise) { @@ -1157,7 +1165,7 @@ export class AuthService extends TypedEventEmitter { null, lastSelectedOrgId: lastPrefs?.lastSelectedOrgId ?? null, }); - this.commitSessionState(session, { + await this.commitSessionState(session, { orgProjectsMap: map, currentOrgId: session.currentOrgId, currentProjectId, diff --git a/packages/core/src/auth/identifiers.ts b/packages/core/src/auth/identifiers.ts index 300efbd74a..b0cb002595 100644 --- a/packages/core/src/auth/identifiers.ts +++ b/packages/core/src/auth/identifiers.ts @@ -89,11 +89,12 @@ export const AUTH_OAUTH_FLOW_SERVICE = Symbol.for( /** * Machine-bound symmetric cipher for the refresh token at rest. Desktop adapter - * wraps the existing encryption util (node:crypto + machine id). + * wraps the existing encryption util (node:crypto + machine id); the web adapter + * uses a non-extractable Web Crypto key (async), so the contract is async. */ export interface IAuthTokenCipher { - encrypt(plaintext: string): string; - decrypt(encrypted: string): string | null; + encrypt(plaintext: string): Promise; + decrypt(encrypted: string): Promise; } export const AUTH_TOKEN_CIPHER = Symbol.for("posthog.core.auth.tokenCipher"); From 3efb396f5e7e3543fa49814acea178f6471c4552 Mon Sep 17 00:00:00 2001 From: Georges-Antoine Assi Date: Tue, 14 Jul 2026 16:25:54 -0400 Subject: [PATCH 02/10] fix(auth): commit session state only after the persist succeeds MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit commitSessionState mutated this.session and saved the project preference before awaiting persistSession. Now that encryption is async and can reject, a failure left the service on the new project in memory (and in the saved preference) while the stored session and published state stayed on the old one — a partial, inconsistent commit the caller saw only as a rejected promise. Build the next session locally, await the persist first (the only fallible step), then assign this.session, save the preference, and publish state — so a rejection leaves every layer on the prior session. Add a regression test that fails encryption mid-selectProject and asserts nothing moved. Generated-By: PostHog Code Task-Id: 77d13a30-444d-4045-9d80-5e2a9f2e68ae --- packages/core/src/auth/auth.test.ts | 64 +++++++++++++++++++++++++++++ packages/core/src/auth/auth.ts | 15 +++++-- 2 files changed, 75 insertions(+), 4 deletions(-) diff --git a/packages/core/src/auth/auth.test.ts b/packages/core/src/auth/auth.test.ts index 4b01e75b4a..9736b1651a 100644 --- a/packages/core/src/auth/auth.test.ts +++ b/packages/core/src/auth/auth.test.ts @@ -500,6 +500,70 @@ describe("AuthService", () => { }); }); + it("keeps the prior selection committed when persisting a new selection fails", async () => { + const orgs = { + "org-1": { + name: "Org 1", + projects: [ + { id: 42, name: "Project 42" }, + { id: 84, name: "Project 84" }, + ], + }, + }; + oauthFlow.startFlow.mockResolvedValue( + mockTokenResponse({ + accessToken: "initial-access-token", + refreshToken: "initial-refresh-token", + }), + ); + stubAuthFetch({ orgs }); + + // Encrypts fine during login, then rejects so the selectProject persist + // fails mid-flow (mirrors the browser cipher's Web Crypto rejecting). + let failEncrypt = false; + const flakyCipher: IAuthTokenCipher = { + encrypt: (plaintext) => + failEncrypt + ? Promise.reject(new Error("encryption unavailable")) + : Promise.resolve(plaintext), + decrypt: (encrypted) => Promise.resolve(encrypted), + }; + service = new AuthService( + preferencePort, + sessionPort, + oauthFlow as unknown as IAuthOAuthFlowService, + connectivity, + flakyCipher, + mockPowerManager as unknown as IPowerManager, + mockLogger, + null, + ); + + // Initialize once while the session store is empty so login/selectProject + // don't later trigger the stored-session restore path. + service.init(); + await service.initialize(); + + await service.login("us"); + const priorProjectId = service.getState().currentProjectId; + expect(priorProjectId).toBe(42); + + failEncrypt = true; + await expect(service.selectProject(84)).rejects.toThrow( + "encryption unavailable", + ); + + // A failed persist must not commit the new project anywhere: published + // state, the stored session, and the saved preference all stay on the + // prior selection (the preference is the tell — the buggy order saved it + // to 84 before the encrypt rejected). + expect(service.getState().currentProjectId).toBe(priorProjectId); + expect(sessionPort.getCurrent()?.selectedProjectId).toBe(priorProjectId); + expect(preferencePort.get("user-1", "us")?.lastSelectedProjectId).toBe( + priorProjectId, + ); + }); + it("restores the selected project after app restart while logged out", async () => { const orgs = { "org-1": { diff --git a/packages/core/src/auth/auth.ts b/packages/core/src/auth/auth.ts index 4253aca900..7f02a6738a 100644 --- a/packages/core/src/auth/auth.ts +++ b/packages/core/src/auth/auth.ts @@ -334,7 +334,7 @@ export class AuthService extends TypedEventEmitter { currentProjectId: number | null; }, ): Promise { - this.session = { + const nextSession: InMemorySession = { ...prevSession, orgProjectsMap: next.orgProjectsMap, currentOrgId: next.currentOrgId, @@ -342,13 +342,20 @@ export class AuthService extends TypedEventEmitter { orgProjectsIncomplete: false, }; - this.persistProjectPreference(this.session); + // Persist the durable session first — it's the only step here that can + // fail (encryption is async and may reject). Mutating this.session, the + // project preference, or the published state before it would strand the + // service on a project change that the stored session and UI never + // committed. Commit those only after the persist resolves, so a rejection + // leaves every layer on the prior session. await this.persistSession({ - refreshToken: this.session.refreshToken, - cloudRegion: this.session.cloudRegion, + refreshToken: nextSession.refreshToken, + cloudRegion: nextSession.cloudRegion, selectedProjectId: next.currentProjectId, }); + this.session = nextSession; + this.persistProjectPreference(nextSession); this.updateState({ orgProjectsMap: next.orgProjectsMap, currentOrgId: next.currentOrgId, From 7023c2c75af8af9cfb716f0046734b3ff14d3e87 Mon Sep 17 00:00:00 2001 From: Georges-Antoine Assi Date: Tue, 14 Jul 2026 16:35:40 -0400 Subject: [PATCH 03/10] fix(auth): close the refresh dedup guard synchronously MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Making getSessionInputForRefresh async put an await (stored-token decryption) between ensureValidSession's `if (this.refreshPromise)` guard and the `this.refreshPromise = ...` assignment. Two concurrent callers could both observe a null refreshPromise, both decrypt, then both start a refresh — spending the rotating refresh token twice, so the second request fails with auth_error and clears the session the first one established, logging the user out. Move the decryption inside refreshAndSync so refreshPromise is assigned synchronously right after the guard, restoring single-flight dedup. Add a regression test firing two concurrent refreshAccessToken() calls and asserting one underlying token request. Generated-By: PostHog Code Task-Id: 77d13a30-444d-4045-9d80-5e2a9f2e68ae --- packages/core/src/auth/auth.test.ts | 46 +++++++++++++++++++++++++++++ packages/core/src/auth/auth.ts | 10 +++++-- 2 files changed, 54 insertions(+), 2 deletions(-) diff --git a/packages/core/src/auth/auth.test.ts b/packages/core/src/auth/auth.test.ts index 9736b1651a..469ee1f8db 100644 --- a/packages/core/src/auth/auth.test.ts +++ b/packages/core/src/auth/auth.test.ts @@ -413,6 +413,52 @@ describe("AuthService", () => { } }); + it("dedupes concurrent refreshes into a single token request", async () => { + oauthFlow.startFlow.mockResolvedValue( + mockTokenResponse({ + accessToken: "initial-access-token", + refreshToken: "initial-refresh-token", + }), + ); + stubAuthFetch(); + + // Keep the refresh in flight so both callers overlap; if the refresh + // isn't deduped, the rotating refresh token is spent twice. + let resolveRefresh!: (value: unknown) => void; + oauthFlow.refreshToken.mockReturnValue( + new Promise((resolve) => { + resolveRefresh = resolve; + }), + ); + + await service.initialize(); + await service.login("us"); + + // Two forced refreshes fired in the same tick. The dedup guard must close + // synchronously — resolving the stored session now awaits decryption, so + // if that await sat between the guard and the refreshPromise assignment, + // both callers would slip through and fire their own request. + oauthFlow.refreshToken.mockClear(); + const both = Promise.all([ + service.refreshAccessToken(), + service.refreshAccessToken(), + ]); + + await new Promise((r) => setTimeout(r, 0)); + expect(oauthFlow.refreshToken).toHaveBeenCalledTimes(1); + + resolveRefresh( + mockTokenResponse({ + accessToken: "refreshed-access-token", + refreshToken: "refreshed-refresh-token", + }), + ); + const [a, b] = await both; + expect(a.accessToken).toBe("refreshed-access-token"); + expect(b.accessToken).toBe("refreshed-access-token"); + expect(oauthFlow.refreshToken).toHaveBeenCalledTimes(1); + }); + it("forces a token refresh when explicitly requested", async () => { oauthFlow.startFlow.mockResolvedValue( mockTokenResponse({ diff --git a/packages/core/src/auth/auth.ts b/packages/core/src/auth/auth.ts index 7f02a6738a..ff6b3a1601 100644 --- a/packages/core/src/auth/auth.ts +++ b/packages/core/src/auth/auth.ts @@ -507,9 +507,15 @@ export class AuthService extends TypedEventEmitter { return this.refreshPromise; } - const sessionInput = await this.getSessionInputForRefresh(); - + // Assign refreshPromise synchronously — with no await between the guard + // above and this assignment — so concurrent callers dedupe onto one + // refresh. Resolving the stored session (which now awaits decryption) + // must therefore happen INSIDE refreshAndSync, not before it; otherwise + // two callers both pass the null guard, both decrypt, and both fire a + // refresh, burning the rotating refresh token twice and logging the user + // out when the second request fails. const refreshAndSync = async (): Promise => { + const sessionInput = await this.getSessionInputForRefresh(); let session: InMemorySession; try { session = await this.refreshSession(sessionInput); From 2a76f87f1f1ede0072a98cb110f884ac91f7b30e Mon Sep 17 00:00:00 2001 From: Georges-Antoine Assi Date: Tue, 14 Jul 2026 16:49:25 -0400 Subject: [PATCH 04/10] fix(auth): serialize session-state commits MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit With async encryption, two overlapping project selections could interleave across the await in commitSessionState: an earlier selection completing last would overwrite the newer stored session and published state (and persistSession's own read of the prior selection could race), leaving the committed selection stale — not the one the user picked last. Serialize commits onto a promise chain so each commit's persist-then-publish runs to completion before the next begins; the latest selection now wins consistently across the in-memory session, storage, and subscribers. The stored chain swallows rejections so a failed commit can't wedge later ones, while the returned promise still rejects for the caller. Add a regression test that overlaps two selections, drains encryption newest-first, and asserts the latest selection wins everywhere. Generated-By: PostHog Code Task-Id: 77d13a30-444d-4045-9d80-5e2a9f2e68ae --- packages/core/src/auth/auth.test.ts | 71 +++++++++++++++++++++++++++++ packages/core/src/auth/auth.ts | 28 +++++++++++- 2 files changed, 98 insertions(+), 1 deletion(-) diff --git a/packages/core/src/auth/auth.test.ts b/packages/core/src/auth/auth.test.ts index 469ee1f8db..38be8ae05e 100644 --- a/packages/core/src/auth/auth.test.ts +++ b/packages/core/src/auth/auth.test.ts @@ -610,6 +610,77 @@ describe("AuthService", () => { ); }); + it("keeps overlapping selections consistent so the latest one wins", async () => { + const orgs = { + "org-1": { + name: "Org 1", + projects: [ + { id: 42, name: "Project 42" }, + { id: 84, name: "Project 84" }, + { id: 99, name: "Project 99" }, + ], + }, + }; + oauthFlow.startFlow.mockResolvedValue( + mockTokenResponse({ + accessToken: "initial-access-token", + refreshToken: "initial-refresh-token", + }), + ); + stubAuthFetch({ orgs }); + + // Encryption is gated so overlapping selection commits stay in flight and + // can be resolved out of order below. + let deferEncrypt = false; + const pending: Array<() => void> = []; + const gatedCipher: IAuthTokenCipher = { + encrypt: (plaintext) => + deferEncrypt + ? new Promise((resolve) => { + pending.push(() => resolve(plaintext)); + }) + : Promise.resolve(plaintext), + decrypt: (encrypted) => Promise.resolve(encrypted), + }; + service = new AuthService( + preferencePort, + sessionPort, + oauthFlow as unknown as IAuthOAuthFlowService, + connectivity, + gatedCipher, + mockPowerManager as unknown as IPowerManager, + mockLogger, + null, + ); + service.init(); + await service.initialize(); + await service.login("us"); + expect(service.getState().currentProjectId).toBe(42); + + deferEncrypt = true; + // Two overlapping selections: 84 first, then 99 (the latest intent). + const first = service.selectProject(84); + const second = service.selectProject(99); + + // Drain pending encryptions newest-first each round to surface any + // out-of-order completion, until both commits settle. Serialized commits + // only expose one pending encryption at a time; unserialized ones would + // let the stale 84 commit land last. + const flush = () => new Promise((r) => setTimeout(r, 0)); + await flush(); + while (pending.length > 0) { + for (const resolve of pending.splice(0).reverse()) resolve(); + await flush(); + } + await Promise.all([first, second]); + + // The latest selection wins everywhere — no stale overwrite and no split + // between the in-memory session (getState), stored session, and preference. + expect(service.getState().currentProjectId).toBe(99); + expect(sessionPort.getCurrent()?.selectedProjectId).toBe(99); + expect(preferencePort.get("user-1", "us")?.lastSelectedProjectId).toBe(99); + }); + it("restores the selected project after app restart while logged out", async () => { const orgs = { "org-1": { diff --git a/packages/core/src/auth/auth.ts b/packages/core/src/auth/auth.ts index ff6b3a1601..a0e4d2470e 100644 --- a/packages/core/src/auth/auth.ts +++ b/packages/core/src/auth/auth.ts @@ -86,6 +86,9 @@ export class AuthService extends TypedEventEmitter { private session: InMemorySession | null = null; private initializePromise: Promise | null = null; private refreshPromise: Promise | null = null; + // Serializes session-state commits so overlapping selections can't interleave + // across async encryption (see commitSessionState). + private commitChain: Promise = Promise.resolve(); constructor( @inject(AUTH_PREFERENCE_STORE) private readonly authPreference: IAuthPreferenceStore, @@ -326,7 +329,30 @@ export class AuthService extends TypedEventEmitter { } return orgProjects[0]?.id ?? null; } - private async commitSessionState( + private commitSessionState( + prevSession: InMemorySession, + next: { + orgProjectsMap: OrgProjectsMap; + currentOrgId: string | null; + currentProjectId: number | null; + }, + ): Promise { + // Serialize commits onto a chain. Encryption is async, so two overlapping + // selections would otherwise interleave across the await — an earlier one + // completing last would clobber the newer stored session and published + // state (and persistSession's own read of the prior selection would race). + // Chaining runs each commit's persist-then-publish to completion before the + // next starts, so the latest selection wins consistently across the + // in-memory session, storage, and subscribers. The stored chain swallows + // rejections so one failed commit doesn't wedge later ones; the returned + // promise still rejects for the caller. + const run = this.commitChain.then(() => + this.applyCommittedSession(prevSession, next), + ); + this.commitChain = run.catch(() => {}); + return run; + } + private async applyCommittedSession( prevSession: InMemorySession, next: { orgProjectsMap: OrgProjectsMap; From 93c4fdb67624fbbc0015266a7abddb8daa1919c6 Mon Sep 17 00:00:00 2001 From: Alessandro Pogliaghi Date: Wed, 15 Jul 2026 12:34:39 +0100 Subject: [PATCH 05/10] fix(agent): handle goal commands in Codex cloud runs Translate Codex /goal commands to native app-server goal RPCs and advertise the command alongside skills. Add regression coverage for reading, setting, clearing, pausing, and resuming goals without starting a model turn. Generated-By: PostHog Code Task-Id: f8939190-58a1-493b-904c-457c7859b3d6 --- .../codex-app-server-agent.test.ts | 96 ++++++++++++- .../codex-app-server-agent.ts | 126 ++++++++++++++++-- .../src/adapters/codex-app-server/protocol.ts | 3 + 3 files changed, 216 insertions(+), 9 deletions(-) diff --git a/packages/agent/src/adapters/codex-app-server/codex-app-server-agent.test.ts b/packages/agent/src/adapters/codex-app-server/codex-app-server-agent.test.ts index f613c0f236..763cd4d08e 100644 --- a/packages/agent/src/adapters/codex-app-server/codex-app-server-agent.test.ts +++ b/packages/agent/src/adapters/codex-app-server/codex-app-server-agent.test.ts @@ -14,6 +14,9 @@ import { sandboxPolicyFor } from "./session-config"; // Required-field invariants the native codex app-server enforces on each request. const REQUIRED_FIELDS: Record = { + "thread/goal/clear": ["threadId"], + "thread/goal/get": ["threadId"], + "thread/goal/set": ["threadId"], "turn/interrupt": ["threadId", "turnId"], "turn/steer": ["threadId", "input", "expectedTurnId"], }; @@ -295,6 +298,89 @@ describe("CodexAppServerAgent", () => { ).toHaveLength(1); }); + it.each([ + { + label: "reads an empty goal", + prompt: "/goal", + method: "thread/goal/get", + response: { goal: null }, + expectedParams: { threadId: "thr_1" }, + expectedText: "No goal set. Usage: `/goal `", + }, + { + label: "reads an active goal", + prompt: "/goal", + method: "thread/goal/get", + response: { goal: { objective: "Ship the fix", status: "active" } }, + expectedParams: { threadId: "thr_1" }, + expectedText: "Goal active: Ship the fix", + }, + { + label: "sets a goal", + prompt: "/goal Ship the fix", + method: "thread/goal/set", + response: { goal: { objective: "Ship the fix", status: "active" } }, + expectedParams: { threadId: "thr_1", objective: "Ship the fix" }, + expectedText: "Goal set: Ship the fix", + }, + { + label: "clears a goal", + prompt: "/goal clear", + method: "thread/goal/clear", + response: { cleared: true }, + expectedParams: { threadId: "thr_1" }, + expectedText: "Goal cleared.", + }, + { + label: "pauses a goal", + prompt: "/goal pause", + method: "thread/goal/set", + response: { goal: { objective: "Ship the fix", status: "paused" } }, + expectedParams: { threadId: "thr_1", status: "paused" }, + expectedText: "Goal paused: Ship the fix", + }, + { + label: "resumes a goal", + prompt: "/goal resume", + method: "thread/goal/set", + response: { goal: { objective: "Ship the fix", status: "active" } }, + expectedParams: { threadId: "thr_1", status: "active" }, + expectedText: "Goal resumed: Ship the fix", + }, + ])("$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 agent = new CodexAppServerAgent(client, { + processOptions: { binaryPath: "/bundle/codex" }, + rpcFactory: stub.factory, + }); + + await agent.newSession({ cwd: "/repo" } as unknown as NewSessionRequest); + const result = await agent.prompt({ + sessionId: "thr_1", + prompt: [{ type: "text", text: testCase.prompt }], + } as unknown as PromptRequest); + + expect(result.stopReason).toBe("end_turn"); + expect(stub.requests).toContainEqual({ + method: testCase.method, + params: testCase.expectedParams, + }); + expect( + stub.requests.some((request) => request.method === "turn/start"), + ).toBe(false); + expect(sessionUpdates).toContainEqual({ + sessionId: "thr_1", + update: { + sessionUpdate: "agent_message_chunk", + content: { type: "text", text: testCase.expectedText }, + }, + }); + }); + it("includes buffered command output when completion omits aggregatedOutput", async () => { const stub = makeStubRpc({ initialize: {}, @@ -1711,6 +1797,7 @@ describe("CodexAppServerAgent", () => { { skills: [ { name: "deploy", description: "Deploy", enabled: true }, + { name: "goal", description: "Duplicate", enabled: true }, { name: "danger", description: "Disabled", enabled: false }, ], }, @@ -1729,7 +1816,14 @@ describe("CodexAppServerAgent", () => { (u: any) => u.update?.sessionUpdate === "available_commands_update", ) as any )?.update?.availableCommands; - expect(cmds.map((c: { name: string }) => c.name)).toEqual(["deploy"]); + expect(cmds).toEqual([ + { + name: "goal", + description: "Set or view the goal for a long-running task", + input: { hint: "[|clear|pause|resume]" }, + }, + { name: "deploy", description: "Deploy" }, + ]); }); it("emits _posthog/sdk_session when a taskRunId is present", async () => { diff --git a/packages/agent/src/adapters/codex-app-server/codex-app-server-agent.ts b/packages/agent/src/adapters/codex-app-server/codex-app-server-agent.ts index 406ec589a3..42a3c0cd40 100644 --- a/packages/agent/src/adapters/codex-app-server/codex-app-server-agent.ts +++ b/packages/agent/src/adapters/codex-app-server/codex-app-server-agent.ts @@ -95,6 +95,52 @@ type AppServerThread = { turns?: Array<{ items?: Parameters[1][] }>; }; +type ThreadGoal = { + objective: string; + status: string; +}; + +type GoalCommand = + | { kind: "get" } + | { kind: "clear" } + | { kind: "pause" } + | { kind: "resume" } + | { kind: "set"; objective: string }; + +type CodexSkill = { + name?: string; + description?: string; + enabled?: boolean; +}; + +const GOAL_COMMAND = { + name: "goal", + description: "Set or view the goal for a long-running task", + input: { hint: "[|clear|pause|resume]" }, +}; + +function parseGoalCommand(prompt: PromptRequest["prompt"]): GoalCommand | null { + if (prompt.some((block) => block.type !== "text")) return null; + const text = prompt + .map((block) => (block.type === "text" ? block.text : "")) + .join("\n") + .trim(); + const match = text.match(/^\/goal(?:\s+([\s\S]*))?$/); + if (!match) return null; + const argument = match[1]?.trim(); + if (!argument) return { kind: "get" }; + switch (argument.toLowerCase()) { + case "clear": + return { kind: "clear" }; + case "pause": + return { kind: "pause" }; + case "resume": + return { kind: "resume" }; + default: + return { kind: "set", objective: argument }; + } +} + // The native app-server owns its config; BaseAcpAgent only calls dispose() on this. class NoopSettingsManager implements BaseSettingsManager { constructor(private cwd: string) {} @@ -519,17 +565,29 @@ export class CodexAppServerAgent extends BaseAcpAgent { /** skills/list → available_commands_update so the slash-command menu fills. */ private async emitAvailableCommands(): Promise { if (!this.sessionId) return; - let commands: Array<{ name: string; description: string }> = []; + let commands: Array<{ + name: string; + description: string; + input?: { hint: string }; + }> = [GOAL_COMMAND]; try { - const res = await this.rpc.request<{ data?: Array<{ skills?: any[] }> }>( - APP_SERVER_METHODS.SKILLS_LIST, - {}, - ); - commands = (res?.data ?? []) + const res = await this.rpc.request<{ + data?: Array<{ skills?: CodexSkill[] }>; + }>(APP_SERVER_METHODS.SKILLS_LIST, {}); + const skills = (res?.data ?? []) .flatMap((entry) => entry?.skills ?? []) // Drop explicitly-disabled skills; lenient `!== false` so a malformed payload still shows. - .filter((s) => s?.name && s?.enabled !== false) - .map((s: any) => ({ name: s.name, description: s.description ?? "" })); + .filter( + (skill): skill is CodexSkill & { name: string } => + !!skill.name && + skill.name !== GOAL_COMMAND.name && + skill.enabled !== false, + ) + .map((skill) => ({ + name: skill.name, + description: skill.description ?? "", + })); + commands = [GOAL_COMMAND, ...skills]; } catch (err) { this.logger.warn("skills/list failed", { error: String(err) }); } @@ -548,6 +606,12 @@ export class CodexAppServerAgent extends BaseAcpAgent { if (!this.threadId) { throw new Error("prompt() called before newSession()"); } + const goalCommand = parseGoalCommand(params.prompt); + if (goalCommand) { + this.broadcastUserInput(params.prompt); + await this.handleGoalCommand(goalCommand); + return { stopReason: "end_turn" }; + } // 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: @@ -643,6 +707,52 @@ export class CodexAppServerAgent extends BaseAcpAgent { return { stopReason: await this.maybeOfferPlanImplementation(stopReason) }; } + private async handleGoalCommand(command: GoalCommand): Promise { + if (!this.threadId) return; + if (command.kind === "clear") { + const result = await this.rpc.request<{ cleared?: boolean }>( + APP_SERVER_METHODS.THREAD_GOAL_CLEAR, + { threadId: this.threadId }, + ); + this.broadcastAgentText( + result.cleared ? "Goal cleared." : "No goal was set.", + ); + return; + } + + if (command.kind === "get") { + const result = await this.rpc.request<{ goal?: ThreadGoal | null }>( + APP_SERVER_METHODS.THREAD_GOAL_GET, + { threadId: this.threadId }, + ); + this.broadcastAgentText( + result.goal + ? `Goal ${result.goal.status}: ${result.goal.objective}` + : "No goal set. Usage: `/goal `", + ); + return; + } + + const params = + command.kind === "set" + ? { threadId: this.threadId, objective: command.objective } + : { + threadId: this.threadId, + status: command.kind === "pause" ? "paused" : "active", + }; + const result = await this.rpc.request<{ goal: ThreadGoal }>( + APP_SERVER_METHODS.THREAD_GOAL_SET, + params, + ); + const prefix = + command.kind === "set" + ? "Goal set" + : command.kind === "pause" + ? "Goal paused" + : "Goal resumed"; + this.broadcastAgentText(`${prefix}: ${result.goal.objective}`); + } + /** Start one codex turn and await its completion. */ private async runTurn(input: CodexUserInput[]): Promise { this.lastAgentMessage = ""; diff --git a/packages/agent/src/adapters/codex-app-server/protocol.ts b/packages/agent/src/adapters/codex-app-server/protocol.ts index e50304b0d6..37c61a75e9 100644 --- a/packages/agent/src/adapters/codex-app-server/protocol.ts +++ b/packages/agent/src/adapters/codex-app-server/protocol.ts @@ -15,6 +15,9 @@ export const APP_SERVER_METHODS = { TURN_INTERRUPT: "turn/interrupt", MODEL_LIST: "model/list", SKILLS_LIST: "skills/list", + THREAD_GOAL_SET: "thread/goal/set", + THREAD_GOAL_GET: "thread/goal/get", + THREAD_GOAL_CLEAR: "thread/goal/clear", THREAD_LIST: "thread/list", } as const; From e51cb337cdceae3fcf8a508e4bebbfcf74db0f85 Mon Sep 17 00:00:00 2001 From: Alessandro Pogliaghi Date: Wed, 15 Jul 2026 13:11:40 +0100 Subject: [PATCH 06/10] fix(agent): preserve Codex goals across cloud resumes Generated-By: PostHog Code Task-Id: f8939190-58a1-493b-904c-457c7859b3d6 --- packages/agent/src/acp-extensions.ts | 14 +++ .../codex-app-server-agent.test.ts | 115 +++++++++++++++++- .../codex-app-server-agent.ts | 86 ++++++++++++- packages/agent/src/resume.ts | 3 + packages/agent/src/sagas/resume-saga.test.ts | 50 ++++++++ packages/agent/src/sagas/resume-saga.ts | 46 ++++++- .../agent/src/server/agent-server.test.ts | 51 ++++++++ packages/agent/src/server/agent-server.ts | 44 ++++++- 8 files changed, 396 insertions(+), 13 deletions(-) diff --git a/packages/agent/src/acp-extensions.ts b/packages/agent/src/acp-extensions.ts index cfa5e7b5be..19554f26f8 100644 --- a/packages/agent/src/acp-extensions.ts +++ b/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 diff --git a/packages/agent/src/adapters/codex-app-server/codex-app-server-agent.test.ts b/packages/agent/src/adapters/codex-app-server/codex-app-server-agent.test.ts index 763cd4d08e..6da88235d7 100644 --- a/packages/agent/src/adapters/codex-app-server/codex-app-server-agent.test.ts +++ b/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 `", + 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 () => { diff --git a/packages/agent/src/adapters/codex-app-server/codex-app-server-agent.ts b/packages/agent/src/adapters/codex-app-server/codex-app-server-agent.ts index 42a3c0cd40..fc646c56ef 100644 --- a/packages/agent/src/adapters/codex-app-server/codex-app-server-agent.ts +++ b/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: "[|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 { 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 { + 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 { + 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 { + 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 { 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, diff --git a/packages/agent/src/resume.ts b/packages/agent/src/resume.ts index fe4c600f29..a2244d53fa 100644 --- a/packages/agent/src/resume.ts +++ b/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, }; } diff --git a/packages/agent/src/sagas/resume-saga.test.ts b/packages/agent/src/sagas/resume-saga.test.ts index a26e6c7824..940efa299e 100644 --- a/packages/agent/src/sagas/resume-saga.test.ts +++ b/packages/agent/src/sagas/resume-saga.test.ts @@ -617,6 +617,56 @@ describe("ResumeSaga", () => { }); }); + describe("Codex goal state", () => { + it.each([ + { + name: "restores the latest persisted goal", + entries: [ + createNotification(POSTHOG_NOTIFICATIONS.CODEX_GOAL, { + goal: { objective: "Old goal", status: "active" }, + }), + createNotification(`_${POSTHOG_NOTIFICATIONS.CODEX_GOAL}`, { + goal: { objective: "Ship the fix", status: "paused" }, + }), + ], + expected: { objective: "Ship the fix", status: "paused" }, + }, + { + name: "preserves an explicit goal clear", + entries: [ + createNotification(POSTHOG_NOTIFICATIONS.CODEX_GOAL, { + goal: { objective: "Ship the fix", status: "active" }, + }), + createNotification(POSTHOG_NOTIFICATIONS.CODEX_GOAL, { goal: null }), + ], + expected: null, + }, + { + name: "leaves goal state undefined when no notification exists", + entries: [createUserMessage("Hello")], + expected: undefined, + }, + ])("$name", async ({ entries, expected }) => { + (mockApiClient.getTaskRun as ReturnType).mockResolvedValue( + createTaskRun(), + ); + ( + mockApiClient.fetchTaskRunLogs as ReturnType + ).mockResolvedValue(entries); + + const result = await new ResumeSaga(mockLogger).run({ + taskId: "task-1", + runId: "run-1", + repositoryPath: repo.path, + apiClient: mockApiClient, + }); + + expect(result.success).toBe(true); + if (!result.success) return; + expect(result.data.codexGoal).toEqual(expected); + }); + }); + describe("log entry count", () => { it("reports correct log entry count", async () => { (mockApiClient.getTaskRun as ReturnType).mockResolvedValue( diff --git a/packages/agent/src/sagas/resume-saga.ts b/packages/agent/src/sagas/resume-saga.ts index 09ad7e45e7..195859a9f4 100644 --- a/packages/agent/src/sagas/resume-saga.ts +++ b/packages/agent/src/sagas/resume-saga.ts @@ -1,6 +1,6 @@ import type { ContentBlock } from "@agentclientprotocol/sdk"; import { Saga } from "@posthog/shared"; -import { POSTHOG_NOTIFICATIONS } from "../acp-extensions"; +import { type CodexGoalState, POSTHOG_NOTIFICATIONS } from "../acp-extensions"; import type { PostHogAPIClient } from "../posthog-api"; import type { DeviceInfo, @@ -37,6 +37,7 @@ export interface ResumeOutput { lastDevice?: DeviceInfo; logEntryCount: number; sessionId: string | null; + codexGoal?: CodexGoalState | null; } export class ResumeSaga extends Saga { @@ -91,6 +92,9 @@ export class ResumeSaga extends Saga { const sessionId = await this.readOnlyStep("find_session_id", () => Promise.resolve(this.findSessionId(entries)), ); + const codexGoal = await this.readOnlyStep("find_codex_goal", () => + Promise.resolve(this.findCodexGoal(entries)), + ); this.log.info("Resume state rebuilt", { turns: conversation.length, @@ -106,6 +110,7 @@ export class ResumeSaga extends Saga { lastDevice, logEntryCount: entries.length, sessionId, + codexGoal, }; } @@ -116,9 +121,48 @@ export class ResumeSaga extends Saga { interrupted: false, logEntryCount: 0, sessionId: null, + codexGoal: undefined, }; } + private findCodexGoal( + entries: StoredNotification[], + ): CodexGoalState | null | undefined { + const statuses = new Set([ + "active", + "paused", + "blocked", + "usageLimited", + "budgetLimited", + "complete", + ]); + const methods = new Set([ + POSTHOG_NOTIFICATIONS.CODEX_GOAL, + `_${POSTHOG_NOTIFICATIONS.CODEX_GOAL}`, + ]); + for (let index = entries.length - 1; index >= 0; index--) { + const notification = entries[index].notification; + if (!methods.has(notification?.method ?? "")) continue; + const goal = (notification?.params as { goal?: unknown } | undefined) + ?.goal; + if (goal === null) return null; + if (!goal || typeof goal !== "object") return undefined; + const value = goal as Record; + if ( + typeof value.objective === "string" && + typeof value.status === "string" && + statuses.has(value.status as CodexGoalState["status"]) + ) { + return { + objective: value.objective, + status: value.status as CodexGoalState["status"], + }; + } + return undefined; + } + return undefined; + } + private findSessionId(entries: StoredNotification[]): string | null { const runStarted = POSTHOG_NOTIFICATIONS.RUN_STARTED; for (let i = entries.length - 1; i >= 0; i--) { diff --git a/packages/agent/src/server/agent-server.test.ts b/packages/agent/src/server/agent-server.test.ts index bc43a7858b..5b48971c43 100644 --- a/packages/agent/src/server/agent-server.test.ts +++ b/packages/agent/src/server/agent-server.test.ts @@ -242,6 +242,10 @@ interface TestableServer { buildClaudeCodeSessionMeta( runtimeAdapter: Adapter, ): { claudeCode: { options: Record } } | undefined; + resumeState: ResumeState | null; + getCodexGoalForFreshSession( + runtimeAdapter: Adapter, + ): ResumeState["codexGoal"]; } interface NativeResumeTestServer { @@ -436,6 +440,37 @@ describe("AgentServer HTTP Mode", () => { ); }; + it("replays ACP notifications emitted before cloud session assignment", () => { + const testServer = createServer() as unknown as { + session: { sseController: null } | null; + pendingEvents: Record[]; + preSessionEvents: Record[]; + handleAcpTransportMessage(message: unknown): void; + flushPreSessionEvents(): void; + }; + const message = { + method: "session/update", + params: { + update: { + sessionUpdate: "available_commands_update", + availableCommands: [{ name: "goal" }], + }, + }, + }; + + testServer.handleAcpTransportMessage(message); + expect(testServer.preSessionEvents).toHaveLength(1); + + testServer.session = { sseController: null }; + testServer.flushPreSessionEvents(); + + expect(testServer.preSessionEvents).toHaveLength(0); + expect(testServer.pendingEvents).toContainEqual( + expect.objectContaining({ notification: message }), + ); + testServer.session = null; + }); + describe("GET /health", () => { it("returns ok status with active session", async () => { await createServer().start(); @@ -2345,6 +2380,22 @@ describe("AgentServer HTTP Mode", () => { }); describe("native resume", () => { + it("restores persisted Codex goals only for fresh Codex sessions", () => { + const s = createServer() as unknown as TestableServer; + const goal = { objective: "Ship the fix", status: "paused" as const }; + s.resumeState = { + conversation: [], + latestGitCheckpoint: null, + interrupted: false, + logEntryCount: 1, + sessionId: "prior-session", + codexGoal: goal, + }; + + expect(s.getCodexGoalForFreshSession("codex")).toEqual(goal); + expect(s.getCodexGoalForFreshSession("claude")).toBeUndefined(); + }); + it.each([ { retryOutcome: "succeeds", retryFails: false }, { retryOutcome: "fails", retryFails: true }, diff --git a/packages/agent/src/server/agent-server.ts b/packages/agent/src/server/agent-server.ts index 5210b7540f..0bbdd20173 100644 --- a/packages/agent/src/server/agent-server.ts +++ b/packages/agent/src/server/agent-server.ts @@ -368,6 +368,8 @@ export class AgentServer { // causing a second session to be created and duplicate Slack messages to be sent. private initializationPromise: Promise | null = null; private pendingEvents: Record[] = []; + /** ACP notifications emitted by newSession/resumeSession before this.session is assigned. */ + private preSessionEvents: Record[] = []; private deliveredMessageIds = new Set(); private pendingCompactContinuationMessageIds = new Set(); private pendingPermissions = new Map< @@ -800,6 +802,13 @@ export class AgentServer { return { sessionId: priorSessionId, warm }; } + private getCodexGoalForFreshSession( + runtimeAdapter: Adapter, + ): ResumeState["codexGoal"] { + if (runtimeAdapter !== "codex") return undefined; + return this.resumeState?.codexGoal; + } + async stop(): Promise { this.logger.debug("Stopping agent server..."); @@ -1220,6 +1229,7 @@ export class AgentServer { this.resumeState = null; this.nativeResume = null; + this.preSessionEvents = []; this.prewarmedRun = false; this.warmAutoPublishResolved = false; @@ -1427,6 +1437,9 @@ export class AgentServer { sessionCwd, initialPermissionMode, ); + let effectiveSessionMeta: typeof sessionMeta & { + codexGoal?: NonNullable; + } = sessionMeta; let acpSessionId: string | null = null; if (nativeResume) { @@ -1435,7 +1448,7 @@ export class AgentServer { sessionId: nativeResume.sessionId, cwd: sessionCwd, mcpServers: this.config.mcpServers ?? [], - _meta: { ...sessionMeta, sessionId: nativeResume.sessionId }, + _meta: { ...effectiveSessionMeta, sessionId: nativeResume.sessionId }, }); acpSessionId = nativeResume.sessionId; this.nativeResume = nativeResume; @@ -1454,10 +1467,15 @@ export class AgentServer { } } if (!acpSessionId) { + const restoredCodexGoal = + this.getCodexGoalForFreshSession(runtimeAdapter); + effectiveSessionMeta = restoredCodexGoal + ? { ...sessionMeta, codexGoal: restoredCodexGoal } + : sessionMeta; const sessionResponse = await clientConnection.newSession({ cwd: sessionCwd, mcpServers: this.config.mcpServers ?? [], - _meta: sessionMeta, + _meta: effectiveSessionMeta, }); acpSessionId = sessionResponse.sessionId; this.logger.debug("ACP session created", { @@ -1480,8 +1498,9 @@ export class AgentServer { permissionMode: initialPermissionMode, hasDesktopConnected: sseController !== null, pendingHandoffGitState: undefined, - sessionMeta, + sessionMeta: effectiveSessionMeta, }; + this.flushPreSessionEvents(); this.logger = new Logger({ debug: true, @@ -3977,6 +3996,7 @@ ${signedCommitInstructions}${prLinkInstructions}${shellEfficiencyInstructions} } this.pendingEvents = []; + this.preSessionEvents = []; this.lastReportedBranch = null; // Run usage is per run: a later session on this instance (e.g. a resume // with a different run_id) must not inherit the previous run's totals. @@ -4098,11 +4118,16 @@ ${signedCommitInstructions}${prLinkInstructions}${shellEfficiencyInstructions} } this.adapterEmittedTurnComplete = true; } - this.broadcastEvent({ + const event = { type: "notification", timestamp: new Date().toISOString(), notification: message, - }); + }; + if (!this.session) { + this.preSessionEvents.push(event); + return; + } + this.broadcastEvent(event); } private broadcastTurnComplete(stopReason: string): void { @@ -4145,6 +4170,15 @@ ${signedCommitInstructions}${prLinkInstructions}${shellEfficiencyInstructions} } } + private flushPreSessionEvents(): void { + if (!this.session || this.preSessionEvents.length === 0) return; + const events = this.preSessionEvents; + this.preSessionEvents = []; + for (const event of events) { + this.broadcastEvent(event); + } + } + private replayPendingEvents(): void { if (!this.session?.sseController || this.pendingEvents.length === 0) return; const events = this.pendingEvents; From 45dff9489a9adbf764b56023a6258d1e55b1c296 Mon Sep 17 00:00:00 2001 From: Alessandro Pogliaghi Date: Wed, 15 Jul 2026 14:54:04 +0100 Subject: [PATCH 07/10] refactor(agent): generalize native goal naming Generated-By: PostHog Code Task-Id: f8939190-58a1-493b-904c-457c7859b3d6 --- packages/agent/src/acp-extensions.ts | 2 +- .../codex-app-server-agent.test.ts | 2 +- .../codex-app-server-agent.ts | 14 ++++++------ packages/agent/src/resume.ts | 6 ++--- packages/agent/src/sagas/resume-saga.test.ts | 2 +- packages/agent/src/sagas/resume-saga.ts | 22 +++++++++---------- .../agent/src/server/agent-server.test.ts | 10 ++++----- packages/agent/src/server/agent-server.ts | 16 +++++++------- 8 files changed, 37 insertions(+), 37 deletions(-) diff --git a/packages/agent/src/acp-extensions.ts b/packages/agent/src/acp-extensions.ts index 19554f26f8..26783a49f3 100644 --- a/packages/agent/src/acp-extensions.ts +++ b/packages/agent/src/acp-extensions.ts @@ -91,7 +91,7 @@ export const POSTHOG_NOTIFICATIONS = { CODEX_GOAL: "_posthog/codex_goal", } as const; -export type CodexGoalState = { +export type NativeGoalState = { objective: string; status: | "active" diff --git a/packages/agent/src/adapters/codex-app-server/codex-app-server-agent.test.ts b/packages/agent/src/adapters/codex-app-server/codex-app-server-agent.test.ts index 6da88235d7..1e619fb96e 100644 --- a/packages/agent/src/adapters/codex-app-server/codex-app-server-agent.test.ts +++ b/packages/agent/src/adapters/codex-app-server/codex-app-server-agent.test.ts @@ -454,7 +454,7 @@ describe("CodexAppServerAgent", () => { await agent.newSession({ cwd: "/repo", - _meta: { codexGoal: restoredGoal }, + _meta: { nativeGoal: restoredGoal }, } as unknown as NewSessionRequest); expect(stub.requests).toContainEqual({ diff --git a/packages/agent/src/adapters/codex-app-server/codex-app-server-agent.ts b/packages/agent/src/adapters/codex-app-server/codex-app-server-agent.ts index fc646c56ef..912e4055f2 100644 --- a/packages/agent/src/adapters/codex-app-server/codex-app-server-agent.ts +++ b/packages/agent/src/adapters/codex-app-server/codex-app-server-agent.ts @@ -21,7 +21,7 @@ import type { } from "@agentclientprotocol/sdk"; import { mcpToolKey, posthogToolMeta } from "@posthog/shared"; import { - type CodexGoalState, + type NativeGoalState, POSTHOG_NOTIFICATIONS, } from "../../acp-extensions"; import { DEFAULT_CODEX_MODEL } from "../../gateway-models"; @@ -90,7 +90,7 @@ type AppServerSessionMeta = { channelMode?: boolean; spokenNarration?: boolean; baseBranch?: string; - codexGoal?: CodexGoalState; + nativeGoal?: NativeGoalState; }; /** The subset of codex's `Thread` the adapter reads: id + persisted `turns` for history replay. */ @@ -101,7 +101,7 @@ type AppServerThread = { type ThreadGoal = { objective: string; - status: CodexGoalState["status"]; + status: NativeGoalState["status"]; }; type GoalCommand = @@ -482,8 +482,8 @@ 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); + if (method === APP_SERVER_METHODS.THREAD_START && params.meta?.nativeGoal) { + await this.restoreGoal(params.meta.nativeGoal); } await this.loadModelConfig(); this.emitConfigOptions(); @@ -785,7 +785,7 @@ export class CodexAppServerAgent extends BaseAcpAgent { this.broadcastAgentText(`${prefix}: ${result.goal.objective}`); } - private async restoreGoal(goal: CodexGoalState): Promise { + private async restoreGoal(goal: NativeGoalState): Promise { if (!this.threadId) return; const result = await this.rpc.request<{ goal: ThreadGoal }>( APP_SERVER_METHODS.THREAD_GOAL_SET, @@ -798,7 +798,7 @@ export class CodexAppServerAgent extends BaseAcpAgent { await this.emitGoalState(result.goal); } - private async emitGoalState(goal: CodexGoalState | null): Promise { + private async emitGoalState(goal: NativeGoalState | null): Promise { await this.client .extNotification(POSTHOG_NOTIFICATIONS.CODEX_GOAL, { goal }) .catch((error) => diff --git a/packages/agent/src/resume.ts b/packages/agent/src/resume.ts index a2244d53fa..fad960cf62 100644 --- a/packages/agent/src/resume.ts +++ b/packages/agent/src/resume.ts @@ -16,7 +16,7 @@ */ import type { ContentBlock } from "@agentclientprotocol/sdk"; -import type { CodexGoalState } from "./acp-extensions"; +import type { NativeGoalState } from "./acp-extensions"; import { selectRecentTurns } from "./adapters/claude/session/jsonl-hydration"; import type { PostHogAPIClient } from "./posthog-api"; import { ResumeSaga } from "./sagas/resume-saga"; @@ -30,7 +30,7 @@ export interface ResumeState { lastDevice?: DeviceInfo; logEntryCount: number; sessionId: string | null; - codexGoal?: CodexGoalState | null; + nativeGoal?: NativeGoalState | null; } export interface ConversationTurn { @@ -97,7 +97,7 @@ export async function resumeFromLog( lastDevice: result.data.lastDevice, logEntryCount: result.data.logEntryCount, sessionId: result.data.sessionId, - codexGoal: result.data.codexGoal, + nativeGoal: result.data.nativeGoal, }; } diff --git a/packages/agent/src/sagas/resume-saga.test.ts b/packages/agent/src/sagas/resume-saga.test.ts index 940efa299e..5e19c57d00 100644 --- a/packages/agent/src/sagas/resume-saga.test.ts +++ b/packages/agent/src/sagas/resume-saga.test.ts @@ -663,7 +663,7 @@ describe("ResumeSaga", () => { expect(result.success).toBe(true); if (!result.success) return; - expect(result.data.codexGoal).toEqual(expected); + expect(result.data.nativeGoal).toEqual(expected); }); }); diff --git a/packages/agent/src/sagas/resume-saga.ts b/packages/agent/src/sagas/resume-saga.ts index 195859a9f4..700369e6af 100644 --- a/packages/agent/src/sagas/resume-saga.ts +++ b/packages/agent/src/sagas/resume-saga.ts @@ -1,6 +1,6 @@ import type { ContentBlock } from "@agentclientprotocol/sdk"; import { Saga } from "@posthog/shared"; -import { type CodexGoalState, POSTHOG_NOTIFICATIONS } from "../acp-extensions"; +import { type NativeGoalState, POSTHOG_NOTIFICATIONS } from "../acp-extensions"; import type { PostHogAPIClient } from "../posthog-api"; import type { DeviceInfo, @@ -37,7 +37,7 @@ export interface ResumeOutput { lastDevice?: DeviceInfo; logEntryCount: number; sessionId: string | null; - codexGoal?: CodexGoalState | null; + nativeGoal?: NativeGoalState | null; } export class ResumeSaga extends Saga { @@ -92,8 +92,8 @@ export class ResumeSaga extends Saga { const sessionId = await this.readOnlyStep("find_session_id", () => Promise.resolve(this.findSessionId(entries)), ); - const codexGoal = await this.readOnlyStep("find_codex_goal", () => - Promise.resolve(this.findCodexGoal(entries)), + const nativeGoal = await this.readOnlyStep("find_native_goal", () => + Promise.resolve(this.findNativeGoal(entries)), ); this.log.info("Resume state rebuilt", { @@ -110,7 +110,7 @@ export class ResumeSaga extends Saga { lastDevice, logEntryCount: entries.length, sessionId, - codexGoal, + nativeGoal, }; } @@ -121,14 +121,14 @@ export class ResumeSaga extends Saga { interrupted: false, logEntryCount: 0, sessionId: null, - codexGoal: undefined, + nativeGoal: undefined, }; } - private findCodexGoal( + private findNativeGoal( entries: StoredNotification[], - ): CodexGoalState | null | undefined { - const statuses = new Set([ + ): NativeGoalState | null | undefined { + const statuses = new Set([ "active", "paused", "blocked", @@ -151,11 +151,11 @@ export class ResumeSaga extends Saga { if ( typeof value.objective === "string" && typeof value.status === "string" && - statuses.has(value.status as CodexGoalState["status"]) + statuses.has(value.status as NativeGoalState["status"]) ) { return { objective: value.objective, - status: value.status as CodexGoalState["status"], + status: value.status as NativeGoalState["status"], }; } return undefined; diff --git a/packages/agent/src/server/agent-server.test.ts b/packages/agent/src/server/agent-server.test.ts index 5b48971c43..7edaffac73 100644 --- a/packages/agent/src/server/agent-server.test.ts +++ b/packages/agent/src/server/agent-server.test.ts @@ -243,9 +243,9 @@ interface TestableServer { runtimeAdapter: Adapter, ): { claudeCode: { options: Record } } | undefined; resumeState: ResumeState | null; - getCodexGoalForFreshSession( + getNativeGoalForFreshSession( runtimeAdapter: Adapter, - ): ResumeState["codexGoal"]; + ): ResumeState["nativeGoal"]; } interface NativeResumeTestServer { @@ -2389,11 +2389,11 @@ describe("AgentServer HTTP Mode", () => { interrupted: false, logEntryCount: 1, sessionId: "prior-session", - codexGoal: goal, + nativeGoal: goal, }; - expect(s.getCodexGoalForFreshSession("codex")).toEqual(goal); - expect(s.getCodexGoalForFreshSession("claude")).toBeUndefined(); + expect(s.getNativeGoalForFreshSession("codex")).toEqual(goal); + expect(s.getNativeGoalForFreshSession("claude")).toBeUndefined(); }); it.each([ diff --git a/packages/agent/src/server/agent-server.ts b/packages/agent/src/server/agent-server.ts index 0bbdd20173..01e69aa342 100644 --- a/packages/agent/src/server/agent-server.ts +++ b/packages/agent/src/server/agent-server.ts @@ -802,11 +802,11 @@ export class AgentServer { return { sessionId: priorSessionId, warm }; } - private getCodexGoalForFreshSession( + private getNativeGoalForFreshSession( runtimeAdapter: Adapter, - ): ResumeState["codexGoal"] { + ): ResumeState["nativeGoal"] { if (runtimeAdapter !== "codex") return undefined; - return this.resumeState?.codexGoal; + return this.resumeState?.nativeGoal; } async stop(): Promise { @@ -1438,7 +1438,7 @@ export class AgentServer { initialPermissionMode, ); let effectiveSessionMeta: typeof sessionMeta & { - codexGoal?: NonNullable; + nativeGoal?: NonNullable; } = sessionMeta; let acpSessionId: string | null = null; @@ -1467,10 +1467,10 @@ export class AgentServer { } } if (!acpSessionId) { - const restoredCodexGoal = - this.getCodexGoalForFreshSession(runtimeAdapter); - effectiveSessionMeta = restoredCodexGoal - ? { ...sessionMeta, codexGoal: restoredCodexGoal } + const restoredNativeGoal = + this.getNativeGoalForFreshSession(runtimeAdapter); + effectiveSessionMeta = restoredNativeGoal + ? { ...sessionMeta, nativeGoal: restoredNativeGoal } : sessionMeta; const sessionResponse = await clientConnection.newSession({ cwd: sessionCwd, From a6ec9ca02c22170f1dbb618cb973558437caaea4 Mon Sep 17 00:00:00 2001 From: Alessandro Pogliaghi Date: Wed, 15 Jul 2026 15:02:50 +0100 Subject: [PATCH 08/10] fix(agent): harden Codex goal recovery --- .../codex-app-server-agent.test.ts | 53 ++++++++++++++++++- .../codex-app-server-agent.ts | 4 +- packages/agent/src/sagas/resume-saga.test.ts | 13 +++++ packages/agent/src/sagas/resume-saga.ts | 3 +- 4 files changed, 69 insertions(+), 4 deletions(-) diff --git a/packages/agent/src/adapters/codex-app-server/codex-app-server-agent.test.ts b/packages/agent/src/adapters/codex-app-server/codex-app-server-agent.test.ts index 1e619fb96e..2f45e2c4b3 100644 --- a/packages/agent/src/adapters/codex-app-server/codex-app-server-agent.test.ts +++ b/packages/agent/src/adapters/codex-app-server/codex-app-server-agent.test.ts @@ -46,7 +46,12 @@ function makeStubRpc(responses: Record) { message: `Invalid request: missing field \`${missing}\``, }; } - return (responses[method] ?? {}) as T; + const response = responses[method]; + return ( + typeof response === "function" + ? await response(params) + : (response ?? {}) + ) as T; }, notify() {}, async close() {}, @@ -494,6 +499,52 @@ describe("CodexAppServerAgent", () => { }); }); + it("retries queued goal cancellation after an interrupt failure", async () => { + let interruptAttempts = 0; + const stub = makeStubRpc({ + "thread/start": { thread: { id: "thr_1" } }, + "thread/goal/set": { + goal: { objective: "Ship the fix", status: "paused" }, + }, + "turn/interrupt": () => { + interruptAttempts++; + if (interruptAttempts === 1) { + throw new Error("interrupt failed"); + } + return {}; + }, + }); + 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(); + await Promise.resolve(); + stub.emit("turn/started", { turn: { id: "goal_tick_2" } }); + await Promise.resolve(); + + expect( + stub.requests.filter((request) => request.method === "turn/interrupt"), + ).toEqual([ + { + method: "turn/interrupt", + params: { threadId: "thr_1", turnId: "goal_tick_1" }, + }, + { + method: "turn/interrupt", + params: { threadId: "thr_1", turnId: "goal_tick_2" }, + }, + ]); + }); + it("includes buffered command output when completion omits aggregatedOutput", async () => { const stub = makeStubRpc({ initialize: {}, diff --git a/packages/agent/src/adapters/codex-app-server/codex-app-server-agent.ts b/packages/agent/src/adapters/codex-app-server/codex-app-server-agent.ts index 912e4055f2..fe19b1c3e9 100644 --- a/packages/agent/src/adapters/codex-app-server/codex-app-server-agent.ts +++ b/packages/agent/src/adapters/codex-app-server/codex-app-server-agent.ts @@ -814,12 +814,14 @@ export class CodexAppServerAgent extends BaseAcpAgent { 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, }) + .then(() => { + this.cancelNextGoalTurn = false; + }) .catch((error) => this.logger.warn("Queued goal turn interrupt failed", error), ); diff --git a/packages/agent/src/sagas/resume-saga.test.ts b/packages/agent/src/sagas/resume-saga.test.ts index 5e19c57d00..a3919c6de7 100644 --- a/packages/agent/src/sagas/resume-saga.test.ts +++ b/packages/agent/src/sagas/resume-saga.test.ts @@ -641,6 +641,19 @@ describe("ResumeSaga", () => { ], expected: null, }, + { + name: "skips malformed newer goal notifications", + entries: [ + createNotification(POSTHOG_NOTIFICATIONS.CODEX_GOAL, { + goal: { objective: "Ship the fix", status: "paused" }, + }), + createNotification(POSTHOG_NOTIFICATIONS.CODEX_GOAL, { + goal: { objective: "Invalid goal", status: "unknown" }, + }), + createNotification(POSTHOG_NOTIFICATIONS.CODEX_GOAL, {}), + ], + expected: { objective: "Ship the fix", status: "paused" }, + }, { name: "leaves goal state undefined when no notification exists", entries: [createUserMessage("Hello")], diff --git a/packages/agent/src/sagas/resume-saga.ts b/packages/agent/src/sagas/resume-saga.ts index 700369e6af..d18309675f 100644 --- a/packages/agent/src/sagas/resume-saga.ts +++ b/packages/agent/src/sagas/resume-saga.ts @@ -146,7 +146,7 @@ export class ResumeSaga extends Saga { const goal = (notification?.params as { goal?: unknown } | undefined) ?.goal; if (goal === null) return null; - if (!goal || typeof goal !== "object") return undefined; + if (!goal || typeof goal !== "object") continue; const value = goal as Record; if ( typeof value.objective === "string" && @@ -158,7 +158,6 @@ export class ResumeSaga extends Saga { status: value.status as NativeGoalState["status"], }; } - return undefined; } return undefined; } From c6fb600862724abf5e92a5f142950a0d2f347eb4 Mon Sep 17 00:00:00 2001 From: Alessandro Pogliaghi Date: Wed, 15 Jul 2026 15:12:10 +0100 Subject: [PATCH 09/10] fix(agent): cancel active Codex goal turns --- .../codex-app-server-agent.test.ts | 26 ++++++++++++++ .../codex-app-server-agent.ts | 34 ++++++++++++++++--- 2 files changed, 55 insertions(+), 5 deletions(-) diff --git a/packages/agent/src/adapters/codex-app-server/codex-app-server-agent.test.ts b/packages/agent/src/adapters/codex-app-server/codex-app-server-agent.test.ts index 2f45e2c4b3..d8688b8b40 100644 --- a/packages/agent/src/adapters/codex-app-server/codex-app-server-agent.test.ts +++ b/packages/agent/src/adapters/codex-app-server/codex-app-server-agent.test.ts @@ -499,6 +499,32 @@ describe("CodexAppServerAgent", () => { }); }); + it("interrupts a native goal turn that started before it was 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); + stub.emit("turn/started", { turn: { id: "goal_tick_1" } }); + + await agent.prompt({ + sessionId: "thr_1", + prompt: [{ type: "text", text: "/goal pause" }], + } as unknown as PromptRequest); + + expect(stub.requests).toContainEqual({ + method: "turn/interrupt", + params: { threadId: "thr_1", turnId: "goal_tick_1" }, + }); + }); + it("retries queued goal cancellation after an interrupt failure", async () => { let interruptAttempts = 0; const stub = makeStubRpc({ diff --git a/packages/agent/src/adapters/codex-app-server/codex-app-server-agent.ts b/packages/agent/src/adapters/codex-app-server/codex-app-server-agent.ts index fe19b1c3e9..7ed57f3339 100644 --- a/packages/agent/src/adapters/codex-app-server/codex-app-server-agent.ts +++ b/packages/agent/src/adapters/codex-app-server/codex-app-server-agent.ts @@ -222,6 +222,8 @@ export class CodexAppServerAgent extends BaseAcpAgent { private readonly usage = new UsageTracker(); /** Pause/clear can race a goal continuation already queued by app-server. */ private cancelNextGoalTurn = false; + /** Native goal ticks start outside prompt(), so TurnController does not own them. */ + private nativeGoalTurnId?: string; constructor( client: AgentSideConnection, @@ -427,6 +429,8 @@ export class CodexAppServerAgent extends BaseAcpAgent { additionalDirectories?: string[]; }, ): Promise<{ threadId: string; thread: AppServerThread | undefined }> { + this.cancelNextGoalTurn = false; + this.nativeGoalTurnId = undefined; this.jsonSchema = params.meta?.jsonSchema ?? undefined; this.taskRunId = params.meta?.taskRunId; this.environment = params.meta?.environment; @@ -807,23 +811,36 @@ export class CodexAppServerAgent extends BaseAcpAgent { } private async cancelRunningGoalTurn(): Promise { - if (!this.turns.isRunning) return; - this.cancelNextGoalTurn = false; - await this.interrupt(); + if (this.turns.isRunning) { + this.cancelNextGoalTurn = false; + await this.interrupt(); + return; + } + if (this.nativeGoalTurnId) { + await this.interruptNativeGoalTurn(this.nativeGoalTurnId); + } } private interruptQueuedGoalTurn(turnId: string | undefined): void { if (!this.cancelNextGoalTurn || !this.threadId || !turnId) return; - void this.rpc + void this.interruptNativeGoalTurn(turnId); + } + + private async interruptNativeGoalTurn(turnId: string): Promise { + if (!this.threadId) return; + await this.rpc .request(APP_SERVER_METHODS.TURN_INTERRUPT, { threadId: this.threadId, turnId, }) .then(() => { this.cancelNextGoalTurn = false; + if (this.nativeGoalTurnId === turnId) { + this.nativeGoalTurnId = undefined; + } }) .catch((error) => - this.logger.warn("Queued goal turn interrupt failed", error), + this.logger.warn("Native goal turn interrupt failed", error), ); } @@ -1106,6 +1123,7 @@ export class CodexAppServerAgent extends BaseAcpAgent { async closeSession(): Promise { this.commandOutputs.clear(); + this.nativeGoalTurnId = undefined; this.session.abortController.abort(); this.session.cancelled = true; this.planHandoffCancel?.(); @@ -1156,6 +1174,9 @@ export class CodexAppServerAgent extends BaseAcpAgent { if (method === APP_SERVER_NOTIFICATIONS.TURN_STARTED) { // Capture the active turn id (steer precondition / interrupt target). const turnId = (params as { turn?: { id?: string } })?.turn?.id; + if (!this.turns.isPending && turnId) { + this.nativeGoalTurnId = turnId; + } this.turns.onStarted(turnId); this.interruptQueuedGoalTurn(turnId); } @@ -1200,6 +1221,9 @@ export class CodexAppServerAgent extends BaseAcpAgent { this.commandOutputs.clear(); const turn = (params as { turn?: { id?: string; status?: string } }) ?.turn; + if (turn?.id === this.nativeGoalTurnId) { + this.nativeGoalTurnId = undefined; + } // Drop the late completion of an already-interrupted turn (else it cancels the follow-up). if (this.turns.shouldDropCompletion(turn?.id)) return; void this.finalizeTurn(mapTurnStopReason(turn?.status)); From 4a90decb13f63a1be0b0698a8b8dfff08e2a0c21 Mon Sep 17 00:00:00 2001 From: Georges-Antoine Assi Date: Wed, 15 Jul 2026 10:16:33 -0400 Subject: [PATCH 10/10] docs(auth): compact the added comment blocks Generated-By: PostHog Code Task-Id: f67230a6-4c1c-4433-91cc-ee487135fead --- packages/core/src/auth/auth.ts | 44 +++++++++++++--------------------- 1 file changed, 17 insertions(+), 27 deletions(-) diff --git a/packages/core/src/auth/auth.ts b/packages/core/src/auth/auth.ts index a0e4d2470e..9f9452ec04 100644 --- a/packages/core/src/auth/auth.ts +++ b/packages/core/src/auth/auth.ts @@ -86,8 +86,8 @@ export class AuthService extends TypedEventEmitter { private session: InMemorySession | null = null; private initializePromise: Promise | null = null; private refreshPromise: Promise | null = null; - // Serializes session-state commits so overlapping selections can't interleave - // across async encryption (see commitSessionState). + // Serializes session-state commits so overlapping selections can't + // interleave across async encryption (see commitSessionState). private commitChain: Promise = Promise.resolve(); constructor( @inject(AUTH_PREFERENCE_STORE) @@ -337,15 +337,10 @@ export class AuthService extends TypedEventEmitter { currentProjectId: number | null; }, ): Promise { - // Serialize commits onto a chain. Encryption is async, so two overlapping - // selections would otherwise interleave across the await — an earlier one - // completing last would clobber the newer stored session and published - // state (and persistSession's own read of the prior selection would race). - // Chaining runs each commit's persist-then-publish to completion before the - // next starts, so the latest selection wins consistently across the - // in-memory session, storage, and subscribers. The stored chain swallows - // rejections so one failed commit doesn't wedge later ones; the returned - // promise still rejects for the caller. + // Serialize commits onto a chain so overlapping selections can't + // interleave across async encryption and clobber a newer one. The chain + // swallows rejections so one failure doesn't wedge later commits; the + // returned promise still rejects for the caller. const run = this.commitChain.then(() => this.applyCommittedSession(prevSession, next), ); @@ -368,12 +363,10 @@ export class AuthService extends TypedEventEmitter { orgProjectsIncomplete: false, }; - // Persist the durable session first — it's the only step here that can - // fail (encryption is async and may reject). Mutating this.session, the - // project preference, or the published state before it would strand the - // service on a project change that the stored session and UI never - // committed. Commit those only after the persist resolves, so a rejection - // leaves every layer on the prior session. + // Persist the durable session first — the only step that can fail (async + // encryption may reject). Mutate this.session, the preference, and + // published state only after it resolves, so a rejection leaves every + // layer on the prior session. await this.persistSession({ refreshToken: nextSession.refreshToken, cloudRegion: nextSession.cloudRegion, @@ -533,13 +526,10 @@ export class AuthService extends TypedEventEmitter { return this.refreshPromise; } - // Assign refreshPromise synchronously — with no await between the guard - // above and this assignment — so concurrent callers dedupe onto one - // refresh. Resolving the stored session (which now awaits decryption) - // must therefore happen INSIDE refreshAndSync, not before it; otherwise - // two callers both pass the null guard, both decrypt, and both fire a - // refresh, burning the rotating refresh token twice and logging the user - // out when the second request fails. + // Assign refreshPromise synchronously — no await before this — so + // concurrent callers dedupe onto one refresh. Resolving the stored session + // (now async) must happen INSIDE refreshAndSync, else two callers both + // refresh and burn the rotating token twice. const refreshAndSync = async (): Promise => { const sessionInput = await this.getSessionInputForRefresh(); let session: InMemorySession; @@ -1118,9 +1108,9 @@ export class AuthService extends TypedEventEmitter { if (!stored) return; if (stored.scopeVersion < OAUTH_SCOPE_VERSION) return; - // Claim the recovery slot synchronously so concurrent triggers (network - // regained + tab resume) don't both kick a token refresh; decryptability - // is now async (Web Crypto), so it's validated inside recoverSession. + // Claim the recovery slot synchronously so concurrent triggers don't both + // kick a token refresh; decryptability is now async (Web Crypto), so it's + // validated inside recoverSession. this.recoveryPromise = this.recoverSession() .catch((error) => { this.logger.warn("Session recovery failed", { error });