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..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,10 +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); @@ -1300,7 +1364,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 +1429,636 @@ 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))), + ); + + 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"}`, + () => + 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..37cd5c78b72a 100644 --- a/apps/server/src/orchestration-v2/Adapters/CodexAdapterV2.ts +++ b/apps/server/src/orchestration-v2/Adapters/CodexAdapterV2.ts @@ -1,3 +1,7 @@ +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 { mcpToolPresentation, type McpToolPresentation, @@ -1204,6 +1208,18 @@ 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( Schema.Struct({ thread: Schema.Struct({ id: Schema.String, updatedAt: Schema.Number }) }), ); @@ -1516,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, @@ -1524,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); @@ -1547,6 +1590,47 @@ 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 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 }); + }); + 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 })), + ); + const goal = + response.goal === null || response.goal === undefined + ? null + : yield* decodeCodexGoal(response.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) => + 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< + string, + { threadId: ThreadId; providerThread: OrchestrationV2ProviderThread } + >(); const activeTurns = yield* Ref.make(new Map()); const turnTokenUsageByThread = new Map(); const usageStateForThread = (nativeThreadId: string) => { @@ -1664,6 +1748,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 +1817,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 +3488,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 +3646,50 @@ export function makeCodexAdapterV2(adapterOptions: CodexAdapterV2Options): Provi }); return; } + const rootInput = rootInputsByNativeThread.get(payload.threadId); + 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.turn.id, pending); + const clear = Effect.gen(function* () { + 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({ + 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.turn.id) === pending + ? effect.pipe(Effect.map(Option.some)) + : Effect.succeed(Option.none()), + }); + return; + } yield* rememberSubagentTurnStarted({ nativeThreadId: payload.threadId, nativeTurnId: payload.turn.id, @@ -4818,7 +4952,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 +4985,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 +5003,85 @@ 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: (thread) => goalLocks.withLock(thread.id, getGoal(thread)), + 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) => goalLocks.withLock(providerThread.id, effect), + 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) => goalLocks.withLock(providerThread.id, effect), + 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( @@ -4902,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( @@ -4976,6 +5196,45 @@ export function makeCodexAdapterV2(adapterOptions: CodexAdapterV2Options): Provi startTurn: (turnInput) => Effect.gen(function* () { const threadId = yield* getNativeThreadId(turnInput.providerThread); + 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, + }); + 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 @@ -4996,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 }); @@ -5065,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 36fc02078c8e..faf96fa6b9c8 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,25 @@ 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, + ) => 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/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/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..b650b5e67ff7 100644 --- a/apps/server/src/ws.ts +++ b/apps/server/src/ws.ts @@ -1,3 +1,9 @@ +import * as RuntimePolicyV2 from "./orchestration-v2/RuntimePolicy.ts"; +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 +562,73 @@ 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.", + }); + } + 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, + 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 +1703,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..4d7541a47ae8 100644 --- a/docs/user/providers-codex.md +++ b/docs/user/providers-codex.md @@ -69,3 +69,19 @@ 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. +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/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..f5b223b8e558 100644 --- a/packages/client-runtime/src/state/threadCommands.ts +++ b/packages/client-runtime/src/state/threadCommands.ts @@ -1,4 +1,12 @@ -import type { ThreadId } from "@t3tools/contracts"; +import * as Schedule from "effect/Schedule"; +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 +17,7 @@ import { createAtomCommandScheduler, createEnvironmentCommand, createEnvironmentRpcCommand, + createEnvironmentRpcSubscriptionAtomFamily, } from "./runtime.ts"; import { type ArchiveThreadInput, @@ -120,7 +129,36 @@ 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.retry(Schedule.spaced("2 seconds")), + 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 +362,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,