From 0ab26f98d3f987bc80326650e4b52b8bef2b21aa Mon Sep 17 00:00:00 2001 From: Javad Asgari Shafique Date: Wed, 9 Sep 2026 15:55:14 +0200 Subject: [PATCH 1/2] feat(codex): port native goals to orchestration V2 --- .../src/features/threads/ThreadComposer.tsx | 4 +- .../features/threads/ThreadDetailScreen.tsx | 118 ++++++++- apps/mobile/src/state/threads.ts | 18 +- apps/server/src/auth/RpcAuthorization.ts | 4 + .../Adapters/CodexAdapterV2.test.ts | 240 +++++++++++++++++- .../Adapters/CodexAdapterV2.ts | 170 ++++++++++++- .../src/orchestration-v2/ProviderAdapter.ts | 16 ++ .../src/provider/Layers/CodexProvider.ts | 5 + apps/server/src/ws.ts | 89 +++++++ apps/web/src/components/ChatView.tsx | 167 +++++++++++- apps/web/src/state/threads.ts | 18 +- docs/user/providers-codex.md | 15 ++ packages/client-runtime/package.json | 4 + packages/client-runtime/src/rpc/client.ts | 1 + .../src/state/threadCommands.test.ts | 148 +++++++++++ .../src/state/threadCommands.ts | 122 ++++++++- packages/contracts/src/codexGoal.test.ts | 20 ++ packages/contracts/src/codexGoal.ts | 83 ++++++ packages/contracts/src/index.ts | 2 + packages/contracts/src/rpc.ts | 43 ++++ 20 files changed, 1270 insertions(+), 17 deletions(-) create mode 100644 packages/client-runtime/src/state/threadCommands.test.ts create mode 100644 packages/contracts/src/codexGoal.test.ts create mode 100644 packages/contracts/src/codexGoal.ts diff --git a/apps/mobile/src/features/threads/ThreadComposer.tsx b/apps/mobile/src/features/threads/ThreadComposer.tsx index 80cb8b7f0570..8ab30a825d15 100644 --- a/apps/mobile/src/features/threads/ThreadComposer.tsx +++ b/apps/mobile/src/features/threads/ThreadComposer.tsx @@ -441,8 +441,8 @@ export const ThreadComposer = memo(function ThreadComposer(props: ThreadComposer if (inFlightThreadIdsRef.current.has(threadKey)) return; inFlightThreadIdsRef.current.add(threadKey); try { - const messageId = await onSendMessage(); - if (messageId === null) { + const sentMessageId = await onSendMessage(); + if (sentMessageId === null) { return; } // Sending a prompt starts agent work: arm the lock-screen card while the diff --git a/apps/mobile/src/features/threads/ThreadDetailScreen.tsx b/apps/mobile/src/features/threads/ThreadDetailScreen.tsx index 9ee3fa2e8ec9..32b678c3fe3c 100644 --- a/apps/mobile/src/features/threads/ThreadDetailScreen.tsx +++ b/apps/mobile/src/features/threads/ThreadDetailScreen.tsx @@ -1,3 +1,16 @@ +import { + formatCodexGoalDescription, + formatCodexGoalError, + formatCodexGoalStatus, + formatCodexGoalUsage, + parseCodexGoalCommand, + toCodexGoalSetInput, +} from "@t3tools/client-runtime/state/threadCommands"; +import { + isAtomCommandInterrupted, + squashAtomCommandFailure, +} from "@t3tools/client-runtime/state/runtime"; +import { AppText as Text } from "../../components/AppText"; import { type EnvironmentConnectionPhase } from "@t3tools/client-runtime/connection"; import type { EnvironmentThreadShell } from "@t3tools/client-runtime/state/shell"; import type { EnvironmentThreadStatus } from "@t3tools/client-runtime/state/threads"; @@ -33,6 +46,7 @@ import { useState, } from "react"; import { + Alert, AppState, Keyboard, Platform, @@ -62,7 +76,7 @@ import type { StatusTone } from "../../components/StatusPill"; import type { DraftComposerAttachment } from "../../lib/composerImages"; import { CHAT_CONTENT_MAX_WIDTH, type LayoutVariant } from "../../lib/layout"; import { scopedThreadKey } from "../../lib/scopedEntities"; -import { threadEnvironment } from "../../state/threads"; +import { threadEnvironment, useCodexGoal } from "../../state/threads"; import { useAtomCommand } from "../../state/use-atom-command"; import type { PendingApproval, @@ -275,7 +289,9 @@ export const ThreadDetailScreen = memo(function ThreadDetailScreen(props: Thread const selectedThreadKey = scopedThreadKey(props.environmentId, props.selectedThread.id); const composerEditorRef = useRef(null); const draftMessageRef = useRef(props.draftMessage); - draftMessageRef.current = props.draftMessage; + useLayoutEffect(() => { + draftMessageRef.current = props.draftMessage; + }, [props.draftMessage]); const composerOverlayRef = useRef(null); const listRef = useRef(null); const feedTouchStartRef = useRef<{ pageX: number; pageY: number } | null>(null); @@ -293,6 +309,11 @@ export const ThreadDetailScreen = memo(function ThreadDetailScreen(props: Thread const [anchorMessageId, setAnchorMessageId] = useState(null); const [submittedMessageId, setSubmittedMessageId] = useState(null); const [endFollowEnabled, setEndFollowEnabled] = useState(true); + const getCodexGoal = useAtomCommand(threadEnvironment.getCodexGoal, { reportFailure: false }); + const setCodexGoal = useAtomCommand(threadEnvironment.setCodexGoal, { reportFailure: false }); + const clearCodexGoal = useAtomCommand(threadEnvironment.clearCodexGoal, { + reportFailure: false, + }); // Android keys the safe-area padding on keyboard visibility (#5988): the // back gesture closes the keyboard while the editor stays focused, and a // focus-keyed inset would leave the toolbar under the gesture bar. iOS must @@ -408,7 +429,6 @@ export const ThreadDetailScreen = memo(function ThreadDetailScreen(props: Thread composerOverlayRef, Math.max(0, estimatedOverlayHeight - nativeInsetOvercount), -nativeInsetOvercount, - Platform.OS === "ios" ? COMPOSER_TRANSITION_DURATION_MS : 0, ); // The expanded questionnaire is an absolute overlay on iOS, so it never // changes the measured overlay height (that constancy is what keeps the @@ -543,6 +563,14 @@ export const ThreadDetailScreen = memo(function ThreadDetailScreen(props: Thread const isSplitLayout = layoutVariant === "split"; const contentMaxWidth = isSplitLayout ? CHAT_CONTENT_MAX_WIDTH : undefined; const selectedInstanceId = props.selectedThread.modelSelection.instanceId; + const selectedProvider = props.serverConfig?.providers.find( + (provider) => provider.instanceId === selectedInstanceId, + ); + const codexGoal = useCodexGoal( + selectedProvider?.driver === "codex" ? props.environmentId : null, + selectedProvider?.driver === "codex" ? props.selectedThread.id : null, + selectedProvider?.driver === "codex" ? selectedInstanceId : null, + ); useStreamingHaptics(props.selectedThread.id, props.selectedThreadFeed); const selectedProviderSkills = useMemo(() => { const provider = props.serverConfig?.providers.find( @@ -655,6 +683,69 @@ export const ThreadDetailScreen = memo(function ThreadDetailScreen(props: Thread ]); const handleSendMessage = useCallback(async () => { + const draftGoalCommand = + props.draftAttachments.length === 0 ? parseCodexGoalCommand(props.draftMessage) : null; + if (draftGoalCommand !== null && selectedProvider === undefined) { + Alert.alert("Provider still loading", "Wait for the provider list to finish loading."); + return null; + } + const goalCommand = selectedProvider?.driver === "codex" ? draftGoalCommand : null; + if (goalCommand !== null) { + if (goalCommand.action === "invalid") { + Alert.alert("Invalid Goal command", goalCommand.message); + return null; + } + const target = { + environmentId: props.environmentId, + input: { threadId: props.selectedThread.id }, + }; + const submittedDraft = props.draftMessage; + const submittedThreadKey = selectedThreadKey; + const stillOnSubmittedThread = () => selectedThreadKeyRef.current === submittedThreadKey; + const clearSubmittedGoalCommandDraft = () => { + if (!stillOnSubmittedThread() || draftMessageRef.current !== submittedDraft) return; + props.onChangeDraftMessage(""); + }; + if (goalCommand.action === "status") { + const result = await getCodexGoal(target); + if (result._tag === "Failure") { + if (!isAtomCommandInterrupted(result) && stillOnSubmittedThread()) { + Alert.alert( + "Codex Goal operation failed", + formatCodexGoalError(squashAtomCommandFailure(result)), + ); + } + return null; + } + clearSubmittedGoalCommandDraft(); + if (!stillOnSubmittedThread()) return null; + Alert.alert( + result.value === null + ? "No active Codex Goal" + : `Goal ${formatCodexGoalStatus(result.value.status)}`, + result.value === null ? undefined : formatCodexGoalDescription(result.value), + ); + return null; + } + const result = + goalCommand.action === "clear" + ? await clearCodexGoal(target) + : await setCodexGoal({ + environmentId: props.environmentId, + input: toCodexGoalSetInput(props.selectedThread.id, goalCommand), + }); + if (result._tag === "Failure") { + if (!isAtomCommandInterrupted(result) && stillOnSubmittedThread()) { + Alert.alert( + "Codex Goal operation failed", + formatCodexGoalError(squashAtomCommandFailure(result)), + ); + } + return null; + } + clearSubmittedGoalCommandDraft(); + return null; + } const targetThreadKey = selectedThreadKey; const hasUserMessage = selectedThreadFeed.some( (entry) => entry.type === "message" && entry.message.role === "user", @@ -678,9 +769,11 @@ export const ThreadDetailScreen = memo(function ThreadDetailScreen(props: Thread return messageId; }, [ anchorMessageId, - props.onSendMessage, - props.selectedThread.latestRun, - props.selectedThreadQueueCount, + clearCodexGoal, + getCodexGoal, + setCodexGoal, + selectedProvider, + props, selectedThreadFeed, selectedThreadKey, ]); @@ -876,6 +969,19 @@ export const ThreadDetailScreen = memo(function ThreadDetailScreen(props: Thread {/* Hidden (not unmounted) while a user-input request owns the composer slot, so composer drafts and editor state survive. */} + {codexGoal !== null ? ( + + + Goal {formatCodexGoalStatus(codexGoal.status)} + + + {codexGoal.objective} + + + {formatCodexGoalUsage(codexGoal)} + + + ) : null} (null)).pipe( + Atom.withLabel("mobile-codex-goal:empty"), +); + +export function useCodexGoal( + environmentId: EnvironmentId | null, + threadId: ThreadId | null, + providerInstanceId: ProviderInstanceId | null, +): CodexGoal | null { + const result = useAtomValue( + environmentId !== null && threadId !== null && providerInstanceId !== null + ? threadEnvironment.codexGoal({ environmentId, input: { threadId, providerInstanceId } }) + : EMPTY_CODEX_GOAL_ATOM, + ); + return Option.getOrNull(AsyncResult.value(result)); +} export function useEnvironmentThread( environmentId: EnvironmentId | null, diff --git a/apps/server/src/auth/RpcAuthorization.ts b/apps/server/src/auth/RpcAuthorization.ts index 86a051828b6b..9feb749575cd 100644 --- a/apps/server/src/auth/RpcAuthorization.ts +++ b/apps/server/src/auth/RpcAuthorization.ts @@ -113,6 +113,10 @@ export const RPC_REQUIRED_SCOPES = { [WS_METHODS.assetsPersistChatAttachments]: AuthOrchestrationOperateScope, [WS_METHODS.attachmentsCreateUploadUrl]: AuthOrchestrationOperateScope, [WS_METHODS.attachmentsDelete]: AuthOrchestrationOperateScope, + [WS_METHODS.codexGoalGet]: AuthOrchestrationReadScope, + [WS_METHODS.subscribeCodexGoal]: AuthOrchestrationReadScope, + [WS_METHODS.codexGoalSet]: AuthOrchestrationOperateScope, + [WS_METHODS.codexGoalClear]: AuthOrchestrationOperateScope, [WS_METHODS.providerUploadFeedback]: AuthOrchestrationOperateScope, [WS_METHODS.subscribeVcsStatus]: AuthOrchestrationReadScope, [WS_METHODS.subscribeResourceTelemetry]: AuthOrchestrationReadScope, diff --git a/apps/server/src/orchestration-v2/Adapters/CodexAdapterV2.test.ts b/apps/server/src/orchestration-v2/Adapters/CodexAdapterV2.test.ts index 3d9e927f4b8d..caff6604f4dc 100644 --- a/apps/server/src/orchestration-v2/Adapters/CodexAdapterV2.test.ts +++ b/apps/server/src/orchestration-v2/Adapters/CodexAdapterV2.test.ts @@ -1266,6 +1266,8 @@ describe("CodexAdapterV2 post-settle continuation", () => { const makeCodexReplayHarness = ( transcript: CodexReplay.CodexAppServerReplayTranscript, onEvent: (event: ProviderAdapterV2Event) => Effect.Effect = () => Effect.void, + onContinuation: (request: ProviderContinuationRequest) => Effect.Effect = () => + Effect.void, ) => Effect.gen(function* () { const fileSystem = yield* FileSystem.FileSystem; @@ -1300,7 +1302,7 @@ describe("CodexAdapterV2 post-settle continuation", () => { offer: (request) => Effect.sync(() => { continuationRequests.push(request); - }), + }).pipe(Effect.andThen(onContinuation(request))), }, }); const threadId = ThreadId.make(`thread-${transcript.scenario}`); @@ -1365,6 +1367,242 @@ describe("CodexAdapterV2 post-settle continuation", () => { event.type === "message.updated" && event.message.role === "assistant", ); + it.effect("adopts native goal continuations without sending a second turn/start", () => + Effect.scoped( + Effect.gen(function* () { + const nativeThreadId = "native-goal-thread"; + const originalTurnId = "goal-creation-turn"; + const nativeTurnId = "autonomous-goal-turn"; + const goal = { + threadId: nativeThreadId, + objective: "Finish the work", + status: "active", + tokensUsed: 0, + timeUsedSeconds: 0, + createdAt: 1782622440, + updatedAt: 1782622440, + }; + const offered = yield* Deferred.make(); + const finished = yield* Deferred.make(); + const transcript = makeCodexReplayTranscript({ + scenario: "native-goal-adoption", + entries: [ + ...codexReplayPreamble({ + nativeThreadId, + nativeTurnId: originalTurnId, + prompt: "Set a goal", + }), + { + type: "emit_inbound", + frame: { method: "thread/goal/updated", params: { threadId: nativeThreadId, goal } }, + }, + { + type: "emit_inbound", + frame: { + method: "turn/completed", + params: { + threadId: nativeThreadId, + turn: makeCodexReplayTurn({ id: originalTurnId, status: "completed" }), + }, + }, + }, + { + type: "emit_inbound", + frame: { + method: "turn/started", + params: { + threadId: nativeThreadId, + turn: makeCodexReplayTurn({ id: nativeTurnId, status: "inProgress" }), + }, + }, + }, + { + type: "emit_inbound", + frame: { + method: "item/completed", + params: { + threadId: nativeThreadId, + turnId: nativeTurnId, + item: { + type: "agentMessage", + id: "goal-result", + text: "Implemented and verified.", + phase: "final_answer", + }, + }, + }, + }, + { + type: "emit_inbound", + frame: { + method: "turn/completed", + params: { + threadId: nativeThreadId, + turn: makeCodexReplayTurn({ id: nativeTurnId, status: "completed" }), + }, + }, + }, + ], + }); + const harness = yield* makeCodexReplayHarness( + transcript, + (event) => + event.type === "turn.terminal" && String(event.providerTurnId).includes(nativeTurnId) + ? Deferred.succeed(finished, undefined) + : Effect.void, + () => Deferred.succeed(offered, undefined), + ); + const first = makeCodexTestTurnInput({ + threadId: harness.threadId, + providerThread: harness.providerThread, + now: yield* DateTime.now, + attemptId: RunAttemptId.make("first-goal-attempt"), + text: "Set a goal", + }); + yield* harness.runtime.startTurn(first); + yield* harness.firstTerminal; + yield* Deferred.await(offered); + assert.lengthOf(harness.continuationRequests, 1); + yield* harness.runtime.startTurn({ + ...first, + attemptId: RunAttemptId.make("next-goal-attempt"), + runId: RunId.make("next-goal-run"), + message: { + ...first.message, + createdBy: "agent", + creationSource: "provider", + text: "Continuing the active Codex goal.", + }, + }); + yield* Deferred.await(finished); + assert.isTrue( + assistantMessages(harness.events).some( + (event) => event.message.text === "Implemented and verified.", + ), + ); + assert.lengthOf(harness.terminalEvents(), 2); + }), + ).pipe(Effect.provide(Layer.mergeAll(NodeServices.layer, idAllocatorLayer))), + ); + + for (const pauseFails of [false, true]) { + it.effect( + `interrupts a goal turn after ${pauseFails ? "a failed pause" : "pausing the goal"}`, + () => + Effect.scoped( + Effect.gen(function* () { + const nativeThreadId = "goal-stop-thread"; + const nativeTurnId = "goal-stop-turn"; + const goal = { + threadId: nativeThreadId, + objective: "Finish the work", + status: "active", + tokensUsed: 0, + timeUsedSeconds: 0, + createdAt: 1782622440, + updatedAt: 1782622440, + }; + const transcript = makeCodexReplayTranscript({ + scenario: `goal-stop-${pauseFails}`, + entries: [ + ...codexReplayPreamble({ nativeThreadId, nativeTurnId, prompt: "Work" }), + { + type: "expect_outbound", + frame: { + id: 4, + method: "thread/goal/set", + params: { + threadId: nativeThreadId, + objective: goal.objective, + status: "active", + }, + }, + }, + { type: "emit_inbound", frame: { id: 4, result: { goal } } }, + { + type: "expect_outbound", + frame: { id: 5, method: "thread/goal/get", params: { threadId: nativeThreadId } }, + }, + { type: "emit_inbound", frame: { id: 5, result: { goal } } }, + { + type: "expect_outbound", + frame: { + id: 6, + method: "thread/goal/set", + params: { threadId: nativeThreadId, status: "paused" }, + }, + }, + { + type: "emit_inbound", + frame: pauseFails + ? { id: 6, error: { code: -32603, message: "Pause unavailable" } } + : { id: 6, result: { goal: { ...goal, status: "paused" } } }, + }, + { + type: "expect_outbound", + frame: { + id: 7, + method: "turn/interrupt", + params: { threadId: nativeThreadId, turnId: nativeTurnId }, + }, + }, + { type: "emit_inbound", frame: { id: 7, result: {} } }, + { + type: "emit_inbound", + frame: { + method: "turn/completed", + params: { + threadId: nativeThreadId, + turn: makeCodexReplayTurn({ id: nativeTurnId, status: "interrupted" }), + }, + }, + }, + { + type: "expect_outbound", + frame: { + id: 8, + method: "thread/goal/clear", + params: { threadId: nativeThreadId }, + }, + }, + { type: "emit_inbound", frame: { id: 8, result: { cleared: true } } }, + ], + }); + const started = yield* Deferred.make(); + const harness = yield* makeCodexReplayHarness(transcript, (event) => + event.type === "provider_turn.updated" + ? Deferred.succeed(started, event.providerTurn.id) + : Effect.void, + ); + yield* harness.runtime.startTurn( + makeCodexTestTurnInput({ + threadId: harness.threadId, + providerThread: harness.providerThread, + now: yield* DateTime.now, + attemptId: RunAttemptId.make("goal-stop-attempt"), + text: "Work", + }), + ); + const control = harness.runtime.codexGoal; + assert.isDefined(control); + const created = yield* control.set(harness.providerThread, { + objective: goal.objective, + status: "active", + }); + assert.equal(created.status, "active"); + yield* harness.runtime.interruptTurn({ + providerThread: harness.providerThread, + providerTurnId: yield* Deferred.await(started), + }); + yield* harness.firstTerminal; + assert.equal(harness.terminalEvents()[0]?.status, "interrupted"); + assert.deepEqual(yield* control.clear(harness.providerThread), { cleared: true }); + assert.isFalse(yield* harness.hasPendingBackgroundWork); + }), + ).pipe(Effect.provide(Layer.mergeAll(NodeServices.layer, idAllocatorLayer))), + ); + } + it.effect("keeps an asynchronous Codex question actionable after the turn completes", () => Effect.scoped( Effect.gen(function* () { diff --git a/apps/server/src/orchestration-v2/Adapters/CodexAdapterV2.ts b/apps/server/src/orchestration-v2/Adapters/CodexAdapterV2.ts index 792e573b9ecc..ba1fe9387ca5 100644 --- a/apps/server/src/orchestration-v2/Adapters/CodexAdapterV2.ts +++ b/apps/server/src/orchestration-v2/Adapters/CodexAdapterV2.ts @@ -1,3 +1,5 @@ +import { CodexGoal, type CodexGoalStreamEvent } from "@t3tools/contracts"; +import * as PubSub from "effect/PubSub"; import { mcpToolPresentation, type McpToolPresentation, @@ -1204,6 +1206,8 @@ export function codexThreadRuntimeParams(input: { }; } +const decodeCodexGoal = Schema.decodeUnknownEffect(CodexGoal); + const decodeCodexResumeMetadata = Schema.decodeUnknownEffect( Schema.Struct({ thread: Schema.Struct({ id: Schema.String, updatedAt: Schema.Number }) }), ); @@ -1547,6 +1551,48 @@ export function makeCodexAdapterV2(adapterOptions: CodexAdapterV2Options): Provi now, }); const events = yield* Queue.unbounded(); + const goalEvents = yield* PubSub.unbounded<{ + readonly nativeThreadId: string; + readonly goal: CodexGoal | null; + }>(); + const activeGoalThreads = new Set(); + const goalChanged = (nativeThreadId: string, goal: CodexGoal | null) => + Effect.gen(function* () { + if (goal?.status === "active") activeGoalThreads.add(nativeThreadId); + else activeGoalThreads.delete(nativeThreadId); + yield* PubSub.publish(goalEvents, { nativeThreadId, goal }); + }); + const getGoal = (providerThread: OrchestrationV2ProviderThread) => + Effect.gen(function* () { + const threadId = yield* getNativeThreadId(providerThread); + const response = yield* ensureInitialized.pipe( + Effect.andThen(client.request("thread/goal/get", { threadId })), + ); + const goal = + response.goal === null || response.goal === undefined + ? null + : yield* decodeCodexGoal(response.goal); + yield* goalChanged(threadId, goal); + return goal; + }).pipe(Effect.mapError((cause) => toProtocolError("Failed to read Codex goal.", cause))); + yield* client.handleServerNotification("thread/goal/updated", (payload) => + decodeCodexGoal(payload.goal).pipe( + Effect.flatMap((goal) => goalChanged(payload.threadId, goal)), + Effect.orDie, + ), + ); + yield* client.handleServerNotification("thread/goal/cleared", (payload) => + goalChanged(payload.threadId, null), + ); + const rootInputsByNativeThread = new Map(); + const nativeGoalTurns = new Map< + string, + { + readonly nativeTurnId: string; + readonly startedAt: DateTime.Utc; + readonly ready: Deferred.Deferred; + } + >(); const activeTurns = yield* Ref.make(new Map()); const turnTokenUsageByThread = new Map(); const usageStateForThread = (nativeThreadId: string) => { @@ -1664,6 +1710,8 @@ export function makeCodexAdapterV2(adapterOptions: CodexAdapterV2Options): Provi readonly startedAt: DateTime.Utc; }) => Effect.gen(function* () { + const nativeThreadId = yield* getNativeThreadId(input.turnInput.providerThread); + rootInputsByNativeThread.set(nativeThreadId, input.turnInput); const existing = (yield* Ref.get(activeTurns)).get(input.nativeTurnId); if (existing !== undefined) { return existing; @@ -1731,6 +1779,10 @@ export function makeCodexAdapterV2(adapterOptions: CodexAdapterV2Options): Provi attemptsRemaining = 1_000, ): Effect.Effect => Effect.gen(function* () { + const pendingGoal = Array.from(nativeGoalTurns.values()).find( + (turn) => turn.nativeTurnId === nativeTurnId, + ); + if (pendingGoal !== undefined) yield* Deferred.await(pendingGoal.ready); const context = (yield* Ref.get(activeTurns)).get(nativeTurnId); if (context !== undefined || attemptsRemaining <= 0) { return context; @@ -3398,7 +3450,7 @@ export function makeCodexAdapterV2(adapterOptions: CodexAdapterV2Options): Provi yield* client.handleServerNotification("item/agentMessage/delta", (payload) => Effect.gen(function* () { - const context = (yield* Ref.get(activeTurns)).get(payload.turnId); + const context = yield* awaitActiveTurn(payload.turnId); if (context !== undefined) { yield* completeProviderRetry(context, yield* DateTime.now); } @@ -3556,6 +3608,40 @@ export function makeCodexAdapterV2(adapterOptions: CodexAdapterV2Options): Provi }); return; } + const rootInput = rootInputsByNativeThread.get(payload.threadId); + if ( + rootInput !== undefined && + activeGoalThreads.has(payload.threadId) && + continuationRequests !== undefined + ) { + const previous = nativeGoalTurns.get(payload.threadId); + if (previous?.nativeTurnId === payload.turn.id) return; + if (previous !== undefined) yield* Deferred.await(previous.ready); + const pending = { + nativeTurnId: payload.turn.id, + startedAt: codexTimestamp(payload.turn.startedAt), + ready: yield* Deferred.make(), + }; + nativeGoalTurns.set(payload.threadId, pending); + const clear = Effect.gen(function* () { + if (nativeGoalTurns.get(payload.threadId) !== pending) return; + nativeGoalTurns.delete(payload.threadId); + yield* Deferred.succeed(pending.ready, undefined); + }); + yield* continuationRequests.offer({ + threadId: rootInput.threadId, + providerThreadId: rootInput.providerThread.id, + driver: CODEX_PROVIDER, + detail: "Continuing the active Codex goal.", + delivery: "adapter_buffered", + clearIfCurrent: () => clear, + dispatchIfCurrent: (effect) => + nativeGoalTurns.get(payload.threadId) === pending + ? effect.pipe(Effect.map(Option.some)) + : Effect.succeed(Option.none()), + }); + return; + } yield* rememberSubagentTurnStarted({ nativeThreadId: payload.threadId, nativeTurnId: payload.turn.id, @@ -4818,7 +4904,7 @@ export function makeCodexAdapterV2(adapterOptions: CodexAdapterV2Options): Provi yield* client.handleServerNotification("turn/completed", (payload) => Effect.gen(function* () { - const context = (yield* Ref.get(activeTurns)).get(payload.turn.id); + const context = yield* awaitActiveTurn(payload.turn.id); if (context === undefined) { return; } @@ -4851,6 +4937,7 @@ export function makeCodexAdapterV2(adapterOptions: CodexAdapterV2Options): Provi // against a long-delayed resume. Codex emits no resume-expected // signal to pin on. hasPendingBackgroundWork: Effect.gen(function* () { + if (activeGoalThreads.size > 0 || nativeGoalTurns.size > 0) return true; for (const items of (yield* Ref.get(runningCommandItemsByTurn)).values()) { if (items.size > 0) { return true; @@ -4868,6 +4955,56 @@ export function makeCodexAdapterV2(adapterOptions: CodexAdapterV2Options): Provi } return false; }), + codexGoal: { + get: getGoal, + set: (providerThread, goalInput) => + Effect.gen(function* () { + const threadId = yield* getNativeThreadId(providerThread); + const response = yield* ensureInitialized.pipe( + Effect.andThen(client.request("thread/goal/set", { ...goalInput, threadId })), + ); + const goal = yield* decodeCodexGoal(response.goal); + yield* goalChanged(threadId, goal); + return goal; + }).pipe( + Effect.mapError((cause) => toProtocolError("Failed to update Codex goal.", cause)), + ), + clear: (providerThread) => + Effect.gen(function* () { + const threadId = yield* getNativeThreadId(providerThread); + const response = yield* ensureInitialized.pipe( + Effect.andThen(client.request("thread/goal/clear", { threadId })), + ); + yield* goalChanged(threadId, null); + return response; + }).pipe( + Effect.mapError((cause) => toProtocolError("Failed to clear Codex goal.", cause)), + ), + subscribe: (providerThread) => + Stream.unwrap( + Effect.gen(function* () { + const nativeThreadId = yield* getNativeThreadId(providerThread); + const subscription = yield* PubSub.subscribe(goalEvents); + const goal = yield* getGoal(providerThread); + const threadId = providerThread.appThreadId!; + return Stream.concat( + Stream.make({ + type: "snapshot", + threadId, + goal, + } satisfies CodexGoalStreamEvent), + Stream.fromSubscription(subscription).pipe( + Stream.filter((event) => event.nativeThreadId === nativeThreadId), + Stream.map((event): CodexGoalStreamEvent => + event.goal === null + ? { type: "cleared", threadId } + : { type: "updated", threadId, goal: event.goal }, + ), + ), + ); + }), + ), + }, ensureThread: (threadInput) => ensureInitialized.pipe( Effect.andThen( @@ -4976,6 +5113,25 @@ export function makeCodexAdapterV2(adapterOptions: CodexAdapterV2Options): Provi startTurn: (turnInput) => Effect.gen(function* () { const threadId = yield* getNativeThreadId(turnInput.providerThread); + const goalTurn = nativeGoalTurns.get(threadId); + if (goalTurn !== undefined) { + if ( + turnInput.message.createdBy !== "agent" || + turnInput.message.creationSource !== "provider" + ) { + return yield* toProtocolError( + "A native goal continuation is being attached. Retry this message when it is running.", + ); + } + yield* registerRootTurn({ + turnInput, + nativeTurnId: goalTurn.nativeTurnId, + startedAt: goalTurn.startedAt, + }); + nativeGoalTurns.delete(threadId); + yield* Deferred.succeed(goalTurn.ready, undefined); + return; + } const codexInput = turnInput.restartContinuationOfRunId === undefined @@ -5056,6 +5212,16 @@ export function makeCodexAdapterV2(adapterOptions: CodexAdapterV2Options): Provi ), interruptTurn: (turnInput) => Effect.gen(function* () { + yield* Effect.gen(function* () { + const threadId = yield* getNativeThreadId(turnInput.providerThread); + if (!activeGoalThreads.has(threadId)) return; + const goal = yield* getGoal(turnInput.providerThread); + if (goal?.status === "active") { + yield* client.request("thread/goal/set", { threadId, status: "paused" }); + yield* goalChanged(threadId, { ...goal, status: "paused" }); + } + }).pipe(Effect.timeout("1 second"), Effect.ignore); + const activeTurnContexts = Array.from((yield* Ref.get(activeTurns)).values()); const activeTurn = activeTurnContexts.find( (candidate) => candidate.providerTurnId === turnInput.providerTurnId, diff --git a/apps/server/src/orchestration-v2/ProviderAdapter.ts b/apps/server/src/orchestration-v2/ProviderAdapter.ts index 36fc02078c8e..1bbc44290633 100644 --- a/apps/server/src/orchestration-v2/ProviderAdapter.ts +++ b/apps/server/src/orchestration-v2/ProviderAdapter.ts @@ -1,3 +1,4 @@ +import type { CodexGoal, CodexGoalSetInput, CodexGoalStreamEvent } from "@t3tools/contracts"; import { ChatAttachment, CheckpointId, @@ -495,6 +496,21 @@ export interface ProviderAdapterV2SessionRuntime { readonly hasPendingBackgroundWorkForThread?: ( providerThread: OrchestrationV2ProviderThread, ) => Effect.Effect; + readonly codexGoal?: { + readonly get: ( + providerThread: OrchestrationV2ProviderThread, + ) => Effect.Effect; + readonly set: ( + providerThread: OrchestrationV2ProviderThread, + input: Omit, + ) => Effect.Effect; + readonly clear: ( + providerThread: OrchestrationV2ProviderThread, + ) => Effect.Effect<{ readonly cleared: boolean }, ProviderAdapterV2Error>; + readonly subscribe: ( + providerThread: OrchestrationV2ProviderThread, + ) => Stream.Stream; + }; readonly ensureThread: ( input: ProviderAdapterV2EnsureThreadInput, ) => Effect.Effect; diff --git a/apps/server/src/provider/Layers/CodexProvider.ts b/apps/server/src/provider/Layers/CodexProvider.ts index d65a09c6d4f9..65cb67c368b7 100644 --- a/apps/server/src/provider/Layers/CodexProvider.ts +++ b/apps/server/src/provider/Layers/CodexProvider.ts @@ -664,6 +664,11 @@ export const checkCodexProviderStatus = Effect.fn("checkCodexProviderStatus")(fu skills: snapshot.skills, slashCommands: [ COMPACT_SLASH_COMMAND, + { + name: "goal", + description: "Manage the native Codex goal", + input: { hint: "create, status, steer, pause, resume, clear, or reset" }, + }, { name: "feedback", description: "Send this thread and Codex logs to OpenAI", diff --git a/apps/server/src/ws.ts b/apps/server/src/ws.ts index 10b593a863fc..4f89855336df 100644 --- a/apps/server/src/ws.ts +++ b/apps/server/src/ws.ts @@ -1,3 +1,8 @@ +import { + CodexGoalOperationError, + type CodexGoalOperation, + type ThreadId as CodexGoalThreadId, +} from "@t3tools/contracts"; import * as DateTime from "effect/DateTime"; import * as Duration from "effect/Duration"; import * as Encoding from "effect/Encoding"; @@ -556,6 +561,37 @@ const makeWsRpcLayer = ( return true; }); const providerSessionsV2 = yield* ProviderSessionManagerV2; + const resolveCodexGoalRuntime = ( + threadId: CodexGoalThreadId, + operation: CodexGoalOperation, + ) => + Effect.gen(function* () { + const projection = yield* threadManagement.getThreadProjection(threadId); + const providerThread = projection.providerThreads.find( + (thread) => thread.id === projection.thread.activeProviderThreadId, + ); + if (providerThread?.driver !== "codex" || providerThread.providerSessionId === null) { + return yield* new CodexGoalOperationError({ + threadId, + operation, + cause: "Start a Codex thread before managing its goal.", + }); + } + const runtime = Option.getOrNull( + yield* providerSessionsV2.get(providerThread.providerSessionId), + ); + if (runtime?.codexGoal === undefined) { + return yield* new CodexGoalOperationError({ + threadId, + operation, + cause: "The Codex session is not running. Resume the thread first.", + }); + } + return { providerThread, goal: runtime.codexGoal }; + }).pipe( + Effect.mapError((cause) => new CodexGoalOperationError({ threadId, operation, cause })), + ); + const analytics = yield* AnalyticsService.AnalyticsService; // Client-origin attribution (#7774): every thread/turn the connecting // client starts is credited to its surface + app version. Best-effort: @@ -1630,6 +1666,59 @@ const makeWsRpcLayer = ( ).pipe(Effect.map((providers) => ({ providers }))), { "rpc.aggregate": "server" }, ), + [WS_METHODS.codexGoalGet]: (input) => + resolveCodexGoalRuntime(input.threadId, "get").pipe( + Effect.flatMap(({ providerThread, goal }) => goal.get(providerThread)), + Effect.mapError( + (cause) => + new CodexGoalOperationError({ threadId: input.threadId, operation: "get", cause }), + ), + ), + [WS_METHODS.codexGoalSet]: (input) => + resolveCodexGoalRuntime(input.threadId, "set").pipe( + Effect.flatMap(({ providerThread, goal }) => goal.set(providerThread, input)), + Effect.mapError( + (cause) => + new CodexGoalOperationError({ threadId: input.threadId, operation: "set", cause }), + ), + ), + [WS_METHODS.codexGoalClear]: (input) => + resolveCodexGoalRuntime(input.threadId, "clear").pipe( + Effect.flatMap(({ providerThread, goal }) => goal.clear(providerThread)), + Effect.mapError( + (cause) => + new CodexGoalOperationError({ + threadId: input.threadId, + operation: "clear", + cause, + }), + ), + ), + [WS_METHODS.subscribeCodexGoal]: (input) => + Stream.unwrap( + resolveCodexGoalRuntime(input.threadId, "subscribe").pipe( + Effect.flatMap(({ providerThread, goal }) => + providerThread.providerInstanceId === input.providerInstanceId + ? Effect.succeed(goal.subscribe(providerThread)) + : Effect.fail( + new CodexGoalOperationError({ + threadId: input.threadId, + operation: "subscribe", + cause: "The thread's provider has changed.", + }), + ), + ), + ), + ).pipe( + Stream.mapError( + (cause) => + new CodexGoalOperationError({ + threadId: input.threadId, + operation: "subscribe", + cause, + }), + ), + ), [WS_METHODS.providerUploadFeedback]: (input) => observeRpcEffect( WS_METHODS.providerUploadFeedback, diff --git a/apps/web/src/components/ChatView.tsx b/apps/web/src/components/ChatView.tsx index f31a509ab708..86346106bbb7 100644 --- a/apps/web/src/components/ChatView.tsx +++ b/apps/web/src/components/ChatView.tsx @@ -1,7 +1,14 @@ +import { + formatCodexGoalDescription, + formatCodexGoalError, + formatCodexGoalStatus, + parseCodexGoalCommand, + toCodexGoalSetInput, +} from "@t3tools/client-runtime/state/threadCommands"; import { getTerminalLabel } from "@t3tools/shared/terminalLabels"; import * as Schema from "effect/Schema"; import { confirmTerminalClose, isTerminalCloseConfirmPending } from "../lib/terminalCloseConfirm"; -import { Minimize2Icon } from "lucide-react"; +import { Minimize2Icon, TargetIcon } from "lucide-react"; import { type AssistantCitation, type ChatFileAttachment, @@ -266,7 +273,7 @@ import { serverEnvironment, } from "../state/server"; import { terminalEnvironment } from "../state/terminal"; -import { threadEnvironment } from "../state/threads"; +import { threadEnvironment, useCodexGoal } from "../state/threads"; import { resolveProviderSkillsForCwd } from "@t3tools/client-runtime/providerSkills"; import { vcsEnvironment } from "../state/vcs"; import { useEnvironments, usePrimaryEnvironment } from "../state/environments"; @@ -1407,6 +1414,9 @@ export default function ChatView(props: ChatViewProps) { const forkThreadFromRun = useAtomCommand(threadEnvironment.forkFromRun, { reportFailure: false, }); + const getCodexGoal = useAtomCommand(threadEnvironment.getCodexGoal, { reportFailure: false }); + const setCodexGoal = useAtomCommand(threadEnvironment.setCodexGoal, { reportFailure: false }); + const clearCodexGoal = useAtomCommand(threadEnvironment.clearCodexGoal, { reportFailure: false }); const openPreview = useAtomCommand(previewEnvironment.open, { reportFailure: false }); const closePreview = useAtomCommand(previewEnvironment.close, "preview close"); const { environments } = useEnvironments(); @@ -1628,6 +1638,10 @@ export default function ChatView(props: ChatViewProps) { const feedbackUploading = feedbackSubmissions.some( (submission) => submission.status === "uploading", ); + const [goalCommandThreadKeysInFlight, setGoalCommandThreadKeysInFlight] = useState< + ReadonlySet + >(() => new Set()); + const goalCommandRunning = goalCommandThreadKeysInFlight.has(routeThreadKey); const optimisticUserMessagesRef = useRef(optimisticUserMessages); optimisticUserMessagesRef.current = optimisticUserMessages; const [localDraftErrorsByDraftId, setLocalDraftErrorsByDraftId] = useState< @@ -1709,6 +1723,7 @@ export default function ChatView(props: ChatViewProps) { const attachmentPreviewPromotionInFlightByMessageIdRef = useRef>({}); const sendInFlightRef = useRef(false); const environmentUnavailableSendToastSlotRef = useRef(0); + const goalCommandsInFlightRef = useRef(new Set()); const feedbackUploadsInFlightRef = useRef(new Set()); const terminalUiOpenByThreadRef = useRef>({}); @@ -1937,6 +1952,10 @@ export default function ChatView(props: ChatViewProps) { [activeThread], ); const activeThreadKey = activeThreadRef ? scopedThreadKey(activeThreadRef) : null; + const activeThreadKeyRef = useRef(activeThreadKey); + useLayoutEffect(() => { + activeThreadKeyRef.current = activeThreadKey; + }, [activeThreadKey]); const activeThreadShell = useThreadShell(isServerThread ? activeThreadRef : null); const changeRequestSnapshotByKey = useAtomValue(threadChangeRequestSnapshotsAtom); const [timelineAnchor, setTimelineAnchor] = useState<{ @@ -2711,6 +2730,11 @@ export default function ChatView(props: ChatViewProps) { ], ); const selectedProvider = selectedProviderEntry?.driverKind ?? requestedDriverKind; + const codexGoal = useCodexGoal( + environmentId, + selectedProvider === "codex" && isServerThread ? activeThreadId : null, + selectedProvider === "codex" ? (activeThread?.modelSelection.instanceId ?? null) : null, + ); const activeProviderInstanceId = selectedProviderEntry?.instanceId ?? null; const activeProviderStatus = selectedProviderEntry?.snapshot ?? null; const { enabled: interactionModeEnabled, interactionMode } = resolveComposerInteractionMode({ @@ -5976,6 +6000,24 @@ export default function ChatView(props: ChatViewProps) { resumeCompactionPermanentlyDismissed, selectedProvider, ]); + const codexGoalBannerItem = useMemo(() => { + if (codexGoal === null) return null; + const goalDescription = formatCodexGoalDescription(codexGoal); + return { + id: `codex-goal:${activeThread?.id ?? "unknown"}`, + variant: "info", + icon: , + title: `Goal ${formatCodexGoalStatus(codexGoal.status)}`, + description: ( + + {goalDescription}} /> + + {goalDescription} + + + ), + }; + }, [activeThread?.id, codexGoal]); const handleRestoreThreadBranch = useCallback(() => { if (gitStatusQuery.data?.hasWorkingTreeChanges) { setBranchRestoreConfirmOpen(true); @@ -5984,6 +6026,7 @@ export default function ChatView(props: ChatViewProps) { void handleSwitchCheckoutToThread(); }, [gitStatusQuery.data?.hasWorkingTreeChanges, handleSwitchCheckoutToThread]); const composerBannerItems = useMemo(() => { + const codexGoalItems = codexGoalBannerItem === null ? [] : [codexGoalBannerItem]; const backgroundWorkItems = backgroundWorkBannerItem === null ? [] : [backgroundWorkBannerItem]; const resumeCompactionItems = resumeCompactionBannerItem === null ? [] : [resumeCompactionBannerItem]; @@ -5996,6 +6039,7 @@ export default function ChatView(props: ChatViewProps) { ...resumeCompactionItems, ...wokeThreadItems, ...parkedThreadItems, + ...codexGoalItems, ]; } return [ @@ -6042,9 +6086,11 @@ export default function ChatView(props: ChatViewProps) { }, }, ...parkedThreadItems, + ...codexGoalItems, ]; }, [ activeBranchMismatchKey, + codexGoalBannerItem, handleRestoreThreadBranch, isRestoringThreadBranch, backgroundWorkBannerItem, @@ -6589,6 +6635,7 @@ export default function ChatView(props: ChatViewProps) { isSendBusy || isConnecting || sendInFlightRef.current || + goalCommandsInFlightRef.current.has(routeThreadKey) || feedbackUploadsInFlightRef.current.has(routeThreadKey) ) { notifyDirectAnnotationAttached(); @@ -6734,6 +6781,114 @@ export default function ChatView(props: ChatViewProps) { composerPreviewAnnotations.length + composerReviewComments.length, }); + const isUnadornedCodexCommand = + selectedProvider === "codex" && + !composerHasAttachments && + !directAnnotation && + composerTerminalContexts.length === 0 && + composerElementContexts.length === 0 && + composerPreviewAnnotations.length === 0 && + composerReviewComments.length === 0; + const codexGoalCommand = isUnadornedCodexCommand ? parseCodexGoalCommand(trimmed) : null; + if (codexGoalCommand !== null) { + if (codexGoalCommand.action === "invalid") { + toastManager.add( + stackedThreadToast({ + type: "warning", + title: "Invalid Goal command", + description: codexGoalCommand.message, + }), + ); + return; + } + if (!isServerThread || activeThreadId === null) { + toastManager.add( + stackedThreadToast({ + type: "warning", + title: "Start the Codex thread first", + description: "Send a message before managing its native Goal.", + }), + ); + return; + } + + const target = { environmentId, input: { threadId: activeThreadId } }; + const submittedThreadKey = activeThreadKey; + const submittedGoalCommandThreadKey = routeThreadKey; + const stillOnSubmittedThread = () => activeThreadKeyRef.current === submittedThreadKey; + const clearSubmittedGoalCommandDraft = () => { + if (!stillOnSubmittedThread() || promptRef.current !== promptForSend) return; + promptRef.current = ""; + clearComposerDraftContent(composerDraftTarget); + composerRef.current?.resetCursorState(); + }; + goalCommandsInFlightRef.current.add(submittedGoalCommandThreadKey); + setGoalCommandThreadKeysInFlight((current) => { + const next = new Set(current); + next.add(submittedGoalCommandThreadKey); + return next; + }); + try { + if (codexGoalCommand.action === "status") { + const result = await getCodexGoal(target); + if (result._tag === "Failure") { + if (!isAtomCommandInterrupted(result) && stillOnSubmittedThread()) { + toastManager.add( + stackedThreadToast({ + type: "error", + title: "Codex Goal operation failed", + description: formatCodexGoalError(squashAtomCommandFailure(result)), + }), + ); + } + return; + } + clearSubmittedGoalCommandDraft(); + if (!stillOnSubmittedThread()) return; + toastManager.add( + stackedThreadToast( + result.value === null + ? { type: "info", title: "No active Codex Goal" } + : { + type: "info", + title: `Goal ${formatCodexGoalStatus(result.value.status)}`, + description: formatCodexGoalDescription(result.value), + }, + ), + ); + return; + } + const result = + codexGoalCommand.action === "clear" + ? await clearCodexGoal(target) + : await setCodexGoal({ + environmentId, + input: toCodexGoalSetInput(activeThreadId, codexGoalCommand), + }); + if (result._tag === "Failure") { + if (!isAtomCommandInterrupted(result) && stillOnSubmittedThread()) { + toastManager.add( + stackedThreadToast({ + type: "error", + title: "Codex Goal operation failed", + description: formatCodexGoalError(squashAtomCommandFailure(result)), + }), + ); + } + return; + } + + clearSubmittedGoalCommandDraft(); + return; + } finally { + goalCommandsInFlightRef.current.delete(submittedGoalCommandThreadKey); + setGoalCommandThreadKeysInFlight((current) => { + const next = new Set(current); + next.delete(submittedGoalCommandThreadKey); + return next; + }); + } + } const feedbackCommand = ctxSelectedProvider === "codex" && composerImages.length === 0 && @@ -8487,7 +8642,13 @@ export default function ChatView(props: ChatViewProps) { activeContextWindow={activeContextWindow} activeTasksProgress={activeComposerTasksProgress} activeTaskSteps={activeComposerTaskSteps} - sendDisabledReason={feedbackUploading ? "Sending feedback" : null} + sendDisabledReason={ + feedbackUploading + ? "Sending feedback" + : goalCommandRunning + ? "Running Goal command" + : null + } compactThreadUnavailable={compactThreadUnavailable} compactDisabled={compactDisabled} compactDisabledReason={compactDisabledReason} diff --git a/apps/web/src/state/threads.ts b/apps/web/src/state/threads.ts index d64a6ac2163d..70d6f6eb0513 100644 --- a/apps/web/src/state/threads.ts +++ b/apps/web/src/state/threads.ts @@ -7,7 +7,7 @@ import { type EnvironmentThreadState, createThreadEnvironmentAtoms, } from "@t3tools/client-runtime/state/threads"; -import type { EnvironmentId, ThreadId } from "@t3tools/contracts"; +import type { CodexGoal, EnvironmentId, ProviderInstanceId, ThreadId } from "@t3tools/contracts"; import * as Option from "effect/Option"; import { AsyncResult, Atom } from "effect/unstable/reactivity"; @@ -28,6 +28,22 @@ export const environmentThreadShells = createEnvironmentThreadShellAtoms({ const EMPTY_THREAD_STATE_ATOM = Atom.make(AsyncResult.success(EMPTY_ENVIRONMENT_THREAD_STATE)).pipe( Atom.withLabel("web-environment-thread:empty"), ); +const EMPTY_CODEX_GOAL_ATOM = Atom.make(AsyncResult.success(null)).pipe( + Atom.withLabel("web-codex-goal:empty"), +); + +export function useCodexGoal( + environmentId: EnvironmentId | null, + threadId: ThreadId | null, + providerInstanceId: ProviderInstanceId | null, +): CodexGoal | null { + const result = useAtomValue( + environmentId !== null && threadId !== null && providerInstanceId !== null + ? threadEnvironment.codexGoal({ environmentId, input: { threadId, providerInstanceId } }) + : EMPTY_CODEX_GOAL_ATOM, + ); + return Option.getOrNull(AsyncResult.value(result)); +} export function useEnvironmentThread( environmentId: EnvironmentId | null, diff --git a/docs/user/providers-codex.md b/docs/user/providers-codex.md index 417287cc0032..d28116a786c3 100644 --- a/docs/user/providers-codex.md +++ b/docs/user/providers-codex.md @@ -69,3 +69,18 @@ In an existing Codex thread, send `/feedback` with an optional description, for example `/feedback The agent stopped before finishing the tests`. This uploads the conversation and Codex logs to OpenAI. The returned thread ID can be shared with OpenAI support. + +## Manage a Codex Goal + +In a running Codex session, use `/goal create ` to keep Codex working +across turns toward an objective. Native continuation turns appear in the thread +as work progresses. This requires a Codex version with native Goal support. + +Use `/goal status` to inspect progress, `/goal steer ` to change the +objective, `/goal pause` to pause, and `/goal resume` to continue. `/goal clear` +removes the goal; `/goal reset` is an alias for clearing it. Send these commands +without attachments. The goal is stored by Codex and shared across clients. + +Stop attempts to pause an active goal before interrupting its current turn. +Goal controls require a running provider session; after a server or provider +restart, send a normal message to reconnect the thread first. diff --git a/packages/client-runtime/package.json b/packages/client-runtime/package.json index da7b17114d76..4aeaf4dbabf7 100644 --- a/packages/client-runtime/package.json +++ b/packages/client-runtime/package.json @@ -187,6 +187,10 @@ "types": "./src/state/threads.ts", "default": "./src/state/threads.ts" }, + "./state/threadCommands": { + "types": "./src/state/threadCommands.ts", + "default": "./src/state/threadCommands.ts" + }, "./state/subagentRuntime": { "types": "./src/state/subagentRuntime.ts", "default": "./src/state/subagentRuntime.ts" diff --git a/packages/client-runtime/src/rpc/client.ts b/packages/client-runtime/src/rpc/client.ts index 7c72d68fc0a4..d92af9dd0eff 100644 --- a/packages/client-runtime/src/rpc/client.ts +++ b/packages/client-runtime/src/rpc/client.ts @@ -54,6 +54,7 @@ export type EnvironmentSubscriptionRpcTag = | typeof WS_METHODS.subscribeDiscoveredLocalServers | typeof WS_METHODS.subscribeResourceTelemetry | typeof WS_METHODS.pullRequestsSubscribeRefreshes + | typeof WS_METHODS.subscribeCodexGoal | typeof WS_METHODS.previewAutomationConnect | typeof WS_METHODS.subscribeVcsStatus | typeof WS_METHODS.terminalAttach; diff --git a/packages/client-runtime/src/state/threadCommands.test.ts b/packages/client-runtime/src/state/threadCommands.test.ts new file mode 100644 index 000000000000..8fafd9bd9833 --- /dev/null +++ b/packages/client-runtime/src/state/threadCommands.test.ts @@ -0,0 +1,148 @@ +import { ThreadId, type CodexGoal, type CodexGoalStreamEvent } from "@t3tools/contracts"; +import { describe, expect, it } from "@effect/vitest"; + +import { + applyCodexGoalStreamEvent, + formatCodexGoalDescription, + formatCodexGoalError, + formatCodexGoalStatus, + formatCodexGoalUsage, + parseCodexGoalCommand, + toCodexGoalSetInput, +} from "./threadCommands.ts"; + +const threadId = ThreadId.make("thread-1"); +const goal = (objective: string): CodexGoal => ({ + objective, + status: "active", + tokenBudget: 100_000, + tokensUsed: 12_000, + timeUsedSeconds: 90, + createdAt: 1_777_000_000, + updatedAt: 1_777_000_090, +}); + +describe("parseCodexGoalCommand", () => { + it("maps all supported Goal commands to native mutations", () => { + const cases = [ + ["/goal", { action: "status" }], + ["/goal status", { action: "status" }], + ["/goal create Ship it", { action: "set", objective: "Ship it", status: "active" }], + ["/goal Ship it", { action: "set", objective: "Ship it", status: "active" }], + ["/goal steer Narrow the patch", { action: "set", objective: "Narrow the patch" }], + ["/goal edit Narrow the patch", { action: "set", objective: "Narrow the patch" }], + ["/goal pause", { action: "set", status: "paused" }], + ["/goal resume", { action: "set", status: "active" }], + ["/goal clear", { action: "clear" }], + ["/goal reset", { action: "clear" }], + [ + "/goal edit", + { + action: "invalid", + message: "T3 does not open Codex's Goal editor. Use /goal steer .", + }, + ], + ["please create a goal", null], + ] as const; + for (const [command, expected] of cases) { + expect(parseCodexGoalCommand(command)).toEqual(expected); + } + }); +}); + +describe("toCodexGoalSetInput", () => { + it("adds the thread id without inventing omitted native fields", () => { + expect(toCodexGoalSetInput(threadId, { action: "set", objective: "Ship it" })).toEqual({ + threadId, + objective: "Ship it", + }); + }); +}); + +describe("applyCodexGoalStreamEvent", () => { + it("formats native usage consistently for clients", () => { + expect(formatCodexGoalDescription(goal("Ship it"))).toBe( + "Ship it - 12,000 tokens / 100,000, 90 seconds", + ); + }); + + it("formats native statuses as user-facing labels", () => { + const statuses = [ + "active", + "paused", + "budgetLimited", + "usageLimited", + "complete", + "blocked", + ] as const; + expect(statuses.map(formatCodexGoalStatus)).toEqual([ + "active", + "paused", + "budget limited", + "usage limited", + "complete", + "blocked", + ]); + }); + + it("applies native updated and cleared notifications", () => { + const updated = applyCodexGoalStreamEvent({ + type: "updated", + threadId, + goal: goal("Updated asynchronously"), + }); + expect(updated?.objective).toBe("Updated asynchronously"); + expect(applyCodexGoalStreamEvent({ type: "cleared", threadId })).toBeNull(); + }); + + it("accepts the authoritative snapshot after reconnect", () => { + const reconnectSnapshot: CodexGoalStreamEvent = { + type: "snapshot", + threadId, + goal: goal("Changed while disconnected"), + }; + expect(applyCodexGoalStreamEvent(reconnectSnapshot)?.objective).toBe( + "Changed while disconnected", + ); + }); +}); + +describe("formatCodexGoalError", () => { + it("appends the provider reason carried in the error cause", () => { + const error = new Error("Codex Goal set failed for thread thread-1", { + cause: new Error("Provider 'claude' is not implemented"), + }); + expect(formatCodexGoalError(error)).toBe( + "Codex Goal set failed for thread thread-1: Provider 'claude' is not implemented", + ); + }); + + it("falls back to the wrapper message when the cause carries no reason", () => { + expect(formatCodexGoalError(new Error("Codex Goal get failed for thread thread-1"))).toBe( + "Codex Goal get failed for thread thread-1", + ); + }); + + it("handles non-error failures", () => { + expect(formatCodexGoalError("boom")).toBe("Codex Goal operation failed."); + }); +}); + +describe("formatCodexGoalUsage", () => { + it("renders the budget when one is set", () => { + expect(formatCodexGoalUsage(goal("Ship it"))).toBe("12,000 tokens / 100,000, 90 seconds"); + }); + + it("omits the budget when there is none", () => { + expect(formatCodexGoalUsage({ ...goal("Ship it"), tokenBudget: null })).toBe( + "12,000 tokens, 90 seconds", + ); + }); + + it("is the usage half of the full description", () => { + const withBudget = goal("Ship it"); + expect(formatCodexGoalDescription(withBudget)).toBe( + `Ship it - ${formatCodexGoalUsage(withBudget)}`, + ); + }); +}); diff --git a/packages/client-runtime/src/state/threadCommands.ts b/packages/client-runtime/src/state/threadCommands.ts index e1762d8b76bd..931b6380f9f4 100644 --- a/packages/client-runtime/src/state/threadCommands.ts +++ b/packages/client-runtime/src/state/threadCommands.ts @@ -1,4 +1,11 @@ -import type { ThreadId } from "@t3tools/contracts"; +import * as Stream from "effect/Stream"; +import type { + CodexGoal, + CodexGoalSetInput, + CodexGoalStatus, + CodexGoalStreamEvent, + ThreadId, +} from "@t3tools/contracts"; import * as Crypto from "effect/Crypto"; import * as Effect from "effect/Effect"; import * as Option from "effect/Option"; @@ -9,6 +16,7 @@ import { createAtomCommandScheduler, createEnvironmentCommand, createEnvironmentRpcCommand, + createEnvironmentRpcSubscriptionAtomFamily, } from "./runtime.ts"; import { type ArchiveThreadInput, @@ -120,7 +128,32 @@ export function createThreadEnvironmentAtoms( key: ({ environmentId, input }: { environmentId: string; input: { threadId: string } }) => JSON.stringify([environmentId, input.threadId]), }; + const codexGoal = createEnvironmentRpcSubscriptionAtomFamily(runtime, { + label: "environment-data:codex-goal", + tag: WS_METHODS.subscribeCodexGoal, + idleTtlMs: 0, + transform: (events) => events.pipe(Stream.map(applyCodexGoalStreamEvent)), + }); return { + codexGoal, + getCodexGoal: createEnvironmentRpcCommand(runtime, { + label: "environment-data:codex-goal:get", + tag: WS_METHODS.codexGoalGet, + scheduler, + concurrency, + }), + setCodexGoal: createEnvironmentRpcCommand(runtime, { + label: "environment-data:codex-goal:set", + tag: WS_METHODS.codexGoalSet, + scheduler, + concurrency, + }), + clearCodexGoal: createEnvironmentRpcCommand(runtime, { + label: "environment-data:codex-goal:clear", + tag: WS_METHODS.codexGoalClear, + scheduler, + concurrency, + }), create: createEnvironmentCommand(runtime, { label: "environment-data:commands:thread:create", execute: (input: CreateThreadInput) => createThread(input), @@ -324,3 +357,90 @@ export function createThreadEnvironmentAtoms( }), }; } + +export type CodexGoalCommand = + | { readonly action: "status" } + | { readonly action: "set"; readonly objective?: string; readonly status?: "active" | "paused" } + | { readonly action: "clear" } + | { readonly action: "invalid"; readonly message: string }; + +const GOAL_USAGE = + "Usage: /goal [status | create | steer | pause | resume | clear | reset]"; + +export function formatCodexGoalUsage(goal: CodexGoal): string { + const budget = goal.tokenBudget == null ? "" : ` / ${goal.tokenBudget.toLocaleString()}`; + return `${goal.tokensUsed.toLocaleString()} tokens${budget}, ${goal.timeUsedSeconds.toLocaleString()} seconds`; +} + +export function formatCodexGoalDescription(goal: CodexGoal): string { + return `${goal.objective} - ${formatCodexGoalUsage(goal)}`; +} + +const CODEX_GOAL_STATUS_LABELS: Record = { + active: "active", + paused: "paused", + budgetLimited: "budget limited", + usageLimited: "usage limited", + complete: "complete", + blocked: "blocked", +}; + +export function formatCodexGoalStatus(status: CodexGoalStatus): string { + return CODEX_GOAL_STATUS_LABELS[status]; +} + +export function formatCodexGoalError(error: unknown): string { + if (!(error instanceof Error)) return "Codex Goal operation failed."; + const reason = error.cause instanceof Error ? error.cause.message.trim() : ""; + return reason.length === 0 ? error.message : `${error.message}: ${reason}`; +} + +export function parseCodexGoalCommand(value: string): CodexGoalCommand | null { + const match = /^\/goal(?:\s+([\s\S]*))?$/i.exec(value.trim()); + if (match === null) return null; + + const argument = match[1]?.trim() ?? ""; + if (argument === "" || argument.toLowerCase() === "status") return { action: "status" }; + + const [rawAction = "", ...rest] = argument.split(/\s+/); + const action = rawAction.toLowerCase(); + const objective = rest.join(" ").trim(); + if (action === "create" || action === "steer") { + if (objective === "") return { action: "invalid", message: GOAL_USAGE }; + return action === "create" + ? { action: "set", objective, status: "active" } + : { action: "set", objective }; + } + if (action === "edit") { + return objective === "" + ? { + action: "invalid", + message: "T3 does not open Codex's Goal editor. Use /goal steer .", + } + : { action: "set", objective }; + } + if (action === "pause" || action === "resume") { + if (objective !== "") return { action: "invalid", message: GOAL_USAGE }; + return { action: "set", status: action === "pause" ? "paused" : "active" }; + } + if (action === "clear" || action === "reset") { + if (objective !== "") return { action: "invalid", message: GOAL_USAGE }; + return { action: "clear" }; + } + if (action === "status") return { action: "invalid", message: GOAL_USAGE }; + + return { action: "set", objective: argument, status: "active" }; +} + +export function toCodexGoalSetInput( + threadId: CodexGoalSetInput["threadId"], + command: Extract, +): CodexGoalSetInput { + const { action: _action, ...input } = command; + return { threadId, ...input }; +} + +export function applyCodexGoalStreamEvent(event: CodexGoalStreamEvent): CodexGoal | null { + if (event.type === "snapshot" || event.type === "updated") return event.goal; + return null; +} diff --git a/packages/contracts/src/codexGoal.test.ts b/packages/contracts/src/codexGoal.test.ts new file mode 100644 index 000000000000..cf3f4ee66af0 --- /dev/null +++ b/packages/contracts/src/codexGoal.test.ts @@ -0,0 +1,20 @@ +import { describe, expect, it } from "vite-plus/test"; +import * as Schema from "effect/Schema"; + +import { CODEX_GOAL_OBJECTIVE_MAX_CHARS, CodexGoalSetInput } from "./codexGoal.ts"; + +const decodeSetInput = Schema.decodeUnknownSync(CodexGoalSetInput); + +describe("CodexGoalSetInput", () => { + it("matches Codex's positive token budget constraint", () => { + expect(decodeSetInput({ threadId: "thread-1", tokenBudget: 1 }).tokenBudget).toBe(1); + expect(decodeSetInput({ threadId: "thread-1", tokenBudget: null }).tokenBudget).toBeNull(); + expect(() => decodeSetInput({ threadId: "thread-1", tokenBudget: 0 })).toThrow(); + }); + + it("matches Codex's 4,000 Unicode-character objective limit", () => { + const maximum = "😀".repeat(CODEX_GOAL_OBJECTIVE_MAX_CHARS); + expect(decodeSetInput({ threadId: "thread-1", objective: maximum }).objective).toBe(maximum); + expect(() => decodeSetInput({ threadId: "thread-1", objective: `${maximum}x` })).toThrow(); + }); +}); diff --git a/packages/contracts/src/codexGoal.ts b/packages/contracts/src/codexGoal.ts new file mode 100644 index 000000000000..56da9bd77eca --- /dev/null +++ b/packages/contracts/src/codexGoal.ts @@ -0,0 +1,83 @@ +import * as Schema from "effect/Schema"; + +import { NonNegativeInt, PositiveInt, ThreadId, TrimmedNonEmptyString } from "./baseSchemas.ts"; +import { ProviderInstanceId } from "./providerInstance.ts"; + +export const CODEX_GOAL_OBJECTIVE_MAX_CHARS = 4_000; +const CodexGoalObjective = TrimmedNonEmptyString.check( + Schema.makeFilter( + (objective) => + Array.from(objective).length <= CODEX_GOAL_OBJECTIVE_MAX_CHARS || + `Goal objective must not exceed ${CODEX_GOAL_OBJECTIVE_MAX_CHARS} characters.`, + ), +); +export const CodexGoalStatus = Schema.Literals([ + "active", + "paused", + "blocked", + "usageLimited", + "budgetLimited", + "complete", +]); +export type CodexGoalStatus = typeof CodexGoalStatus.Type; +export const CodexGoal = Schema.Struct({ + objective: TrimmedNonEmptyString, + status: CodexGoalStatus, + tokenBudget: Schema.optionalKey(Schema.NullOr(NonNegativeInt)), + tokensUsed: NonNegativeInt, + timeUsedSeconds: NonNegativeInt, + createdAt: NonNegativeInt, + updatedAt: NonNegativeInt, +}); +export type CodexGoal = typeof CodexGoal.Type; +export const CodexGoalThreadInput = Schema.Struct({ + threadId: ThreadId, +}); +export type CodexGoalThreadInput = typeof CodexGoalThreadInput.Type; +export const CodexGoalSubscriptionInput = Schema.Struct({ + threadId: ThreadId, + providerInstanceId: ProviderInstanceId, +}); +export type CodexGoalSubscriptionInput = typeof CodexGoalSubscriptionInput.Type; +export const CodexGoalSetInput = Schema.Struct({ + threadId: ThreadId, + objective: Schema.optionalKey(CodexGoalObjective), + status: Schema.optionalKey(CodexGoalStatus), + tokenBudget: Schema.optionalKey(Schema.NullOr(PositiveInt)), +}); +export type CodexGoalSetInput = typeof CodexGoalSetInput.Type; +export const CodexGoalClearResult = Schema.Struct({ + cleared: Schema.Boolean, +}); +export type CodexGoalClearResult = typeof CodexGoalClearResult.Type; +export const CodexGoalStreamEvent = Schema.Union([ + Schema.Struct({ + type: Schema.Literal("snapshot"), + threadId: ThreadId, + goal: Schema.NullOr(CodexGoal), + }), + Schema.Struct({ + type: Schema.Literal("updated"), + threadId: ThreadId, + goal: CodexGoal, + }), + Schema.Struct({ + type: Schema.Literal("cleared"), + threadId: ThreadId, + }), +]); +export type CodexGoalStreamEvent = typeof CodexGoalStreamEvent.Type; +export const CodexGoalOperation = Schema.Literals(["get", "set", "clear", "subscribe"]); +export type CodexGoalOperation = typeof CodexGoalOperation.Type; +export class CodexGoalOperationError extends Schema.TaggedErrorClass()( + "CodexGoalOperationError", + { + operation: CodexGoalOperation, + threadId: ThreadId, + cause: Schema.Defect(), + }, +) { + override get message(): string { + return `Codex Goal ${this.operation} failed for thread ${this.threadId}`; + } +} diff --git a/packages/contracts/src/index.ts b/packages/contracts/src/index.ts index 092de0f48f26..e78bd0dc0c3c 100644 --- a/packages/contracts/src/index.ts +++ b/packages/contracts/src/index.ts @@ -51,3 +51,5 @@ export * from "./scheduledTask.ts"; export * from "./worktreeMcp.ts"; export * from "./resourceTelemetry.ts"; export * from "./rpc.ts"; + +export * from "./codexGoal.ts"; diff --git a/packages/contracts/src/rpc.ts b/packages/contracts/src/rpc.ts index 0ba697b1ff9d..5923d104c20d 100644 --- a/packages/contracts/src/rpc.ts +++ b/packages/contracts/src/rpc.ts @@ -1,3 +1,12 @@ +import { + CodexGoal, + CodexGoalClearResult, + CodexGoalOperationError, + CodexGoalSetInput, + CodexGoalSubscriptionInput, + CodexGoalStreamEvent, + CodexGoalThreadInput, +} from "./codexGoal.ts"; import * as Schema from "effect/Schema"; import * as Rpc from "effect/unstable/rpc/Rpc"; import * as RpcGroup from "effect/unstable/rpc/RpcGroup"; @@ -281,6 +290,11 @@ export const WS_METHODS = { providerInstallSubscribe: "provider.install.subscribe", providerInstallRemove: "provider.install.remove", + codexGoalGet: "codex.goal.get", + codexGoalSet: "codex.goal.set", + codexGoalClear: "codex.goal.clear", + subscribeCodexGoal: "codex.goal.subscribe", + // VCS methods vcsPull: "vcs.pull", vcsRefreshStatus: "vcs.refreshStatus", @@ -1287,6 +1301,31 @@ export const WsSubscribeResourceTelemetryRpc = Rpc.make(WS_METHODS.subscribeReso stream: true, }); +export const WsCodexGoalGetRpc = Rpc.make(WS_METHODS.codexGoalGet, { + payload: CodexGoalThreadInput, + success: Schema.NullOr(CodexGoal), + error: Schema.Union([CodexGoalOperationError, EnvironmentAuthorizationError]), +}); + +export const WsCodexGoalSetRpc = Rpc.make(WS_METHODS.codexGoalSet, { + payload: CodexGoalSetInput, + success: CodexGoal, + error: Schema.Union([CodexGoalOperationError, EnvironmentAuthorizationError]), +}); + +export const WsCodexGoalClearRpc = Rpc.make(WS_METHODS.codexGoalClear, { + payload: CodexGoalThreadInput, + success: CodexGoalClearResult, + error: Schema.Union([CodexGoalOperationError, EnvironmentAuthorizationError]), +}); + +export const WsSubscribeCodexGoalRpc = Rpc.make(WS_METHODS.subscribeCodexGoal, { + payload: CodexGoalSubscriptionInput, + success: CodexGoalStreamEvent, + error: Schema.Union([CodexGoalOperationError, EnvironmentAuthorizationError]), + stream: true, +}); + export const WsRpcGroup = RpcGroup.make( WsServerProbeRpc, WsServerGetConfigRpc, @@ -1400,6 +1439,10 @@ export const WsRpcGroup = RpcGroup.make( WsPreviewAutomationRespondRpc, WsPreviewAutomationFocusHostRpc, WsSubscribePreviewEventsRpc, + WsCodexGoalGetRpc, + WsCodexGoalSetRpc, + WsCodexGoalClearRpc, + WsSubscribeCodexGoalRpc, WsSubscribeDiscoveredLocalServersRpc, WsSubscribeServerConfigRpc, WsSubscribeServerLifecycleRpc, From 7964c7bed45c2c7013c7f07b1c01f24ec0dcc021 Mon Sep 17 00:00:00 2001 From: Javad Asgari Shafique Date: Wed, 9 Sep 2026 16:20:25 +0200 Subject: [PATCH 2/2] fix(codex): recover and buffer native goal continuations --- .../Adapters/CodexAdapterV2.test.ts | 458 +++++++++++++++++- .../Adapters/CodexAdapterV2.ts | 210 ++++++-- .../src/orchestration-v2/ProviderAdapter.ts | 4 + .../ProviderSessionManager.ts | 18 + apps/server/src/ws.ts | 39 +- docs/user/providers-codex.md | 5 +- .../src/state/threadCommands.ts | 7 +- 7 files changed, 693 insertions(+), 48 deletions(-) diff --git a/apps/server/src/orchestration-v2/Adapters/CodexAdapterV2.test.ts b/apps/server/src/orchestration-v2/Adapters/CodexAdapterV2.test.ts index caff6604f4dc..cdb053846779 100644 --- a/apps/server/src/orchestration-v2/Adapters/CodexAdapterV2.test.ts +++ b/apps/server/src/orchestration-v2/Adapters/CodexAdapterV2.test.ts @@ -1,3 +1,4 @@ +import * as Option from "effect/Option"; import * as NodeServices from "@effect/platform-node/NodeServices"; import { CheckpointId, @@ -1251,6 +1252,66 @@ function makeCodexReplayTranscript(input: { }; } +function withNoGoalInterruptLookup( + input: CodexReplay.CodexAppServerReplayTranscript, +): CodexReplay.CodexAppServerReplayTranscript { + // Older interruption fixtures predate native goals. Explicitly answer the new + // pre-interrupt lookup with no goal, preserving their command-cleanup scenarios. + const hasGoalProtocol = input.entries.some( + (entry) => + "frame" in entry && + typeof entry.frame === "object" && + entry.frame !== null && + "method" in entry.frame && + String(entry.frame.method).startsWith("thread/goal/"), + ); + const firstInterrupt = input.entries.findIndex( + (entry) => + entry.type === "expect_outbound" && + typeof entry.frame === "object" && + entry.frame !== null && + "method" in entry.frame && + entry.frame.method === "turn/interrupt", + ); + let entries = input.entries; + if (!hasGoalProtocol && firstInterrupt >= 0) { + const interrupt = input.entries[firstInterrupt]!; + if ( + interrupt.type === "expect_outbound" && + typeof interrupt.frame === "object" && + interrupt.frame !== null && + "id" in interrupt.frame && + typeof interrupt.frame.id === "number" && + "params" in interrupt.frame + ) { + const id = interrupt.frame.id; + const params = interrupt.frame.params as { threadId: string }; + entries = input.entries.flatMap((entry, index): CodexReplay.CodexAppServerReplayEntry[] => { + const shifted = + "frame" in entry && + typeof entry.frame === "object" && + entry.frame !== null && + "id" in entry.frame && + typeof entry.frame.id === "number" && + entry.frame.id >= id + ? { ...entry, frame: { ...entry.frame, id: entry.frame.id + 1 } } + : entry; + return index === firstInterrupt + ? [ + { + type: "expect_outbound", + frame: { id, method: "thread/goal/get", params: { threadId: params.threadId } }, + }, + { type: "emit_inbound", frame: { id, result: { goal: null } } }, + shifted, + ] + : [shifted]; + }); + } + } + return { ...input, entries }; +} + describe("CodexAdapterV2 post-settle continuation", () => { const awaitUntil = (predicate: () => boolean, label: string): Effect.Effect => Effect.gen(function* () { @@ -1264,12 +1325,13 @@ describe("CodexAdapterV2 post-settle continuation", () => { }); const makeCodexReplayHarness = ( - transcript: CodexReplay.CodexAppServerReplayTranscript, + sourceTranscript: CodexReplay.CodexAppServerReplayTranscript, onEvent: (event: ProviderAdapterV2Event) => Effect.Effect = () => Effect.void, onContinuation: (request: ProviderContinuationRequest) => Effect.Effect = () => Effect.void, ) => Effect.gen(function* () { + const transcript = withNoGoalInterruptLookup(sourceTranscript); const fileSystem = yield* FileSystem.FileSystem; const idAllocator = yield* IdAllocatorV2; const serverConfig = yield* makeReplayServerConfig(transcript.scenario).pipe(Effect.orDie); @@ -1485,6 +1547,400 @@ describe("CodexAdapterV2 post-settle continuation", () => { ).pipe(Effect.provide(Layer.mergeAll(NodeServices.layer, idAllocatorLayer))), ); + it.effect("keeps several fast native continuations without a goal notification", () => + Effect.scoped( + Effect.gen(function* () { + const nativeThreadId = "fast-goal-thread"; + const offered = yield* Deferred.make(); + const completions = yield* Effect.forEach([0, 1, 2], () => Deferred.make()); + let count = 0; + const entries: CodexReplay.CodexAppServerReplayEntry[] = [ + ...codexReplayPreamble({ nativeThreadId, nativeTurnId: "first", prompt: "Work" }).slice( + 0, + 5, + ), + { + type: "expect_outbound", + frame: { + id: 3, + method: "thread/resume", + params: { threadId: nativeThreadId, excludeTurns: true }, + }, + }, + { + type: "emit_inbound", + frame: { id: 3, result: { thread: { id: nativeThreadId, updatedAt: 1782622450 } } }, + }, + ]; + for (let index = 0; index < 3; index++) { + const id = `fast-${index}`; + entries.push( + { + type: "emit_inbound", + frame: { + method: "turn/started", + params: { + threadId: nativeThreadId, + turn: makeCodexReplayTurn({ id, status: "inProgress" }), + }, + }, + }, + { + type: "emit_inbound", + frame: { + method: "item/completed", + params: { + threadId: nativeThreadId, + turnId: id, + item: { + type: "agentMessage", + id: `result-${index}`, + text: `Result ${index}`, + phase: "final_answer", + }, + }, + }, + }, + { + type: "emit_inbound", + frame: { + method: "turn/completed", + params: { + threadId: nativeThreadId, + turn: makeCodexReplayTurn({ id, status: "completed" }), + }, + }, + }, + ); + } + entries.push( + { + type: "expect_outbound", + frame: { id: 4, method: "thread/goal/get", params: { threadId: nativeThreadId } }, + }, + { type: "emit_inbound", frame: { id: 4, result: { goal: null } } }, + ); + const harness = yield* makeCodexReplayHarness( + makeCodexReplayTranscript({ scenario: "fast-native-goals", entries }), + (event) => { + const index = + event.type === "turn.terminal" + ? [0, 1, 2].find((i) => String(event.providerTurnId).endsWith(`fast-${i}`)) + : undefined; + return index === undefined + ? Effect.void + : Deferred.succeed(completions[index]!, undefined); + }, + () => (++count === 3 ? Deferred.succeed(offered, undefined) : Effect.void), + ); + const first = makeCodexTestTurnInput({ + threadId: harness.threadId, + providerThread: harness.providerThread, + now: yield* DateTime.now, + attemptId: RunAttemptId.make("fast-first"), + text: "Work", + }); + yield* harness.runtime.resumeThread({ providerThread: harness.providerThread }); + yield* Deferred.await(offered); + assert.isNull(yield* harness.runtime.codexGoal!.get(harness.providerThread)); + for (let index = 0; index < 3; index++) { + yield* harness.runtime.startTurn({ + ...first, + attemptId: RunAttemptId.make(`fast-attempt-${index}`), + runId: RunId.make(`fast-run-${index}`), + message: { ...first.message, createdBy: "agent", creationSource: "provider" }, + }); + yield* Deferred.await(completions[index]!); + } + assert.lengthOf(harness.terminalEvents(), 3); + assert.deepEqual( + assistantMessages(harness.events).map((event) => event.message.text), + ["Result 0", "Result 1", "Result 2"], + ); + }), + ).pipe(Effect.provide(Layer.mergeAll(NodeServices.layer, idAllocatorLayer))), + ); + + it.effect("accepts a user message racing native adoption and retains buffered output", () => + Effect.scoped( + Effect.gen(function* () { + const nativeThreadId = "goal-user-race"; + const goal = { + threadId: nativeThreadId, + objective: "Finish", + status: "active", + tokensUsed: 0, + timeUsedSeconds: 0, + createdAt: 1782622440, + updatedAt: 1782622440, + }; + const offered = yield* Deferred.make(); + const userDone = yield* Deferred.make(); + const bufferedDone = yield* Deferred.make(); + const entries: CodexReplay.CodexAppServerReplayEntry[] = [ + ...codexReplayPreamble({ nativeThreadId, nativeTurnId: "first", prompt: "Work" }), + { + type: "emit_inbound", + frame: { + method: "turn/completed", + params: { + threadId: nativeThreadId, + turn: makeCodexReplayTurn({ id: "first", status: "completed" }), + }, + }, + }, + { + type: "emit_inbound", + frame: { + method: "turn/started", + params: { + threadId: nativeThreadId, + turn: makeCodexReplayTurn({ id: "automatic", status: "inProgress" }), + }, + }, + }, + { + type: "emit_inbound", + frame: { + method: "item/completed", + params: { + threadId: nativeThreadId, + turnId: "automatic", + item: { + type: "agentMessage", + id: "buffered-result", + text: "Buffered work", + phase: "final_answer", + }, + }, + }, + }, + { + type: "expect_outbound", + frame: { id: 4, method: "thread/goal/get", params: { threadId: nativeThreadId } }, + }, + { type: "emit_inbound", frame: { id: 4, result: { goal } } }, + { + type: "expect_outbound", + frame: { + id: 5, + method: "thread/goal/set", + params: { threadId: nativeThreadId, status: "paused" }, + }, + }, + { + type: "emit_inbound", + frame: { id: 5, result: { goal: { ...goal, status: "paused" } } }, + }, + { + type: "expect_outbound", + frame: { + id: 6, + method: "turn/interrupt", + params: { threadId: nativeThreadId, turnId: "automatic" }, + }, + }, + { type: "emit_inbound", frame: { id: 6, result: {} } }, + { + type: "emit_inbound", + frame: { + method: "turn/completed", + params: { + threadId: nativeThreadId, + turn: makeCodexReplayTurn({ id: "automatic", status: "interrupted" }), + }, + }, + }, + { + type: "expect_outbound", + frame: { + id: 7, + method: "turn/start", + params: { + threadId: nativeThreadId, + input: [{ type: "text", text: "Use the exact baseline" }], + cwd: "/workspace", + model: "gpt-5.4", + approvalPolicy: "never", + approvalsReviewer: "user", + sandboxPolicy: { type: "dangerFullAccess" }, + }, + }, + }, + { + type: "emit_inbound", + frame: { + id: 7, + result: { turn: makeCodexReplayTurn({ id: "user", status: "inProgress" }) }, + }, + }, + { + type: "expect_outbound", + frame: { + id: 8, + method: "thread/goal/set", + params: { threadId: nativeThreadId, status: "active" }, + }, + }, + { type: "emit_inbound", frame: { id: 8, result: { goal } } }, + { + type: "emit_inbound", + frame: { + method: "turn/completed", + params: { + threadId: nativeThreadId, + turn: makeCodexReplayTurn({ id: "user", status: "completed" }), + }, + }, + }, + ]; + const harness = yield* makeCodexReplayHarness( + makeCodexReplayTranscript({ scenario: "user-races-goal", entries }), + (event) => + event.type !== "turn.terminal" + ? Effect.void + : String(event.providerTurnId).endsWith(":user") + ? Deferred.succeed(userDone, undefined) + : String(event.providerTurnId).endsWith(":automatic") + ? Deferred.succeed(bufferedDone, undefined) + : Effect.void, + () => Deferred.succeed(offered, undefined), + ); + const first = makeCodexTestTurnInput({ + threadId: harness.threadId, + providerThread: harness.providerThread, + now: yield* DateTime.now, + attemptId: RunAttemptId.make("race-first"), + text: "Work", + }); + yield* harness.runtime.startTurn(first); + yield* harness.firstTerminal; + yield* Deferred.await(offered); + yield* harness.runtime.startTurn({ + ...first, + attemptId: RunAttemptId.make("race-user"), + runId: RunId.make("race-user"), + message: { ...first.message, text: "Use the exact baseline" }, + }); + yield* Deferred.await(userDone); + yield* harness.runtime.startTurn({ + ...first, + attemptId: RunAttemptId.make("race-wake"), + runId: RunId.make("race-wake"), + message: { ...first.message, createdBy: "agent", creationSource: "provider" }, + }); + yield* Deferred.await(bufferedDone); + assert.lengthOf(harness.terminalEvents(), 3); + assert.isTrue( + assistantMessages(harness.events).some((event) => event.message.text === "Buffered work"), + ); + }), + ).pipe(Effect.provide(Layer.mergeAll(NodeServices.layer, idAllocatorLayer))), + ); + + for (const viaArchive of [false, true]) { + it.effect(`discards a native continuation via ${viaArchive ? "archive" : "stale offer"}`, () => + Effect.scoped( + Effect.gen(function* () { + const nativeThreadId = "cancel-pending-goal"; + const goal = { + threadId: nativeThreadId, + objective: "Finish", + status: "paused", + tokensUsed: 0, + timeUsedSeconds: 0, + createdAt: 1782622440, + updatedAt: 1782622440, + }; + const offered = yield* Deferred.make(); + const offset = viaArchive ? 1 : 0; + const transcript = makeCodexReplayTranscript({ + scenario: "cancel-pending-goal", + entries: [ + ...codexReplayPreamble({ nativeThreadId, nativeTurnId: "first", prompt: "Work" }), + { + type: "emit_inbound", + frame: { + method: "turn/completed", + params: { + threadId: nativeThreadId, + turn: makeCodexReplayTurn({ id: "first", status: "completed" }), + }, + }, + }, + { + type: "emit_inbound", + frame: { + method: "turn/started", + params: { + threadId: nativeThreadId, + turn: makeCodexReplayTurn({ id: "pending", status: "inProgress" }), + }, + }, + }, + ...(viaArchive + ? [ + { + type: "expect_outbound" as const, + frame: { + id: 4, + method: "thread/goal/get", + params: { threadId: nativeThreadId }, + }, + }, + { + type: "emit_inbound" as const, + frame: { id: 4, result: { goal: { ...goal, status: "active" } } }, + }, + ] + : []), + { + type: "expect_outbound", + frame: { + id: 4 + offset, + method: "thread/goal/set", + params: { threadId: nativeThreadId, status: "paused" }, + }, + }, + { type: "emit_inbound", frame: { id: 4 + offset, result: { goal } } }, + { + type: "expect_outbound", + frame: { + id: 5 + offset, + method: "turn/interrupt", + params: { threadId: nativeThreadId, turnId: "pending" }, + }, + }, + { type: "emit_inbound", frame: { id: 5 + offset, result: {} } }, + ], + }); + const harness = yield* makeCodexReplayHarness( + transcript, + () => Effect.void, + () => Deferred.succeed(offered, undefined), + ); + yield* harness.runtime.startTurn( + makeCodexTestTurnInput({ + threadId: harness.threadId, + providerThread: harness.providerThread, + now: yield* DateTime.now, + attemptId: RunAttemptId.make("cancel-first"), + text: "Work", + }), + ); + yield* Deferred.await(offered); + const wake = harness.continuationRequests[0]!; + if (viaArchive) yield* harness.runtime.discardPendingTurns!(harness.providerThread); + else yield* wake.clearIfCurrent!(); + assert.isTrue( + Option.isNone(yield* wake.dispatchIfCurrent!(Effect.succeed("must not dispatch"))), + ); + assert.isFalse(yield* harness.hasPendingBackgroundWork); + }), + ).pipe(Effect.provide(Layer.mergeAll(NodeServices.layer, idAllocatorLayer))), + ); + } + for (const pauseFails of [false, true]) { it.effect( `interrupts a goal turn after ${pauseFails ? "a failed pause" : "pausing the goal"}`, diff --git a/apps/server/src/orchestration-v2/Adapters/CodexAdapterV2.ts b/apps/server/src/orchestration-v2/Adapters/CodexAdapterV2.ts index ba1fe9387ca5..37cd5c78b72a 100644 --- a/apps/server/src/orchestration-v2/Adapters/CodexAdapterV2.ts +++ b/apps/server/src/orchestration-v2/Adapters/CodexAdapterV2.ts @@ -1,3 +1,5 @@ +import type { CodexAppServerError } from "effect-codex-app-server/errors"; +import { makeKeyedSerialExecutor } from "../KeyedSerialExecutor.ts"; import { CodexGoal, type CodexGoalStreamEvent } from "@t3tools/contracts"; import * as PubSub from "effect/PubSub"; import { @@ -1206,6 +1208,16 @@ export function codexThreadRuntimeParams(input: { }; } +function codexNotificationTurnId(payload: unknown): string | undefined { + if (typeof payload !== "object" || payload === null) return undefined; + const direct: unknown = Reflect.get(payload, "turnId"); + if (typeof direct === "string") return direct; + const turn: unknown = Reflect.get(payload, "turn"); + if (typeof turn !== "object" || turn === null) return undefined; + const id: unknown = Reflect.get(turn, "id"); + return typeof id === "string" ? id : undefined; +} + const decodeCodexGoal = Schema.decodeUnknownEffect(CodexGoal); const decodeCodexResumeMetadata = Schema.decodeUnknownEffect( @@ -1520,7 +1532,18 @@ export function makeCodexAdapterV2(adapterOptions: CodexAdapterV2Options): Provi planSelectionTransition: () => Effect.succeed(turnScopedSelectionTransition()), openSession: (input) => Effect.gen(function* () { - const client = yield* clientFactory.open({ + const nativeGoalTurns = new Map< + string, + { + readonly nativeThreadId: string; + readonly notifications: Array>; + adopted: boolean; + readonly nativeTurnId: string; + readonly startedAt: DateTime.Utc; + readonly ready: Deferred.Deferred; + } + >(); + const nativeClient = yield* clientFactory.open({ instanceId: adapterOptions.instanceId, threadId: input.threadId, providerSessionId: input.providerSessionId, @@ -1528,6 +1551,22 @@ export function makeCodexAdapterV2(adapterOptions: CodexAdapterV2Options): Provi settings: adapterOptions.settings, environment: adapterOptions.environment, }); + const client: typeof nativeClient = { + ...nativeClient, + handleServerNotification: (method, handler) => + nativeClient.handleServerNotification(method, (payload) => + Effect.suspend(() => { + const turnId = codexNotificationTurnId(payload); + const pending = + typeof turnId === "string" ? nativeGoalTurns.get(turnId) : undefined; + if (pending !== undefined && method !== "turn/started") { + pending.notifications.push(handler(payload)); + return Effect.void; + } + return handler(payload); + }), + ), + }; const initialized = yield* Ref.make(false); const ensureInitialized = Effect.gen(function* () { const alreadyInitialized = yield* Ref.get(initialized); @@ -1556,8 +1595,11 @@ export function makeCodexAdapterV2(adapterOptions: CodexAdapterV2Options): Provi readonly goal: CodexGoal | null; }>(); const activeGoalThreads = new Set(); + const goalLocks = yield* makeKeyedSerialExecutor(); + const goalRevisions = new Map(); const goalChanged = (nativeThreadId: string, goal: CodexGoal | null) => Effect.gen(function* () { + goalRevisions.set(nativeThreadId, (goalRevisions.get(nativeThreadId) ?? 0) + 1); if (goal?.status === "active") activeGoalThreads.add(nativeThreadId); else activeGoalThreads.delete(nativeThreadId); yield* PubSub.publish(goalEvents, { nativeThreadId, goal }); @@ -1565,6 +1607,7 @@ export function makeCodexAdapterV2(adapterOptions: CodexAdapterV2Options): Provi const getGoal = (providerThread: OrchestrationV2ProviderThread) => Effect.gen(function* () { const threadId = yield* getNativeThreadId(providerThread); + const revision = goalRevisions.get(threadId) ?? 0; const response = yield* ensureInitialized.pipe( Effect.andThen(client.request("thread/goal/get", { threadId })), ); @@ -1572,7 +1615,7 @@ export function makeCodexAdapterV2(adapterOptions: CodexAdapterV2Options): Provi response.goal === null || response.goal === undefined ? null : yield* decodeCodexGoal(response.goal); - yield* goalChanged(threadId, goal); + if ((goalRevisions.get(threadId) ?? 0) === revision) yield* goalChanged(threadId, goal); return goal; }).pipe(Effect.mapError((cause) => toProtocolError("Failed to read Codex goal.", cause))); yield* client.handleServerNotification("thread/goal/updated", (payload) => @@ -1584,14 +1627,9 @@ export function makeCodexAdapterV2(adapterOptions: CodexAdapterV2Options): Provi yield* client.handleServerNotification("thread/goal/cleared", (payload) => goalChanged(payload.threadId, null), ); - const rootInputsByNativeThread = new Map(); - const nativeGoalTurns = new Map< + const rootInputsByNativeThread = new Map< string, - { - readonly nativeTurnId: string; - readonly startedAt: DateTime.Utc; - readonly ready: Deferred.Deferred; - } + { threadId: ThreadId; providerThread: OrchestrationV2ProviderThread } >(); const activeTurns = yield* Ref.make(new Map()); const turnTokenUsageByThread = new Map(); @@ -3609,23 +3647,33 @@ export function makeCodexAdapterV2(adapterOptions: CodexAdapterV2Options): Provi return; } const rootInput = rootInputsByNativeThread.get(payload.threadId); - if ( - rootInput !== undefined && - activeGoalThreads.has(payload.threadId) && - continuationRequests !== undefined - ) { - const previous = nativeGoalTurns.get(payload.threadId); - if (previous?.nativeTurnId === payload.turn.id) return; - if (previous !== undefined) yield* Deferred.await(previous.ready); + if (rootInput !== undefined && continuationRequests !== undefined) { + if (nativeGoalTurns.has(payload.turn.id)) return; const pending = { + notifications: [] as Array>, + adopted: false, + nativeThreadId: payload.threadId, nativeTurnId: payload.turn.id, startedAt: codexTimestamp(payload.turn.startedAt), ready: yield* Deferred.make(), }; - nativeGoalTurns.set(payload.threadId, pending); + nativeGoalTurns.set(payload.turn.id, pending); const clear = Effect.gen(function* () { - if (nativeGoalTurns.get(payload.threadId) !== pending) return; - nativeGoalTurns.delete(payload.threadId); + if (nativeGoalTurns.get(payload.turn.id) !== pending) return; + yield* Effect.gen(function* () { + const response = yield* client.request("thread/goal/set", { + threadId: payload.threadId, + status: "paused", + }); + yield* goalChanged(payload.threadId, yield* decodeCodexGoal(response.goal)); + }).pipe(Effect.timeout("1 second"), Effect.ignore); + yield* client + .request("turn/interrupt", { + threadId: payload.threadId, + turnId: pending.nativeTurnId, + }) + .pipe(Effect.timeout("1 second"), Effect.ignore); + nativeGoalTurns.delete(payload.turn.id); yield* Deferred.succeed(pending.ready, undefined); }); yield* continuationRequests.offer({ @@ -3636,7 +3684,7 @@ export function makeCodexAdapterV2(adapterOptions: CodexAdapterV2Options): Provi delivery: "adapter_buffered", clearIfCurrent: () => clear, dispatchIfCurrent: (effect) => - nativeGoalTurns.get(payload.threadId) === pending + nativeGoalTurns.get(payload.turn.id) === pending ? effect.pipe(Effect.map(Option.some)) : Effect.succeed(Option.none()), }); @@ -4955,8 +5003,35 @@ export function makeCodexAdapterV2(adapterOptions: CodexAdapterV2Options): Provi } return false; }), + discardPendingTurns: (providerThread) => + Effect.gen(function* () { + const threadId = yield* getNativeThreadId(providerThread); + yield* goalLocks + .withLock( + providerThread.id, + Effect.gen(function* () { + const goal = yield* getGoal(providerThread); + if (goal?.status !== "active") return; + const response = yield* client.request("thread/goal/set", { + threadId, + status: "paused", + }); + yield* goalChanged(threadId, yield* decodeCodexGoal(response.goal)); + }), + ) + .pipe(Effect.timeout("1 second"), Effect.ignore); + for (const pending of nativeGoalTurns.values()) { + if (pending.nativeThreadId !== threadId || pending.adopted) continue; + yield* client + .request("turn/interrupt", { threadId, turnId: pending.nativeTurnId }) + .pipe(Effect.timeout("1 second"), Effect.ignore); + nativeGoalTurns.delete(pending.nativeTurnId); + yield* Deferred.succeed(pending.ready, undefined); + } + rootInputsByNativeThread.delete(threadId); + }), codexGoal: { - get: getGoal, + get: (thread) => goalLocks.withLock(thread.id, getGoal(thread)), set: (providerThread, goalInput) => Effect.gen(function* () { const threadId = yield* getNativeThreadId(providerThread); @@ -4967,6 +5042,7 @@ export function makeCodexAdapterV2(adapterOptions: CodexAdapterV2Options): Provi yield* goalChanged(threadId, goal); return goal; }).pipe( + (effect) => goalLocks.withLock(providerThread.id, effect), Effect.mapError((cause) => toProtocolError("Failed to update Codex goal.", cause)), ), clear: (providerThread) => @@ -4978,6 +5054,7 @@ export function makeCodexAdapterV2(adapterOptions: CodexAdapterV2Options): Provi yield* goalChanged(threadId, null); return response; }).pipe( + (effect) => goalLocks.withLock(providerThread.id, effect), Effect.mapError((cause) => toProtocolError("Failed to clear Codex goal.", cause)), ), subscribe: (providerThread) => @@ -5039,6 +5116,12 @@ export function makeCodexAdapterV2(adapterOptions: CodexAdapterV2Options): Provi resumeThread: (threadInput) => Effect.gen(function* () { const nativeThreadId = yield* getNativeThreadId(threadInput.providerThread); + const appThreadId = threadInput.threadId ?? threadInput.providerThread.appThreadId; + if (appThreadId !== null && appThreadId !== undefined) + rootInputsByNativeThread.set(nativeThreadId, { + threadId: appThreadId, + providerThread: threadInput.providerThread, + }); const response = yield* ensureInitialized.pipe( Effect.andThen( @@ -5113,25 +5196,45 @@ export function makeCodexAdapterV2(adapterOptions: CodexAdapterV2Options): Provi startTurn: (turnInput) => Effect.gen(function* () { const threadId = yield* getNativeThreadId(turnInput.providerThread); - const goalTurn = nativeGoalTurns.get(threadId); - if (goalTurn !== undefined) { - if ( - turnInput.message.createdBy !== "agent" || - turnInput.message.creationSource !== "provider" - ) { - return yield* toProtocolError( - "A native goal continuation is being attached. Retry this message when it is running.", - ); - } + const goalTurn = Array.from(nativeGoalTurns.values()).find( + (turn) => turn.nativeThreadId === threadId && !turn.adopted, + ); + const isBufferedWake = + turnInput.message.createdBy === "agent" && + turnInput.message.creationSource === "provider"; + let resumeGoal = false; + if (goalTurn !== undefined && isBufferedWake) { yield* registerRootTurn({ turnInput, nativeTurnId: goalTurn.nativeTurnId, startedAt: goalTurn.startedAt, }); - nativeGoalTurns.delete(threadId); + goalTurn.adopted = true; yield* Deferred.succeed(goalTurn.ready, undefined); + while (goalTurn.notifications.length > 0) yield* goalTurn.notifications.shift()!; + nativeGoalTurns.delete(goalTurn.nativeTurnId); return; } + if (goalTurn !== undefined) { + // A user run can win the queue race against a native wake. Keep + // its buffered output for that wake, but give this message its + // own native turn instead of failing or swallowing the input. + const goal = yield* getGoal(turnInput.providerThread); + resumeGoal = goal?.status === "active"; + if (resumeGoal) { + const response = yield* client.request("thread/goal/set", { + threadId, + status: "paused", + }); + yield* goalChanged(threadId, yield* decodeCodexGoal(response.goal)); + } + for (const pending of nativeGoalTurns.values()) { + if (pending.nativeThreadId !== threadId) continue; + yield* client + .request("turn/interrupt", { threadId, turnId: pending.nativeTurnId }) + .pipe(Effect.timeout("1 second"), Effect.ignore); + } + } const codexInput = turnInput.restartContinuationOfRunId === undefined @@ -5152,6 +5255,13 @@ export function makeCodexAdapterV2(adapterOptions: CodexAdapterV2Options): Provi return updated; }); const started = yield* client.request("turn/start", turnStartParams); + if (resumeGoal) { + const response = yield* client.request("thread/goal/set", { + threadId, + status: "active", + }); + yield* goalChanged(threadId, yield* decodeCodexGoal(response.goal)); + } const nativeTurnId = started.turn.id; const startedAt = codexTimestamp(started.turn.startedAt); yield* registerRootTurn({ turnInput, nativeTurnId, startedAt }); @@ -5212,16 +5322,6 @@ export function makeCodexAdapterV2(adapterOptions: CodexAdapterV2Options): Provi ), interruptTurn: (turnInput) => Effect.gen(function* () { - yield* Effect.gen(function* () { - const threadId = yield* getNativeThreadId(turnInput.providerThread); - if (!activeGoalThreads.has(threadId)) return; - const goal = yield* getGoal(turnInput.providerThread); - if (goal?.status === "active") { - yield* client.request("thread/goal/set", { threadId, status: "paused" }); - yield* goalChanged(threadId, { ...goal, status: "paused" }); - } - }).pipe(Effect.timeout("1 second"), Effect.ignore); - const activeTurnContexts = Array.from((yield* Ref.get(activeTurns)).values()); const activeTurn = activeTurnContexts.find( (candidate) => candidate.providerTurnId === turnInput.providerTurnId, @@ -5231,6 +5331,30 @@ export function makeCodexAdapterV2(adapterOptions: CodexAdapterV2Options): Provi `Provider turn ${turnInput.providerTurnId} is not active and cannot be interrupted.`, ); } + yield* Effect.gen(function* () { + const threadId = yield* getNativeThreadId(turnInput.providerThread); + const goal = yield* getGoal(turnInput.providerThread); + if (goal?.status === "active") { + yield* client.request("thread/goal/set", { threadId, status: "paused" }); + yield* goalChanged(threadId, { ...goal, status: "paused" }); + } + }).pipe( + (effect) => goalLocks.withLock(turnInput.providerThread.id, effect), + Effect.timeout("1 second"), + Effect.ignore, + ); + + const nativeThreadId = yield* getNativeThreadId(turnInput.providerThread); + for (const pending of nativeGoalTurns.values()) { + if (pending.nativeThreadId !== nativeThreadId || pending.adopted) continue; + yield* client + .request("turn/interrupt", { + threadId: nativeThreadId, + turnId: pending.nativeTurnId, + }) + .pipe(Effect.timeout("1 second"), Effect.ignore); + } + const interruptTargetContexts = [ activeTurn, ...activeTurnContexts.filter( diff --git a/apps/server/src/orchestration-v2/ProviderAdapter.ts b/apps/server/src/orchestration-v2/ProviderAdapter.ts index 1bbc44290633..faf96fa6b9c8 100644 --- a/apps/server/src/orchestration-v2/ProviderAdapter.ts +++ b/apps/server/src/orchestration-v2/ProviderAdapter.ts @@ -496,6 +496,10 @@ export interface ProviderAdapterV2SessionRuntime { readonly hasPendingBackgroundWorkForThread?: ( providerThread: OrchestrationV2ProviderThread, ) => Effect.Effect; + /** Discard native work that has not yet been attached to an app run on archive/delete. */ + readonly discardPendingTurns?: ( + providerThread: OrchestrationV2ProviderThread, + ) => Effect.Effect; readonly codexGoal?: { readonly get: ( providerThread: OrchestrationV2ProviderThread, diff --git a/apps/server/src/orchestration-v2/ProviderSessionManager.ts b/apps/server/src/orchestration-v2/ProviderSessionManager.ts index 18831f54cd39..844257952284 100644 --- a/apps/server/src/orchestration-v2/ProviderSessionManager.ts +++ b/apps/server/src/orchestration-v2/ProviderSessionManager.ts @@ -1623,6 +1623,24 @@ export const layerWithOptions = ( ); } } + if (input.revokeMcpCredential === true) { + const live = (yield* Ref.get(sessions)).get(key); + const discard = live?.exposedRuntime.discardPendingTurns; + if (discard !== undefined) { + yield* projectionStore.getThreadProjection(input.threadId).pipe( + Effect.flatMap((projection) => + Effect.forEach( + projection.providerThreads.filter( + (thread) => thread.providerSessionId === input.providerSessionId, + ), + (thread) => discard(thread).pipe(Effect.ignore), + { discard: true }, + ), + ), + Effect.ignore, + ); + } + } const detached = yield* Ref.modify(sessions, (current) => { const entry = current.get(key); if (entry === undefined || !entry.attachedThreadIds.has(input.threadId)) { diff --git a/apps/server/src/ws.ts b/apps/server/src/ws.ts index 4f89855336df..b650b5e67ff7 100644 --- a/apps/server/src/ws.ts +++ b/apps/server/src/ws.ts @@ -1,3 +1,4 @@ +import * as RuntimePolicyV2 from "./orchestration-v2/RuntimePolicy.ts"; import { CodexGoalOperationError, type CodexGoalOperation, @@ -577,9 +578,45 @@ const makeWsRpcLayer = ( cause: "Start a Codex thread before managing its goal.", }); } - const runtime = Option.getOrNull( + if (projection.thread.archivedAt !== null || projection.thread.deletedAt !== null) { + return yield* new CodexGoalOperationError({ + threadId, + operation, + cause: "Restore the thread before managing its goal.", + }); + } + let runtime = Option.getOrNull( yield* providerSessionsV2.get(providerThread.providerSessionId), ); + if (runtime === null && (operation === "set" || operation === "clear")) { + const modelSelection = projection.thread.modelSelection; + if (modelSelection.instanceId !== providerThread.providerInstanceId) { + return yield* new CodexGoalOperationError({ + threadId, + operation, + cause: "Send a message with the selected provider before managing its goal.", + }); + } + const runtimePolicy = yield* Effect.flatMap(RuntimePolicyV2.RuntimePolicyV2, (policy) => + policy.resolve({ thread: projection.thread, modelSelection }), + ).pipe(Effect.provide(RuntimePolicyV2.layerFromProjectRepository)); + const resumeFromSession = projection.providerSessions.find( + (session) => session.id === providerThread.providerSessionId, + ); + runtime = yield* providerSessionsV2.open({ + threadId, + providerSessionId: providerThread.providerSessionId, + modelSelection, + runtimePolicy, + ...(resumeFromSession === undefined ? {} : { resumeFromSession }), + }); + yield* runtime.resumeThread({ + threadId, + providerThread, + modelSelection, + runtimePolicy, + }); + } if (runtime?.codexGoal === undefined) { return yield* new CodexGoalOperationError({ threadId, diff --git a/docs/user/providers-codex.md b/docs/user/providers-codex.md index d28116a786c3..4d7541a47ae8 100644 --- a/docs/user/providers-codex.md +++ b/docs/user/providers-codex.md @@ -82,5 +82,6 @@ removes the goal; `/goal reset` is an alias for clearing it. Send these commands without attachments. The goal is stored by Codex and shared across clients. Stop attempts to pause an active goal before interrupting its current turn. -Goal controls require a running provider session; after a server or provider -restart, send a normal message to reconnect the thread first. +After a server or provider restart, `/goal resume` reconnects the thread and +resumes its saved goal. Status subscriptions do not wake a stopped session merely +by opening a thread. Archiving or deleting a thread pauses its goal. diff --git a/packages/client-runtime/src/state/threadCommands.ts b/packages/client-runtime/src/state/threadCommands.ts index 931b6380f9f4..f5b223b8e558 100644 --- a/packages/client-runtime/src/state/threadCommands.ts +++ b/packages/client-runtime/src/state/threadCommands.ts @@ -1,3 +1,4 @@ +import * as Schedule from "effect/Schedule"; import * as Stream from "effect/Stream"; import type { CodexGoal, @@ -132,7 +133,11 @@ export function createThreadEnvironmentAtoms( label: "environment-data:codex-goal", tag: WS_METHODS.subscribeCodexGoal, idleTtlMs: 0, - transform: (events) => events.pipe(Stream.map(applyCodexGoalStreamEvent)), + transform: (events) => + events.pipe( + Stream.retry(Schedule.spaced("2 seconds")), + Stream.map(applyCodexGoalStreamEvent), + ), }); return { codexGoal,