diff --git a/apps/server/src/provider/Layers/CodexAdapter.test.ts b/apps/server/src/provider/Layers/CodexAdapter.test.ts index ea53c8ef88cf..44ebd6cbd440 100644 --- a/apps/server/src/provider/Layers/CodexAdapter.test.ts +++ b/apps/server/src/provider/Layers/CodexAdapter.test.ts @@ -42,6 +42,7 @@ import type { CodexAdapterShape } from "../Services/CodexAdapter.ts"; import { ProviderSessionDirectory } from "../Services/ProviderSessionDirectory.ts"; import { type CodexSessionRuntimeOptions, + type CodexSessionRuntimeError, type CodexSessionRuntimeSendTurnInput, type CodexSessionRuntimeShape, type CodexThreadSnapshot, @@ -104,6 +105,14 @@ class FakeCodexRuntime implements CodexSessionRuntimeShape { }), ); + public readonly revertThreadImpl = vi.fn( + (_beforeTurnId: TurnId): Promise => + Promise.resolve({ + threadId: "provider-thread-1", + turns: [], + }), + ); + public readonly deleteThreadImpl = vi.fn((): Promise => Promise.resolve(undefined)); public readonly uploadFeedbackImpl = vi.fn((_reason?: string) => Promise.resolve({ threadId: "provider-thread-1" }), @@ -147,6 +156,13 @@ class FakeCodexRuntime implements CodexSessionRuntimeShape { return Effect.promise(() => this.rollbackThreadImpl(numTurns)); } + revertThread(beforeTurnId: TurnId) { + return Effect.tryPromise({ + try: () => this.revertThreadImpl(beforeTurnId), + catch: (cause) => cause as CodexSessionRuntimeError, + }); + } + deleteThread = Effect.promise(() => this.deleteThreadImpl()); uploadFeedback(reason?: string) { return Effect.promise(() => this.uploadFeedbackImpl(reason)); @@ -389,21 +405,59 @@ sessionErrorLayer("CodexAdapterLive session errors", (it) => { turns: turnIds.map((id) => ({ id: asTurnId(id), items: [] })), }); const targetTurnId = asTurnId("turn-retracted"); - runtime.readThreadImpl - .mockResolvedValueOnce(snapshot(["turn-1", "turn-2", targetTurnId])) - .mockResolvedValue(snapshot(["turn-1", "turn-2"])); - runtime.rollbackThreadImpl.mockResolvedValue(snapshot(["turn-1", "turn-2"])); + runtime.revertThreadImpl.mockResolvedValue(snapshot(["turn-1", "turn-2"])); const first = yield* adapter.rollbackThreadTo(threadId, 3, targetTurnId); NodeAssert.deepStrictEqual( first.turns.map((turn) => turn.id), [asTurnId("turn-1"), asTurnId("turn-2")], ); - NodeAssert.deepStrictEqual(runtime.rollbackThreadImpl.mock.calls, [[1]]); + NodeAssert.deepStrictEqual(runtime.revertThreadImpl.mock.calls, [[targetTurnId]]); + NodeAssert.equal(runtime.readThreadImpl.mock.calls.length, 0); + NodeAssert.equal(runtime.rollbackThreadImpl.mock.calls.length, 0); const repeated = yield* adapter.rollbackThreadTo(threadId, 3, targetTurnId); NodeAssert.equal(repeated.turns.length, 2); + NodeAssert.deepStrictEqual(runtime.revertThreadImpl.mock.calls, [ + [targetTurnId], + [targetTurnId], + ]); + }), + ); + + it.effect("verifies a rollback from its response when thread reads still lag", () => + Effect.gen(function* () { + const adapter = yield* CodexAdapter; + const threadId = asThreadId("lagging-read-rollback-thread"); + yield* adapter.startSession({ + provider: ProviderDriverKind.make("codex"), + threadId, + runtimeMode: "full-access", + }); + const runtime = sessionRuntimeFactory.lastRuntime; + NodeAssert.ok(runtime); + NodeAssert.ok(adapter.rollbackThreadTo); + const snapshot = (turnIds: ReadonlyArray): CodexThreadSnapshot => ({ + threadId: "provider-thread-1", + turns: turnIds.map((id) => ({ id: asTurnId(id), items: [] })), + }); + const targetTurnId = asTurnId("turn-retracted"); + const staleSnapshot = snapshot(["turn-1", targetTurnId]); + const rolledBackSnapshot = snapshot(["turn-1"]); + runtime.readThreadImpl.mockResolvedValue(staleSnapshot); + runtime.rollbackThreadImpl.mockResolvedValue(rolledBackSnapshot); + runtime.revertThreadImpl.mockRejectedValue( + CodexErrors.CodexAppServerRequestError.methodNotFound("thread/revert"), + ); + + const result = yield* adapter.rollbackThreadTo(threadId, 1, targetTurnId); + + NodeAssert.deepStrictEqual( + result.turns.map((turn) => turn.id), + [asTurnId("turn-1")], + ); NodeAssert.deepStrictEqual(runtime.rollbackThreadImpl.mock.calls, [[1]]); + NodeAssert.equal(runtime.readThreadImpl.mock.calls.length, 1); }), ); diff --git a/apps/server/src/provider/Layers/CodexAdapter.ts b/apps/server/src/provider/Layers/CodexAdapter.ts index 5a99b910df0f..ab624207405a 100644 --- a/apps/server/src/provider/Layers/CodexAdapter.ts +++ b/apps/server/src/provider/Layers/CodexAdapter.ts @@ -33,6 +33,7 @@ import * as Exit from "effect/Exit"; import * as Fiber from "effect/Fiber"; import * as FileSystem from "effect/FileSystem"; import * as Queue from "effect/Queue"; +import * as Result from "effect/Result"; import * as Schema from "effect/Schema"; import * as Scope from "effect/Scope"; import * as Stream from "effect/Stream"; @@ -68,6 +69,7 @@ import { type EventNdjsonLogger, makeEventNdjsonLogger } from "./EventNdjsonLogg import { resolveCodexLaunchArgs } from "./codexLaunchArgs.ts"; const isCodexAppServerProcessExitedError = Schema.is(CodexErrors.CodexAppServerProcessExitedError); const isCodexAppServerTransportError = Schema.is(CodexErrors.CodexAppServerTransportError); +const isCodexAppServerRequestError = Schema.is(CodexErrors.CodexAppServerRequestError); const isCodexSessionRuntimeThreadIdMissingError = Schema.is( CodexSessionRuntimeThreadIdMissingError, ); @@ -75,6 +77,9 @@ const isCodexResumeCursorSchema = Schema.is(CodexResumeCursorSchema); const PROVIDER = ProviderDriverKind.make("codex"); +const isThreadRevertUnavailable = (error: CodexSessionRuntimeError): boolean => + isCodexAppServerRequestError(error) && error.code === -32601; + export interface CodexAdapterLiveOptions { readonly instanceId?: ProviderInstanceId; readonly environment?: NodeJS.ProcessEnv; @@ -1925,6 +1930,21 @@ export const makeCodexAdapter = Effect.fn("makeCodexAdapter")(function* ( }); } const session = yield* requireSession(threadId); + if (targetTurnId !== undefined) { + const revertResult = yield* session.runtime.revertThread(targetTurnId).pipe(Effect.result); + if (Result.isSuccess(revertResult)) { + if (session.lastStartedTurnId === targetTurnId) { + session.lastStartedTurnId = undefined; + } + return { + threadId, + turns: revertResult.success.turns, + }; + } + if (!isThreadRevertUnavailable(revertResult.failure)) { + return yield* mapCodexRuntimeError(threadId, "thread/revert", revertResult.failure); + } + } const current = yield* readThread(threadId); const targetIndex = targetTurnId === undefined ? -1 : current.turns.findIndex((turn) => turn.id === targetTurnId); @@ -1950,10 +1970,15 @@ export const makeCodexAdapter = Effect.fn("makeCodexAdapter")(function* ( : targetTurnId !== undefined ? 0 : current.turns.length - retainedTurnCount; - if (remainingDelta > 0) { - yield* rollbackThread(threadId, remainingDelta); - } - const verified = yield* readThread(threadId); + // thread/rollback returns the post-rollback snapshot. Use that response + // for the first verification instead of immediately calling thread/read: + // recent Codex app-server builds can briefly serve the pre-rollback turn + // from thread/read after the mutation has already committed. Treating + // that lagging read as failure leaves a durable T3 retraction pending and + // repeatedly rolls back the same provider turn. + const rolledBack = + remainingDelta > 0 ? yield* rollbackThread(threadId, remainingDelta) : undefined; + const verified = rolledBack ?? (yield* readThread(threadId)); if (targetTurnId !== undefined) { if (verified.turns.some((turn) => turn.id === targetTurnId)) { return yield* new ProviderAdapterRequestError({ diff --git a/apps/server/src/provider/Layers/CodexSessionRuntime.test.ts b/apps/server/src/provider/Layers/CodexSessionRuntime.test.ts index 138c572321eb..5621e8173507 100644 --- a/apps/server/src/provider/Layers/CodexSessionRuntime.test.ts +++ b/apps/server/src/provider/Layers/CodexSessionRuntime.test.ts @@ -4,7 +4,7 @@ import { it } from "@effect/vitest"; import * as Effect from "effect/Effect"; import * as Schema from "effect/Schema"; import { describe } from "vite-plus/test"; -import { DEFAULT_MODEL, ThreadId } from "@t3tools/contracts"; +import { DEFAULT_MODEL, ThreadId, TurnId } from "@t3tools/contracts"; import * as CodexErrors from "effect-codex-app-server/errors"; import * as CodexRpc from "effect-codex-app-server/rpc"; import * as EffectCodexSchema from "effect-codex-app-server/schema"; @@ -23,6 +23,7 @@ import { isRecoverableThreadResumeError, makeMemoryConsolidationNotificationFilter, openCodexThread, + revertCodexThread, toMcpElicitationResponse, } from "./CodexSessionRuntime.ts"; const isCodexAppServerRequestError = Schema.is(CodexErrors.CodexAppServerRequestError); @@ -875,3 +876,40 @@ describe("deleteCodexThread", () => { }), ); }); + +describe("revertCodexThread", () => { + it.effect("reverts paginated history before the target turn", () => + Effect.gen(function* () { + const calls: Array<{ method: string; payload: unknown }> = []; + const client = { + request: (method: string, payload?: unknown) => { + calls.push({ method, payload }); + return Effect.succeed({ + thread: { + id: "provider-thread-paginated", + turns: [], + }, + turnsBackwardsCursor: "turn-cursor", + }); + }, + }; + + const response = yield* revertCodexThread( + client, + "provider-thread-paginated", + TurnId.make("turn-retracted"), + ); + + NodeAssert.equal(response.thread.id, "provider-thread-paginated"); + NodeAssert.deepStrictEqual(calls, [ + { + method: "thread/revert", + payload: { + threadId: "provider-thread-paginated", + beforeTurnId: "turn-retracted", + }, + }, + ]); + }), + ); +}); diff --git a/apps/server/src/provider/Layers/CodexSessionRuntime.ts b/apps/server/src/provider/Layers/CodexSessionRuntime.ts index 3dab8e199920..411c9d02e141 100644 --- a/apps/server/src/provider/Layers/CodexSessionRuntime.ts +++ b/apps/server/src/provider/Layers/CodexSessionRuntime.ts @@ -195,6 +195,9 @@ export interface CodexSessionRuntimeShape { readonly rollbackThread: ( numTurns: number, ) => Effect.Effect; + readonly revertThread: ( + beforeTurnId: TurnId, + ) => Effect.Effect; readonly deleteThread: Effect.Effect; readonly uploadFeedback: ( reason?: string, @@ -693,12 +696,43 @@ interface CodexThreadDeleteClient { >; } +interface CodexThreadRevertClient { + readonly request: ( + method: string, + payload?: unknown, + ) => Effect.Effect; +} + +const CodexThreadRevertResponse = Schema.Struct({ + thread: Schema.Struct({ + id: Schema.String, + }), +}); +const decodeCodexThreadRevertResponse = Schema.decodeUnknownEffect(CodexThreadRevertResponse); + export const deleteCodexThread = ( client: CodexThreadDeleteClient, threadId: string, ): Effect.Effect => client.request("thread/delete", { threadId }).pipe(Effect.asVoid); +export const revertCodexThread = Effect.fn("revertCodexThread")(function* ( + client: CodexThreadRevertClient, + threadId: string, + beforeTurnId: TurnId, +) { + const response = yield* client.request("thread/revert", { threadId, beforeTurnId }); + return yield* decodeCodexThreadRevertResponse(response).pipe( + Effect.mapError((cause) => + CodexErrors.CodexAppServerProtocolParseError.fromSchemaError( + "decode-response-payload", + cause, + { method: "thread/revert" }, + ), + ), + ); +}); + export const openCodexThread = (input: { readonly client: CodexThreadOpenClient; readonly threadId: ThreadId; @@ -2218,6 +2252,19 @@ export const makeCodexSessionRuntime = ( }); return parseThreadSnapshot(response); }), + revertThread: (beforeTurnId) => + Effect.gen(function* () { + const providerThreadId = yield* readProviderThreadId; + const response = yield* revertCodexThread(client.raw, providerThreadId, beforeTurnId); + yield* updateSession(sessionRef, { + status: "ready", + activeTurnId: undefined, + }); + return { + threadId: response.thread.id, + turns: [], + }; + }), deleteThread: Effect.flatMap(readProviderThreadId, (providerThreadId) => deleteCodexThread(client, providerThreadId), ), diff --git a/apps/web/src/components/chat/RetractionRecoveryHandoff.test.ts b/apps/web/src/components/chat/RetractionRecoveryHandoff.test.ts index b88b2bb4bf6c..85faf0ea73cf 100644 --- a/apps/web/src/components/chat/RetractionRecoveryHandoff.test.ts +++ b/apps/web/src/components/chat/RetractionRecoveryHandoff.test.ts @@ -367,4 +367,33 @@ describe("retraction recovery handoff", () => { expect(signal).toBeNull(); expect(useComposerDraftStore.getState().getDraftSession(draftId)?.hidden).toBe(true); }); + + it("does not move an unresolved mid-thread message into a new-thread draft", async () => { + await seedRecovery(); + const persisted = useRetractionRecoveryStore.getState().byRequestId[requestId]; + if (!persisted) throw new Error("Expected recovery fixture"); + useRetractionRecoveryStore.setState({ + byRequestId: { + [requestId]: { ...persisted, firstUserMessage: false }, + }, + }); + const recovery = useRetractionRecoveryStore.getState().byRequestId[requestId]; + if (!recovery) throw new Error("Expected mid-thread recovery fixture"); + + const signal = resolveRetractionRecoverySignal({ + recovery, + liveCompletion: null, + projectedRetraction: null, + activities: [], + threadStatus: "live", + threadDetailExists: true, + shellSnapshotReady: true, + sourceThreadInShell: true, + nowMs: Date.parse(createdAt) + RETRACTION_RECOVERY_STALE_AFTER_MS * 2, + }); + + expect(signal).toBeNull(); + expect(useComposerDraftStore.getState().getDraftSession(draftId)?.hidden).toBe(true); + expect(useRetractionRecoveryStore.getState().byRequestId[requestId]).toBeDefined(); + }); }); diff --git a/apps/web/src/components/chat/RetractionRecoveryHandoff.tsx b/apps/web/src/components/chat/RetractionRecoveryHandoff.tsx index e149672db3a3..dc7549b0ceeb 100644 --- a/apps/web/src/components/chat/RetractionRecoveryHandoff.tsx +++ b/apps/web/src/components/chat/RetractionRecoveryHandoff.tsx @@ -129,7 +129,14 @@ export function resolveRetractionRecoverySignal(input: { const hasCorrelatedPendingRow = input.projectedRetraction?.requestId === input.recovery.requestId && input.projectedRetraction.status === "requested"; + // A mid-thread pop is already restored in the source composer. Never turn + // that message into a brand-new-thread draft just because the acknowledgement + // is late: keeping the recovery pending also keeps resends gated until the + // server reaches a terminal result. The draft fallback is only meaningful + // for a first-message retraction, whose successful outcome deletes its + // transient source thread, plus legacy records that predate this marker. const stale = + input.recovery.firstUserMessage !== false && input.threadDetailExists && !hasCorrelatedPendingRow && Number.isFinite(createdAtMs) &&