From 29cdd53d5d5f540dbe9f3bf78de6034709e739d4 Mon Sep 17 00:00:00 2001 From: Fredrik Ekman Date: Sun, 23 Aug 2026 01:25:06 +0200 Subject: [PATCH 1/7] feat(codex): add native Goal lifecycle controls --- .../src/features/threads/ThreadComposer.tsx | 4 +- .../features/threads/ThreadDetailScreen.tsx | 128 +++++++++- apps/mobile/src/state/threads.ts | 17 +- ...ProviderSessionStartup.integration.test.ts | 3 + apps/server/src/auth/RpcAuthorization.ts | 4 + .../Layers/CheckpointReactor.test.ts | 3 + .../Layers/ProviderCommandReactor.test.ts | 3 + .../Layers/ProviderRuntimeIngestion.test.ts | 3 + .../src/provider/Layers/CodexAdapter.test.ts | 154 ++++++++++++ .../src/provider/Layers/CodexAdapter.ts | 76 ++++++ .../src/provider/Layers/CodexProvider.ts | 13 + .../provider/Layers/CodexSessionRuntime.ts | 34 +++ .../provider/Layers/ProviderService.test.ts | 131 +++++++++- .../src/provider/Layers/ProviderService.ts | 60 +++++ .../Layers/ProviderSessionReaper.test.ts | 3 + .../src/provider/Services/ProviderAdapter.ts | 10 + .../src/provider/Services/ProviderService.ts | 16 ++ apps/server/src/server.test.ts | 225 ++++++++++++++++++ .../serverRuntimeStartup.reconcile.test.ts | 3 + apps/server/src/ws.ts | 73 ++++++ apps/web/src/components/ChatView.tsx | 167 ++++++++++++- apps/web/src/state/threads.ts | 17 +- docs/user/providers-codex.md | 18 ++ packages/client-runtime/src/rpc/client.ts | 1 + .../src/state/threadCommands.test.ts | 145 +++++++++++ .../src/state/threadCommands.ts | 149 +++++++++++- packages/contracts/src/codexGoal.ts | 77 ++++++ packages/contracts/src/index.ts | 1 + packages/contracts/src/providerRuntime.ts | 31 +++ packages/contracts/src/rpc.ts | 43 ++++ 30 files changed, 1596 insertions(+), 16 deletions(-) create mode 100644 packages/client-runtime/src/state/threadCommands.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 c771aaebcb6e..10f680104951 100644 --- a/apps/mobile/src/features/threads/ThreadComposer.tsx +++ b/apps/mobile/src/features/threads/ThreadComposer.tsx @@ -552,8 +552,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 2c6860199722..884ee11a8523 100644 --- a/apps/mobile/src/features/threads/ThreadDetailScreen.tsx +++ b/apps/mobile/src/features/threads/ThreadDetailScreen.tsx @@ -1,5 +1,16 @@ import { type EnvironmentConnectionPhase } from "@t3tools/client-runtime/connection"; -import type { EnvironmentThreadStatus } from "@t3tools/client-runtime/state/threads"; +import { + formatCodexGoalDescription, + formatCodexGoalError, + formatCodexGoalStatus, + formatCodexGoalUsage, + parseCodexGoalCommand, + type EnvironmentThreadStatus, +} from "@t3tools/client-runtime/state/threads"; +import { + isAtomCommandInterrupted, + squashAtomCommandFailure, +} from "@t3tools/client-runtime/state/runtime"; import { useKeyboardChatComposerInset, useKeyboardScrollToEnd } from "@legendapp/list/keyboard"; import type { LegendListRef } from "@legendapp/list/react-native"; import { HeaderHeightContext } from "@react-navigation/elements"; @@ -30,6 +41,7 @@ import { import { isLiquidGlassSupported, LiquidGlassView } from "@callstack/liquid-glass"; import { AppState, + Alert, Keyboard, Platform, useWindowDimensions, @@ -52,6 +64,9 @@ import Animated, { import { useSafeAreaInsets } from "react-native-safe-area-context"; import { ControlPill } from "../../components/ControlPill"; +import { AppText as Text } from "../../components/AppText"; +import { threadEnvironment, useCodexGoal } from "../../state/threads"; +import { useAtomCommand } from "../../state/use-atom-command"; import { useAppearancePreferences } from "../settings/appearance/AppearancePreferencesProvider"; import type { ComposerEditorHandle } from "../../components/ComposerEditor"; import type { StatusTone } from "../../components/StatusPill"; @@ -258,11 +273,17 @@ export const ThreadDetailScreen = memo(function ThreadDetailScreen(props: Thread const listRef = useRef(null); const feedTouchStartRef = useRef<{ pageX: number; pageY: number } | null>(null); const selectedThreadKeyRef = useRef(selectedThreadKey); + const draftMessageRef = useRef(props.draftMessage); const lastScrolledSubmittedMessageIdRef = useRef(null); const [composerExpanded, setComposerExpanded] = useState(false); 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 @@ -446,6 +467,13 @@ 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, + ); useStreamingHaptics(props.selectedThread.id, props.selectedThreadFeed); const selectedProviderSkills = useMemo( () => @@ -458,6 +486,10 @@ export const ThreadDetailScreen = memo(function ThreadDetailScreen(props: Thread selectedThreadKeyRef.current = selectedThreadKey; }, [selectedThreadKey]); + useLayoutEffect(() => { + draftMessageRef.current = props.draftMessage; + }, [props.draftMessage]); + useEffect(() => { setAnchorMessageId(null); setSubmittedMessageId(null); @@ -521,6 +553,78 @@ 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 thread's provider to load before running a Goal command.", + ); + 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: { + threadId: props.selectedThread.id, + ...(goalCommand.objective === undefined + ? {} + : { objective: goalCommand.objective }), + ...(goalCommand.status === undefined ? {} : { status: goalCommand.status }), + }, + }); + 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", @@ -544,11 +648,20 @@ export const ThreadDetailScreen = memo(function ThreadDetailScreen(props: Thread return messageId; }, [ anchorMessageId, + clearCodexGoal, + getCodexGoal, props.onSendMessage, + props.draftAttachments, + props.draftMessage, + props.environmentId, + props.onChangeDraftMessage, + props.selectedThread.id, props.selectedThread.latestTurn, props.selectedThreadQueueCount, selectedThreadFeed, selectedThreadKey, + selectedProvider?.driver, + setCodexGoal, ]); const collapseComposer = useCallback(() => { @@ -739,6 +852,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, +): CodexGoal | null { + const result = useAtomValue( + environmentId !== null && threadId !== null + ? threadEnvironment.codexGoal({ environmentId, input: { threadId } }) + : EMPTY_CODEX_GOAL_ATOM, + ); + return Option.getOrNull(AsyncResult.value(result)); +} export function useEnvironmentThread( environmentId: EnvironmentId | null, diff --git a/apps/server/integration/orphanedProviderSessionStartup.integration.test.ts b/apps/server/integration/orphanedProviderSessionStartup.integration.test.ts index 78a33364f5a3..6cf071c608ed 100644 --- a/apps/server/integration/orphanedProviderSessionStartup.integration.test.ts +++ b/apps/server/integration/orphanedProviderSessionStartup.integration.test.ts @@ -117,6 +117,9 @@ const startupDependencies = Layer.mergeAll( getInstanceInfo: () => Effect.die("unused"), rollbackConversation: () => Effect.die("unused"), uploadFeedback: () => Effect.die("unused"), + getCodexGoal: () => Effect.die("unused"), + setCodexGoal: () => Effect.die("unused"), + clearCodexGoal: () => Effect.die("unused"), streamEvents: Stream.empty, }), ); diff --git a/apps/server/src/auth/RpcAuthorization.ts b/apps/server/src/auth/RpcAuthorization.ts index 28ceac4cec99..2b913a8354dd 100644 --- a/apps/server/src/auth/RpcAuthorization.ts +++ b/apps/server/src/auth/RpcAuthorization.ts @@ -87,6 +87,10 @@ export const RPC_REQUIRED_SCOPES = { [WS_METHODS.attachmentsCreateUploadUrl]: AuthOrchestrationOperateScope, [WS_METHODS.attachmentsDelete]: AuthOrchestrationOperateScope, [WS_METHODS.providerUploadFeedback]: AuthOrchestrationOperateScope, + [WS_METHODS.codexGoalGet]: AuthOrchestrationReadScope, + [WS_METHODS.codexGoalSet]: AuthOrchestrationOperateScope, + [WS_METHODS.codexGoalClear]: AuthOrchestrationOperateScope, + [WS_METHODS.subscribeCodexGoal]: AuthOrchestrationReadScope, [WS_METHODS.subscribeVcsStatus]: AuthOrchestrationReadScope, [WS_METHODS.subscribeResourceTelemetry]: AuthOrchestrationReadScope, [WS_METHODS.vcsRefreshStatus]: AuthOrchestrationReadScope, diff --git a/apps/server/src/orchestration/Layers/CheckpointReactor.test.ts b/apps/server/src/orchestration/Layers/CheckpointReactor.test.ts index ca4cb7afd9ab..9cf4b7d90981 100644 --- a/apps/server/src/orchestration/Layers/CheckpointReactor.test.ts +++ b/apps/server/src/orchestration/Layers/CheckpointReactor.test.ts @@ -126,6 +126,9 @@ function createProviderServiceHarness( }), rollbackConversation, uploadFeedback: () => unsupported(), + getCodexGoal: () => unsupported(), + setCodexGoal: () => unsupported(), + clearCodexGoal: () => unsupported(), get streamEvents() { return Stream.fromPubSub(runtimeEventPubSub); }, diff --git a/apps/server/src/orchestration/Layers/ProviderCommandReactor.test.ts b/apps/server/src/orchestration/Layers/ProviderCommandReactor.test.ts index a22a7acfb705..44e0957d9787 100644 --- a/apps/server/src/orchestration/Layers/ProviderCommandReactor.test.ts +++ b/apps/server/src/orchestration/Layers/ProviderCommandReactor.test.ts @@ -352,6 +352,9 @@ describe("ProviderCommandReactor", () => { }, rollbackConversation: () => unsupported(), uploadFeedback: () => unsupported(), + getCodexGoal: () => unsupported(), + setCodexGoal: () => unsupported(), + clearCodexGoal: () => unsupported(), get streamEvents() { return Stream.fromPubSub(runtimeEventPubSub); }, diff --git a/apps/server/src/orchestration/Layers/ProviderRuntimeIngestion.test.ts b/apps/server/src/orchestration/Layers/ProviderRuntimeIngestion.test.ts index 84858b6affe9..1db50fddf774 100644 --- a/apps/server/src/orchestration/Layers/ProviderRuntimeIngestion.test.ts +++ b/apps/server/src/orchestration/Layers/ProviderRuntimeIngestion.test.ts @@ -126,6 +126,9 @@ function createProviderServiceHarness() { }, rollbackConversation: () => unsupported(), uploadFeedback: () => unsupported(), + getCodexGoal: () => unsupported(), + setCodexGoal: () => unsupported(), + clearCodexGoal: () => unsupported(), get streamEvents() { return Stream.fromPubSub(runtimeEventPubSub); }, diff --git a/apps/server/src/provider/Layers/CodexAdapter.test.ts b/apps/server/src/provider/Layers/CodexAdapter.test.ts index 4986d02c9b67..30cafc3f1b66 100644 --- a/apps/server/src/provider/Layers/CodexAdapter.test.ts +++ b/apps/server/src/provider/Layers/CodexAdapter.test.ts @@ -34,6 +34,7 @@ import * as Scope from "effect/Scope"; import * as Stream from "effect/Stream"; import * as TestClock from "effect/testing/TestClock"; import * as CodexErrors from "effect-codex-app-server/errors"; +import * as EffectCodexSchema from "effect-codex-app-server/schema"; import { ServerConfig } from "../../config.ts"; import { ServerSettingsService } from "../../serverSettings.ts"; @@ -42,6 +43,7 @@ import type { CodexAdapterShape } from "../Services/CodexAdapter.ts"; import { ProviderSessionDirectory } from "../Services/ProviderSessionDirectory.ts"; import { type CodexSessionRuntimeOptions, + type CodexSessionRuntimeGoalSetInput, type CodexSessionRuntimeSendTurnInput, type CodexSessionRuntimeShape, type CodexThreadSnapshot, @@ -59,6 +61,22 @@ const asTurnId = (value: string): TurnId => TurnId.make(value); const asEventId = (value: string): EventId => EventId.make(value); const asItemId = (value: string): ProviderItemId => ProviderItemId.make(value); +function makeNativeGoal( + overrides: Partial = {}, +): EffectCodexSchema.V2ThreadGoalUpdatedNotification["goal"] { + return { + threadId: "provider-thread-1", + objective: "Ship native Goal controls", + status: "active", + tokenBudget: 100_000, + tokensUsed: 12_000, + timeUsedSeconds: 90, + createdAt: 1_777_000_000, + updatedAt: 1_777_000_090, + ...overrides, + }; +} + class FakeCodexRuntime implements CodexSessionRuntimeShape { private readonly eventQueue = Effect.runSync(Queue.unbounded()); private readonly now = "2026-01-01T00:00:00.000Z"; @@ -108,6 +126,21 @@ class FakeCodexRuntime implements CodexSessionRuntimeShape { Promise.resolve({ threadId: "provider-thread-1" }), ); + public readonly getGoalImpl = vi.fn(() => Promise.resolve({ goal: makeNativeGoal() })); + + public readonly setGoalImpl = vi.fn((input: CodexSessionRuntimeGoalSetInput) => + Promise.resolve({ + goal: makeNativeGoal({ + objective: input.objective ?? "Ship native Goal controls", + status: input.status ?? "active", + tokenBudget: input.tokenBudget ?? 100_000, + updatedAt: 1_777_000_100, + }), + }), + ); + + public readonly clearGoalImpl = vi.fn(() => Promise.resolve({ cleared: true })); + public readonly respondToRequestImpl = vi.fn( (_requestId: ApprovalRequestId, _decision: ProviderApprovalDecision): Promise => Promise.resolve(undefined), @@ -150,6 +183,14 @@ class FakeCodexRuntime implements CodexSessionRuntimeShape { return Effect.promise(() => this.uploadFeedbackImpl(reason)); } + getGoal: CodexSessionRuntimeShape["getGoal"] = Effect.promise(() => this.getGoalImpl()); + + setGoal(input: CodexSessionRuntimeGoalSetInput) { + return Effect.promise(() => this.setGoalImpl(input)); + } + + clearGoal = Effect.promise(() => this.clearGoalImpl()); + respondToRequest(requestId: ApprovalRequestId, decision: ProviderApprovalDecision) { return Effect.promise(() => this.respondToRequestImpl(requestId, decision)); } @@ -317,6 +358,21 @@ const sessionErrorLayer = it.layer( ), ); +const startGoalSession = (value: string) => + Effect.gen(function* () { + const adapter = yield* CodexAdapter; + const threadId = asThreadId(value); + yield* adapter.startSession({ + provider: ProviderDriverKind.make("codex"), + threadId, + runtimeMode: "full-access", + }); + const runtime = sessionRuntimeFactory.lastRuntime; + NodeAssert.ok(runtime); + NodeAssert.ok(adapter.codexGoal); + return { goal: adapter.codexGoal, runtime, threadId }; + }); + sessionErrorLayer("CodexAdapterLive session errors", (it) => { it.effect("maps missing adapter sessions to ProviderAdapterSessionNotFoundError", () => Effect.gen(function* () { @@ -405,6 +461,57 @@ sessionErrorLayer("CodexAdapterLive session errors", (it) => { }), ); + it.effect("routes the native Goal lifecycle through the active Codex runtime", () => + Effect.gen(function* () { + const { goal, runtime, threadId } = yield* startGoalSession("goal-thread"); + + const current = yield* goal.get(threadId); + NodeAssert.equal(current?.objective, "Ship native Goal controls"); + yield* goal.set({ + threadId, + objective: "Create the native Goal", + status: "active", + }); + yield* goal.set({ threadId, status: "paused" }); + yield* goal.set({ threadId, status: "active" }); + yield* goal.set({ threadId, objective: "Steer the active Goal" }); + const cleared = yield* goal.clear(threadId); + + NodeAssert.deepStrictEqual( + runtime.setGoalImpl.mock.calls.map(([input]) => input), + [ + { objective: "Create the native Goal", status: "active" }, + { status: "paused" }, + { status: "active" }, + { objective: "Steer the active Goal" }, + ], + ); + NodeAssert.equal(runtime.getGoalImpl.mock.calls.length, 1); + NodeAssert.equal(runtime.clearGoalImpl.mock.calls.length, 1); + NodeAssert.deepStrictEqual(cleared, { cleared: true }); + }), + ); + + it.effect("maps native Goal request rejection to an adapter request error", () => + Effect.gen(function* () { + const { goal, runtime, threadId } = yield* startGoalSession("goal-rejection-thread"); + runtime.getGoal = Effect.fail( + new CodexErrors.CodexAppServerRequestError({ + code: -32603, + errorMessage: "native Goal rejected", + method: "thread/goal/get", + }), + ); + + const result = yield* goal.get(threadId).pipe(Effect.result); + NodeAssert.equal(result._tag, "Failure"); + NodeAssert.equal(result.failure._tag, "ProviderAdapterRequestError"); + if (result.failure._tag === "ProviderAdapterRequestError") { + NodeAssert.equal(result.failure.method, "thread/goal/get"); + } + }), + ); + it.effect("passes configured launch args into the session runtime", () => { const runtimeFactory = makeRuntimeFactory(); const layer = Layer.effect( @@ -618,6 +725,53 @@ lifecycleLayer("CodexAdapterLive lifecycle", (it) => { }), ); + it.effect("maps native Goal updated and cleared notifications", () => + Effect.gen(function* () { + const { adapter, runtime } = yield* startLifecycleRuntime(); + const eventsFiber = yield* adapter.streamEvents.pipe( + Stream.take(2), + Stream.runCollect, + Effect.forkChild, + ); + yield* runtime.emit({ + id: asEventId("evt-goal-updated"), + kind: "notification", + provider: ProviderDriverKind.make("codex"), + createdAt: "2026-01-01T00:00:00.000Z", + method: "thread/goal/updated", + threadId: asThreadId("thread-1"), + payload: { + threadId: "provider-thread-1", + goal: makeNativeGoal({ + objective: "Updated asynchronously", + status: "paused", + tokenBudget: 50_000, + tokensUsed: 5_000, + timeUsedSeconds: 45, + updatedAt: 1_777_000_045, + }), + }, + } satisfies ProviderEvent); + yield* runtime.emit({ + id: asEventId("evt-goal-cleared"), + kind: "notification", + provider: ProviderDriverKind.make("codex"), + createdAt: "2026-01-01T00:00:01.000Z", + method: "thread/goal/cleared", + threadId: asThreadId("thread-1"), + payload: { threadId: "provider-thread-1" }, + } satisfies ProviderEvent); + const [updated, cleared] = Array.from(yield* Fiber.join(eventsFiber)); + NodeAssert.equal(updated?.type, "thread.goal.updated"); + if (updated?.type === "thread.goal.updated") { + NodeAssert.equal(updated.threadId, "thread-1"); + NodeAssert.equal(updated.payload.goal.objective, "Updated asynchronously"); + } + NodeAssert.equal(cleared?.type, "thread.goal.cleared"); + NodeAssert.equal(cleared?.threadId, "thread-1"); + }), + ); + it.effect("maps completed agent message items to canonical item.completed events", () => Effect.gen(function* () { const { adapter, runtime } = yield* startLifecycleRuntime(); diff --git a/apps/server/src/provider/Layers/CodexAdapter.ts b/apps/server/src/provider/Layers/CodexAdapter.ts index 0f7d999662e9..8a4303240d0d 100644 --- a/apps/server/src/provider/Layers/CodexAdapter.ts +++ b/apps/server/src/provider/Layers/CodexAdapter.ts @@ -10,6 +10,7 @@ import { type CanonicalItemType, type CanonicalRequestType, + type CodexGoal, type CodexSettings, ProviderDriverKind, type ProviderEvent, @@ -463,6 +464,18 @@ function runtimeEventBase( }; } +function toCodexGoal(goal: EffectCodexSchema.V2ThreadGoalUpdatedNotification["goal"]): CodexGoal { + return { + objective: goal.objective, + status: goal.status, + ...(goal.tokenBudget !== undefined ? { tokenBudget: goal.tokenBudget } : {}), + tokensUsed: goal.tokensUsed, + timeUsedSeconds: goal.timeUsedSeconds, + createdAt: goal.createdAt, + updatedAt: goal.updatedAt, + }; +} + function mapItemLifecycle( event: ProviderEvent, canonicalThreadId: ThreadId, @@ -1035,6 +1048,34 @@ function mapToRuntimeEvents( ]; } + if (event.method === "thread/goal/updated") { + const payload = readPayload(EffectCodexSchema.V2ThreadGoalUpdatedNotification, event.payload); + if (!payload) { + return []; + } + return [ + { + type: "thread.goal.updated", + ...runtimeEventBase(event, canonicalThreadId), + payload: { goal: toCodexGoal(payload.goal) }, + }, + ]; + } + + if (event.method === "thread/goal/cleared") { + const payload = readPayload(EffectCodexSchema.V2ThreadGoalClearedNotification, event.payload); + if (!payload) { + return []; + } + return [ + { + type: "thread.goal.cleared", + ...runtimeEventBase(event, canonicalThreadId), + payload: {}, + }, + ]; + } + if (event.method === "turn/started") { const turnId = event.turnId; if (!turnId) { @@ -1920,6 +1961,40 @@ export const makeCodexAdapter = Effect.fn("makeCodexAdapter")(function* ( ), ); + const codexGoal: NonNullable = { + get: (threadId) => + requireSession(threadId).pipe( + Effect.flatMap((session) => session.runtime.getGoal), + Effect.map((response) => (response.goal ? toCodexGoal(response.goal) : null)), + Effect.mapError((cause) => + cause._tag === "ProviderAdapterSessionNotFoundError" + ? cause + : mapCodexRuntimeError(threadId, "thread/goal/get", cause), + ), + ), + set: (input) => { + const { threadId, ...params } = input; + return requireSession(threadId).pipe( + Effect.flatMap((session) => session.runtime.setGoal(params)), + Effect.map((response) => toCodexGoal(response.goal)), + Effect.mapError((cause) => + cause._tag === "ProviderAdapterSessionNotFoundError" + ? cause + : mapCodexRuntimeError(threadId, "thread/goal/set", cause), + ), + ); + }, + clear: (threadId) => + requireSession(threadId).pipe( + Effect.flatMap((session) => session.runtime.clearGoal), + Effect.mapError((cause) => + cause._tag === "ProviderAdapterSessionNotFoundError" + ? cause + : mapCodexRuntimeError(threadId, "thread/goal/clear", cause), + ), + ), + }; + const respondToRequest: CodexAdapterShape["respondToRequest"] = (threadId, requestId, decision) => requireSession(threadId).pipe( Effect.flatMap((session) => session.runtime.respondToRequest(requestId, decision)), @@ -2008,6 +2083,7 @@ export const makeCodexAdapter = Effect.fn("makeCodexAdapter")(function* ( readThread, rollbackThread, uploadFeedback, + codexGoal, respondToRequest, respondToUserInput, stopSession, diff --git a/apps/server/src/provider/Layers/CodexProvider.ts b/apps/server/src/provider/Layers/CodexProvider.ts index 93730046dc49..74cfa8fad27b 100644 --- a/apps/server/src/provider/Layers/CodexProvider.ts +++ b/apps/server/src/provider/Layers/CodexProvider.ts @@ -42,6 +42,13 @@ const CODEX_PRESENTATION = { displayName: "Codex", showInteractionModeToggle: true, } as const; +const CODEX_SLASH_COMMANDS = [ + { + name: "goal", + description: "Manage the native Codex Goal for this thread", + input: { hint: "[status|create|steer|pause|resume|clear|reset] [objective]" }, + }, +] as const; export interface CodexAppServerProviderSnapshot { readonly account: CodexSchema.V2GetAccountResponse; @@ -450,6 +457,7 @@ const makePendingCodexProvider = ( enabled: false, checkedAt, models, + slashCommands: CODEX_SLASH_COMMANDS, skills: [], probe: { installed: false, @@ -466,6 +474,7 @@ const makePendingCodexProvider = ( enabled: true, checkedAt, models, + slashCommands: CODEX_SLASH_COMMANDS, skills: [], probe: { installed: false, @@ -536,6 +545,7 @@ export const checkCodexProviderStatus = Effect.fn("checkCodexProviderStatus")(fu enabled: false, checkedAt, models: emptyModels, + slashCommands: CODEX_SLASH_COMMANDS, skills: [], probe: { installed: false, @@ -568,6 +578,7 @@ export const checkCodexProviderStatus = Effect.fn("checkCodexProviderStatus")(fu enabled: codexSettings.enabled, checkedAt, models: emptyModels, + slashCommands: CODEX_SLASH_COMMANDS, skills: [], probe: { installed, @@ -587,6 +598,7 @@ export const checkCodexProviderStatus = Effect.fn("checkCodexProviderStatus")(fu enabled: codexSettings.enabled, checkedAt, models: emptyModels, + slashCommands: CODEX_SLASH_COMMANDS, skills: [], probe: { installed: true, @@ -608,6 +620,7 @@ export const checkCodexProviderStatus = Effect.fn("checkCodexProviderStatus")(fu models: snapshot.models, skills: snapshot.skills, slashCommands: [ + ...CODEX_SLASH_COMMANDS, { name: "feedback", description: "Send this thread and Codex logs to OpenAI", diff --git a/apps/server/src/provider/Layers/CodexSessionRuntime.ts b/apps/server/src/provider/Layers/CodexSessionRuntime.ts index b34067b7fb90..5f854b41b5c0 100644 --- a/apps/server/src/provider/Layers/CodexSessionRuntime.ts +++ b/apps/server/src/provider/Layers/CodexSessionRuntime.ts @@ -184,6 +184,11 @@ export interface CodexThreadSnapshot { readonly turns: ReadonlyArray; } +export type CodexSessionRuntimeGoalSetInput = Omit< + EffectCodexSchema.V2ThreadGoalSetParams, + "threadId" +>; + export interface CodexSessionRuntimeShape { readonly start: () => Effect.Effect; readonly getSession: Effect.Effect; @@ -198,6 +203,17 @@ export interface CodexSessionRuntimeShape { readonly uploadFeedback: ( reason?: string, ) => Effect.Effect; + readonly getGoal: Effect.Effect< + EffectCodexSchema.V2ThreadGoalGetResponse, + CodexSessionRuntimeError + >; + readonly setGoal: ( + input: CodexSessionRuntimeGoalSetInput, + ) => Effect.Effect; + readonly clearGoal: Effect.Effect< + EffectCodexSchema.V2ThreadGoalClearResponse, + CodexSessionRuntimeError + >; readonly respondToRequest: ( requestId: ApprovalRequestId, decision: ProviderApprovalDecision, @@ -732,6 +748,8 @@ function readNotificationThreadId(notification: CodexServerNotification): string case "thread/closed": case "thread/name/updated": case "thread/tokenUsage/updated": + case "thread/goal/updated": + case "thread/goal/cleared": case "turn/started": case "hook/started": case "turn/completed": @@ -2211,6 +2229,22 @@ export const makeCodexSessionRuntime = ( threadId: providerThreadId, }); }), + getGoal: Effect.gen(function* () { + const providerThreadId = yield* readProviderThreadId; + return yield* client.request("thread/goal/get", { threadId: providerThreadId }); + }), + setGoal: (input) => + Effect.gen(function* () { + const providerThreadId = yield* readProviderThreadId; + return yield* client.request("thread/goal/set", { + threadId: providerThreadId, + ...input, + }); + }), + clearGoal: Effect.gen(function* () { + const providerThreadId = yield* readProviderThreadId; + return yield* client.request("thread/goal/clear", { threadId: providerThreadId }); + }), respondToRequest: (requestId, decision) => Effect.gen(function* () { const pending = (yield* Ref.get(pendingApprovalsRef)).get(requestId); diff --git a/apps/server/src/provider/Layers/ProviderService.test.ts b/apps/server/src/provider/Layers/ProviderService.test.ts index bd89dc4f8812..61e464fd7132 100644 --- a/apps/server/src/provider/Layers/ProviderService.test.ts +++ b/apps/server/src/provider/Layers/ProviderService.test.ts @@ -4,6 +4,8 @@ import * as NodeOS from "node:os"; import * as NodePath from "node:path"; import type { + CodexGoal, + CodexGoalSetInput, ProviderApprovalDecision, ProviderRuntimeEvent, ProviderSendTurnInput, @@ -14,7 +16,6 @@ import type { } from "@t3tools/contracts"; import { ApprovalRequestId, - EnvironmentId, EventId, ProviderDriverKind, ProviderInstanceId, @@ -93,6 +94,7 @@ type LegacyProviderRuntimeEvent = { function makeFakeCodexAdapter(provider: ProviderDriverKind = CODEX_DRIVER) { const sessions = new Map(); + const goals = new Map(); const runtimeEventPubSub = Effect.runSync(PubSub.unbounded()); const startSession = vi.fn((input: ProviderSessionStartInput) => @@ -213,6 +215,29 @@ function makeFakeCodexAdapter(provider: ProviderDriverKind = CODEX_DRIVER) { }), ); + const getCodexGoal = vi.fn((threadId: ThreadId) => Effect.succeed(goals.get(threadId) ?? null)); + const setCodexGoal = vi.fn((input: CodexGoalSetInput) => + Effect.sync(() => { + const { threadId, ...updates } = input; + const next: CodexGoal = { + objective: "Test Goal", + status: "active", + tokenBudget: null, + tokensUsed: 0, + timeUsedSeconds: 0, + createdAt: 1_777_000_000, + updatedAt: 1_777_000_001, + ...goals.get(threadId), + ...updates, + }; + goals.set(threadId, next); + return next; + }), + ); + const clearCodexGoal = vi.fn((threadId: ThreadId) => + Effect.sync(() => ({ cleared: goals.delete(threadId) })), + ); + const adapter: ProviderAdapterShape = { provider, capabilities: { @@ -228,7 +253,16 @@ function makeFakeCodexAdapter(provider: ProviderDriverKind = CODEX_DRIVER) { hasSession, readThread, rollbackThread, - ...(provider === CODEX_DRIVER ? { uploadFeedback } : {}), + ...(provider === CODEX_DRIVER + ? { + uploadFeedback, + codexGoal: { + get: getCodexGoal, + set: setCodexGoal, + clear: clearCodexGoal, + }, + } + : {}), stopAll, get streamEvents() { return Stream.fromPubSub(runtimeEventPubSub); @@ -265,6 +299,9 @@ function makeFakeCodexAdapter(provider: ProviderDriverKind = CODEX_DRIVER) { readThread, rollbackThread, uploadFeedback, + getCodexGoal, + setCodexGoal, + clearCodexGoal, stopAll, }; } @@ -928,6 +965,96 @@ it.effect( ); routing.layer("ProviderServiceLive routing", (it) => { + it.effect("keeps native Codex Goals scoped to their routed threads", () => + Effect.gen(function* () { + const provider = yield* ProviderService.ProviderService; + const goals = [ + [asThreadId("goal-thread-1"), "First thread Goal"], + [asThreadId("goal-thread-2"), "Second thread Goal"], + ] as const; + yield* Effect.forEach( + goals, + ([threadId, objective]) => + Effect.gen(function* () { + yield* provider.startSession(threadId, { + provider: CODEX_DRIVER, + providerInstanceId: codexInstanceId, + threadId, + cwd: `/tmp/${threadId}`, + runtimeMode: "full-access", + }); + yield* provider.setCodexGoal({ threadId, objective, status: "active" }); + assert.equal((yield* provider.getCodexGoal(threadId))?.objective, objective); + }), + { discard: true }, + ); + assert.deepEqual( + routing.codex.setCodexGoal.mock.calls.slice(-2).map(([input]) => input.threadId), + goals.map(([threadId]) => threadId), + ); + yield* Effect.forEach(goals, ([threadId]) => provider.stopSession({ threadId }), { + discard: true, + }); + routing.codex.startSession.mockClear(); + routing.codex.stopSession.mockClear(); + }), + ); + + it.effect("reads Codex Goal snapshots without recovering inactive sessions", () => + Effect.gen(function* () { + const provider = yield* ProviderService.ProviderService; + const threadId = asThreadId("inactive-goal-thread"); + yield* provider.startSession(threadId, { + provider: CODEX_DRIVER, + providerInstanceId: codexInstanceId, + threadId, + cwd: "/tmp/inactive-goal-thread", + runtimeMode: "full-access", + }); + yield* provider.setCodexGoal({ threadId, objective: "Resume only on demand" }); + yield* provider.stopSession({ threadId }); + routing.codex.startSession.mockClear(); + routing.codex.getCodexGoal.mockClear(); + + const snapshot = yield* provider.getCodexGoal(threadId, { allowRecovery: false }); + assert.equal(snapshot, null); + assert.equal(routing.codex.startSession.mock.calls.length, 0); + assert.equal(routing.codex.getCodexGoal.mock.calls.length, 0); + + const recovered = yield* provider.getCodexGoal(threadId); + assert.equal(recovered?.objective, "Resume only on demand"); + assert.equal(routing.codex.startSession.mock.calls.length, 1); + assert.equal(routing.codex.getCodexGoal.mock.calls.length, 1); + + yield* provider.stopSession({ threadId }); + routing.codex.startSession.mockClear(); + routing.codex.stopSession.mockClear(); + }), + ); + + it.effect("rejects native Codex Goal operations for unsupported providers", () => + Effect.gen(function* () { + const provider = yield* ProviderService.ProviderService; + const threadId = asThreadId("claude-goal-thread"); + yield* provider.startSession(threadId, { + provider: CLAUDE_AGENT_DRIVER, + providerInstanceId: claudeAgentInstanceId, + threadId, + cwd: "/tmp/claude-goal-thread", + runtimeMode: "full-access", + }); + + const result = yield* provider.getCodexGoal(threadId).pipe(Effect.result); + assert.equal(result._tag, "Failure"); + if (result._tag === "Failure") { + assert.equal(result.failure._tag, "ProviderValidationError"); + } + yield* provider.stopSession({ threadId }); + routing.claude.startSession.mockClear(); + routing.claude.stopSession.mockClear(); + }), + ); + it.effect("routes provider operations and rollback conversation", () => Effect.gen(function* () { const provider = yield* ProviderService.ProviderService; diff --git a/apps/server/src/provider/Layers/ProviderService.ts b/apps/server/src/provider/Layers/ProviderService.ts index b8cd0df539ac..4d6a74b42877 100644 --- a/apps/server/src/provider/Layers/ProviderService.ts +++ b/apps/server/src/provider/Layers/ProviderService.ts @@ -1219,6 +1219,63 @@ const makeProviderService = Effect.fn("makeProviderService")(function* ( ), ); + const getCodexGoal: ProviderServiceMethod<"getCodexGoal"> = Effect.fn("getCodexGoal")( + function* (threadId, options) { + const routed = yield* resolveRoutableSession({ + threadId, + operation: "ProviderService.getCodexGoal", + allowRecovery: options?.allowRecovery ?? true, + }); + const goal = routed.adapter.codexGoal; + if (!goal) { + return yield* toValidationError( + "ProviderService.getCodexGoal", + `Provider '${routed.adapter.provider}' does not support native Codex Goals.`, + ); + } + if (!routed.isActive) { + return null; + } + return yield* goal.get(routed.threadId); + }, + ); + + const setCodexGoal: ProviderServiceMethod<"setCodexGoal"> = Effect.fn("setCodexGoal")( + function* (input) { + const routed = yield* resolveRoutableSession({ + threadId: input.threadId, + operation: "ProviderService.setCodexGoal", + allowRecovery: true, + }); + const goal = routed.adapter.codexGoal; + if (!goal) { + return yield* toValidationError( + "ProviderService.setCodexGoal", + `Provider '${routed.adapter.provider}' does not support native Codex Goals.`, + ); + } + return yield* goal.set(input); + }, + ); + + const clearCodexGoal: ProviderServiceMethod<"clearCodexGoal"> = Effect.fn("clearCodexGoal")( + function* (threadId) { + const routed = yield* resolveRoutableSession({ + threadId, + operation: "ProviderService.clearCodexGoal", + allowRecovery: true, + }); + const goal = routed.adapter.codexGoal; + if (!goal) { + return yield* toValidationError( + "ProviderService.clearCodexGoal", + `Provider '${routed.adapter.provider}' does not support native Codex Goals.`, + ); + } + return yield* goal.clear(routed.threadId); + }, + ); + return { startSession, sendTurn, @@ -1231,6 +1288,9 @@ const makeProviderService = Effect.fn("makeProviderService")(function* ( getInstanceInfo, rollbackConversation, uploadFeedback, + getCodexGoal, + setCodexGoal, + clearCodexGoal, // Each access creates a fresh PubSub subscription so that multiple // consumers (ProviderRuntimeIngestion, CheckpointReactor, etc.) each // independently receive all runtime events. diff --git a/apps/server/src/provider/Layers/ProviderSessionReaper.test.ts b/apps/server/src/provider/Layers/ProviderSessionReaper.test.ts index 0b1bc9e149f7..a2842e8b71d3 100644 --- a/apps/server/src/provider/Layers/ProviderSessionReaper.test.ts +++ b/apps/server/src/provider/Layers/ProviderSessionReaper.test.ts @@ -185,6 +185,9 @@ describe("ProviderSessionReaper", () => { }, rollbackConversation: () => unsupported(), uploadFeedback: () => unsupported(), + getCodexGoal: () => unsupported(), + setCodexGoal: () => unsupported(), + clearCodexGoal: () => unsupported(), streamEvents: Stream.empty, }; diff --git a/apps/server/src/provider/Services/ProviderAdapter.ts b/apps/server/src/provider/Services/ProviderAdapter.ts index 634745832b37..1047318a8136 100644 --- a/apps/server/src/provider/Services/ProviderAdapter.ts +++ b/apps/server/src/provider/Services/ProviderAdapter.ts @@ -9,6 +9,9 @@ */ import type { ApprovalRequestId, + CodexGoal, + CodexGoalClearResult, + CodexGoalSetInput, ProviderApprovalDecision, ProviderDriverKind, ProviderUserInputAnswers, @@ -116,6 +119,13 @@ export interface ProviderAdapterShape { numTurns: number, ) => Effect.Effect; + /** Native Codex Goal operations. Absent for providers that do not support them. */ + readonly codexGoal?: { + readonly get: (threadId: ThreadId) => Effect.Effect; + readonly set: (input: CodexGoalSetInput) => Effect.Effect; + readonly clear: (threadId: ThreadId) => Effect.Effect; + }; + /** * Upload a thread to the provider when the adapter supports feedback. */ diff --git a/apps/server/src/provider/Services/ProviderService.ts b/apps/server/src/provider/Services/ProviderService.ts index 545641d2e866..0d54e09645a3 100644 --- a/apps/server/src/provider/Services/ProviderService.ts +++ b/apps/server/src/provider/Services/ProviderService.ts @@ -12,6 +12,9 @@ * @module ProviderService */ import type { + CodexGoal, + CodexGoalClearResult, + CodexGoalSetInput, ProviderInterruptTurnInput, ProviderInstanceId, ProviderRespondToRequestInput, @@ -107,6 +110,19 @@ export interface ProviderServiceShape { readonly numTurns: number; }) => Effect.Effect; + readonly getCodexGoal: ( + threadId: ThreadId, + options?: { readonly allowRecovery?: boolean }, + ) => Effect.Effect; + + readonly setCodexGoal: ( + input: CodexGoalSetInput, + ) => Effect.Effect; + + readonly clearCodexGoal: ( + threadId: ThreadId, + ) => Effect.Effect; + /** * Upload a thread and return the provider's shareable feedback identifier. */ diff --git a/apps/server/src/server.test.ts b/apps/server/src/server.test.ts index 5e4f19172eff..fee486087559 100644 --- a/apps/server/src/server.test.ts +++ b/apps/server/src/server.test.ts @@ -8,6 +8,7 @@ import { AuthAccessTokenType, AuthEnvironmentBootstrapTokenType, AuthTokenExchangeGrantType, + CodexGoalOperationError, CommandId, DEFAULT_SERVER_SETTINGS, EnvironmentId, @@ -27,6 +28,7 @@ import { ProjectId, ProviderDriverKind, ProviderInstanceId, + type ProviderRuntimeEvent, ResolvedKeybindingRule, ThreadId, WS_METHODS, @@ -75,6 +77,7 @@ import * as Socket from "effect/unstable/socket/Socket"; import { vi } from "vite-plus/test"; const TEST_EPOCH = DateTime.makeUnsafe("1970-01-01T00:00:00.000Z"); +const isCodexGoalOperationError = Schema.is(CodexGoalOperationError); const decodeTransferThreadSnapshot = Schema.decodeUnknownEffect( Schema.fromJsonString(OrchestrationThreadDetailSnapshot), ); @@ -110,6 +113,7 @@ import * as ExternalLauncher from "./process/externalLauncher.ts"; import * as RemoteOpenTargets from "./environment/RemoteOpenTargets.ts"; import * as OrchestrationEngine from "./orchestration/Services/OrchestrationEngine.ts"; import { OrchestrationListenerCallbackError } from "./orchestration/Errors.ts"; +import { ProviderUnsupportedError } from "./provider/Errors.ts"; import * as ProjectionSnapshotQuery from "./orchestration/Services/ProjectionSnapshotQuery.ts"; import { SqlitePersistenceMemory } from "./persistence/Layers/Sqlite.ts"; import { PersistenceSqlError } from "./persistence/Errors.ts"; @@ -646,7 +650,21 @@ const buildAppUnderTest = (options?: { ...options?.layers?.providerRegistry, }), Layer.mock(ProviderService.ProviderService)({ + startSession: () => Effect.die("ProviderService not stubbed in this test"), + sendTurn: () => Effect.die("ProviderService not stubbed in this test"), + interruptTurn: () => Effect.die("ProviderService not stubbed in this test"), + respondToRequest: () => Effect.die("ProviderService not stubbed in this test"), + respondToUserInput: () => Effect.die("ProviderService not stubbed in this test"), + stopSession: () => Effect.die("ProviderService not stubbed in this test"), + listSessions: () => Effect.succeed([]), + getCapabilities: () => Effect.die("ProviderService not stubbed in this test"), + getInstanceInfo: () => Effect.die("ProviderService not stubbed in this test"), + rollbackConversation: () => Effect.die("ProviderService not stubbed in this test"), uploadFeedback: () => Effect.die("Provider feedback is not stubbed in this test"), + getCodexGoal: () => Effect.die("ProviderService not stubbed in this test"), + setCodexGoal: () => Effect.die("ProviderService not stubbed in this test"), + clearCodexGoal: () => Effect.die("ProviderService not stubbed in this test"), + streamEvents: Stream.empty, ...options?.layers?.providerService, }), ), @@ -4732,6 +4750,213 @@ it.layer(NodeServices.layer)("server router seam", (it) => { }).pipe(Effect.provide(NodeHttpServer.layerTest)), ); + it.effect("routes native Codex Goal controls and notifications over websocket", () => + Effect.gen(function* () { + const threadId = ThreadId.make("goal-rpc-thread"); + const events = yield* PubSub.unbounded(); + const setInputs: unknown[] = []; + const getOptions: Array<{ readonly allowRecovery?: boolean } | undefined> = []; + const initialGoal = { + objective: "Initial Goal", + status: "active" as const, + tokenBudget: 100_000, + tokensUsed: 1_000, + timeUsedSeconds: 10, + createdAt: 1_777_000_000, + updatedAt: 1_777_000_010, + }; + const steeredGoal = { + ...initialGoal, + objective: "Steered Goal", + updatedAt: 1_777_000_020, + }; + + yield* buildAppUnderTest({ + layers: { + providerService: { + getCodexGoal: (_threadId, options) => + Effect.sync(() => { + getOptions.push(options); + return initialGoal; + }), + setCodexGoal: (input) => + Effect.sync(() => { + setInputs.push(input); + return steeredGoal; + }), + clearCodexGoal: () => Effect.succeed({ cleared: true }), + streamEvents: Stream.fromPubSub(events), + }, + }, + }); + + const wsUrl = yield* getWsServerUrl("/ws"); + const result = yield* Effect.scoped( + withWsRpcClient(wsUrl, (client) => + Effect.gen(function* () { + const current = yield* client[WS_METHODS.codexGoalGet]({ threadId }); + const steered = yield* client[WS_METHODS.codexGoalSet]({ + threadId, + objective: "Steered Goal", + }); + const cleared = yield* client[WS_METHODS.codexGoalClear]({ threadId }); + const snapshotSeen = yield* Deferred.make(); + const streamed = yield* client[WS_METHODS.subscribeCodexGoal]({ threadId }).pipe( + Stream.tap((event) => + event.type === "snapshot" + ? Deferred.succeed(snapshotSeen, undefined).pipe(Effect.ignore) + : Effect.void, + ), + Stream.take(3), + Stream.runCollect, + Effect.forkScoped, + ); + yield* Deferred.await(snapshotSeen); + yield* PubSub.publish(events, { + type: "thread.goal.updated", + eventId: EventId.make("goal-updated-event"), + provider: ProviderDriverKind.make("codex"), + createdAt: "2026-01-01T00:00:00.000Z", + threadId, + payload: { goal: steeredGoal }, + }); + yield* PubSub.publish(events, { + type: "thread.goal.cleared", + eventId: EventId.make("goal-cleared-event"), + provider: ProviderDriverKind.make("codex"), + createdAt: "2026-01-01T00:00:01.000Z", + threadId, + payload: {}, + }); + return { current, steered, cleared, streamed: yield* Fiber.join(streamed) }; + }), + ), + ); + + if (result.current === null) { + throw new Error("Expected native Codex Goal snapshot"); + } + assert.equal(result.current.objective, "Initial Goal"); + assert.equal(result.steered.objective, "Steered Goal"); + assert.deepEqual(result.cleared, { cleared: true }); + assert.deepEqual(setInputs, [{ threadId, objective: "Steered Goal" }]); + assert.deepEqual(getOptions, [undefined, { allowRecovery: false }]); + assert.deepEqual( + Array.from(result.streamed).map((event) => event.type), + ["snapshot", "updated", "cleared"], + ); + }).pipe(Effect.provide(NodeHttpServer.layerTest)), + ); + + it.effect("codexGoalSet failures carry the operation, thread, and provider detail", () => + Effect.gen(function* () { + const threadId = ThreadId.make("goal-error-thread"); + + yield* buildAppUnderTest({ + layers: { + providerService: { + setCodexGoal: () => Effect.fail(new ProviderUnsupportedError({ provider: "claude" })), + streamEvents: Stream.empty, + }, + }, + }); + + const wsUrl = yield* getWsServerUrl("/ws"); + const failure = yield* Effect.scoped( + withWsRpcClient(wsUrl, (client) => + client[WS_METHODS.codexGoalSet]({ threadId, objective: "Ship it" }).pipe(Effect.flip), + ), + ); + + assertTrue(isCodexGoalOperationError(failure)); + assert.equal(failure.operation, "set"); + assert.equal(failure.threadId, threadId); + assert.equal(failure.message, `Codex Goal set failed for thread ${threadId}`); + const cause = failure.cause; + assertTrue(cause instanceof Error); + assert.equal(cause.message, "Provider 'claude' is not implemented"); + }).pipe(Effect.provide(NodeHttpServer.layerTest)), + ); + + it.effect("subscribeCodexGoal delivers goal updates published while the snapshot loads", () => + Effect.gen(function* () { + const threadId = ThreadId.make("goal-race-thread"); + const events = yield* PubSub.unbounded({ replay: 1 }); + const snapshotRequested = yield* Deferred.make(); + const releaseSnapshot = yield* Deferred.make(); + const subscriptionSteps: string[] = []; + const initialGoal = { + objective: "Initial Goal", + status: "active" as const, + tokenBudget: null, + tokensUsed: 0, + timeUsedSeconds: 0, + createdAt: 1_777_000_000, + updatedAt: 1_777_000_000, + }; + const steeredGoal = { + ...initialGoal, + objective: "Steered Goal", + updatedAt: 1_777_000_020, + }; + + yield* buildAppUnderTest({ + layers: { + providerService: { + getCodexGoal: () => + Effect.yieldNow.pipe( + Effect.andThen(Effect.sync(() => subscriptionSteps.push("snapshot-read"))), + Effect.andThen(Deferred.succeed(snapshotRequested, undefined)), + Effect.andThen(Deferred.await(releaseSnapshot)), + Effect.as(initialGoal), + ), + streamEvents: Stream.unwrap( + Effect.sync(() => { + subscriptionSteps.push("live-attached"); + return Stream.fromPubSub(events); + }), + ), + }, + }, + }); + + const wsUrl = yield* getWsServerUrl("/ws"); + const streamed = yield* Effect.scoped( + withWsRpcClient(wsUrl, (client) => + Effect.gen(function* () { + const collected = yield* client[WS_METHODS.subscribeCodexGoal]({ threadId }).pipe( + Stream.take(2), + Stream.runCollect, + Effect.forkScoped, + ); + yield* Deferred.await(snapshotRequested); + yield* PubSub.publish(events, { + type: "thread.goal.updated", + eventId: EventId.make("goal-race-updated-event"), + provider: ProviderDriverKind.make("codex"), + createdAt: "2026-01-01T00:00:00.000Z", + threadId, + payload: { goal: steeredGoal }, + }); + yield* Deferred.succeed(releaseSnapshot, undefined); + return yield* Fiber.join(collected); + }), + ), + ); + + assert.deepEqual(subscriptionSteps, ["live-attached", "snapshot-read"]); + const [snapshot, updated] = Array.from(streamed); + assert.equal(snapshot?.type, "snapshot"); + if (snapshot?.type === "snapshot") { + assert.equal(snapshot.goal?.objective, "Initial Goal"); + } + assert.equal(updated?.type, "updated"); + if (updated?.type === "updated") { + assert.equal(updated.goal.objective, "Steered Goal"); + } + }).pipe(Effect.provide(NodeHttpServer.layerTest)), + ); + it.effect("routes websocket rpc subscribeServerConfig emits provider status updates", () => Effect.gen(function* () { const nextProviders = [ diff --git a/apps/server/src/serverRuntimeStartup.reconcile.test.ts b/apps/server/src/serverRuntimeStartup.reconcile.test.ts index 485cd5bb08a4..57f9ccc24809 100644 --- a/apps/server/src/serverRuntimeStartup.reconcile.test.ts +++ b/apps/server/src/serverRuntimeStartup.reconcile.test.ts @@ -56,6 +56,9 @@ const makeProviderService = (liveThreadIds: ReadonlyArray = []) => getInstanceInfo: () => Effect.die("unused"), rollbackConversation: () => Effect.die("unused"), uploadFeedback: () => Effect.die("unused"), + getCodexGoal: () => Effect.die("unused"), + setCodexGoal: () => Effect.die("unused"), + clearCodexGoal: () => Effect.die("unused"), streamEvents: Stream.empty, }) satisfies ProviderService.ProviderService["Service"]; diff --git a/apps/server/src/ws.ts b/apps/server/src/ws.ts index 55b0be07c667..19da5f40cd7d 100644 --- a/apps/server/src/ws.ts +++ b/apps/server/src/ws.ts @@ -7,6 +7,7 @@ import * as Layer from "effect/Layer"; import * as Option from "effect/Option"; import * as Queue from "effect/Queue"; import * as Ref from "effect/Ref"; +import * as Result from "effect/Result"; import * as Schema from "effect/Schema"; import * as Stream from "effect/Stream"; import { @@ -16,6 +17,9 @@ import { type AuthEnvironmentScope, AuthSessionId, ClientSurface, + type CodexGoalOperation, + CodexGoalOperationError, + type CodexGoalStreamEvent, CommandId, type DiscoveredLocalServerList, EventId, @@ -163,6 +167,11 @@ function legacySetupFailureDescription(cause: unknown): string { return String(cause); } +function codexGoalOperationError(operation: CodexGoalOperation, threadId: ThreadId) { + return (cause: unknown): CodexGoalOperationError => + new CodexGoalOperationError({ operation, threadId, cause }); +} + function projectEntriesFailureContext(error: WorkspaceEntries.WorkspaceEntriesError): { readonly failure: ProjectEntriesFailure; readonly normalizedCwd?: string; @@ -2209,6 +2218,70 @@ const makeWsRpcLayer = ( ), { "rpc.aggregate": "terminal" }, ), + [WS_METHODS.codexGoalGet]: (input) => + observeRpcEffect( + WS_METHODS.codexGoalGet, + providerService + .getCodexGoal(input.threadId) + .pipe(Effect.mapError(codexGoalOperationError("get", input.threadId))), + { "rpc.aggregate": "codex-goal" }, + ), + [WS_METHODS.codexGoalSet]: (input) => + observeRpcEffect( + WS_METHODS.codexGoalSet, + providerService + .setCodexGoal(input) + .pipe(Effect.mapError(codexGoalOperationError("set", input.threadId))), + { "rpc.aggregate": "codex-goal" }, + ), + [WS_METHODS.codexGoalClear]: (input) => + observeRpcEffect( + WS_METHODS.codexGoalClear, + providerService + .clearCodexGoal(input.threadId) + .pipe(Effect.mapError(codexGoalOperationError("clear", input.threadId))), + { "rpc.aggregate": "codex-goal" }, + ), + [WS_METHODS.subscribeCodexGoal]: (input) => + observeRpcStreamEffect( + WS_METHODS.subscribeCodexGoal, + Effect.gen(function* () { + const liveGoalEvents = yield* Stream.toQueue( + providerService.streamEvents.pipe( + Stream.filterMap((event) => { + if (event.threadId !== input.threadId) { + return Result.failVoid; + } + if (event.type === "thread.goal.updated") { + return Result.succeed({ + type: "updated", + threadId: input.threadId, + goal: event.payload.goal, + }); + } + if (event.type === "thread.goal.cleared") { + return Result.succeed({ + type: "cleared", + threadId: input.threadId, + }); + } + return Result.failVoid; + }), + ), + { capacity: "unbounded" }, + ); + const goal = yield* providerService + .getCodexGoal(input.threadId, { allowRecovery: false }) + .pipe(Effect.mapError(codexGoalOperationError("subscribe", input.threadId))); + const snapshot: CodexGoalStreamEvent = { + type: "snapshot", + threadId: input.threadId, + goal, + }; + return Stream.concat(Stream.make(snapshot), Stream.fromQueue(liveGoalEvents)); + }), + { "rpc.aggregate": "codex-goal" }, + ), [WS_METHODS.previewOpen]: (input) => observeRpcEffect(WS_METHODS.previewOpen, previewManager.open(input), { "rpc.aggregate": "preview", diff --git a/apps/web/src/components/ChatView.tsx b/apps/web/src/components/ChatView.tsx index cb1cf698535a..715c89663a2a 100644 --- a/apps/web/src/components/ChatView.tsx +++ b/apps/web/src/components/ChatView.tsx @@ -180,6 +180,7 @@ import { ChevronDownIcon, GitBranchIcon, PaperclipIcon, + TargetIcon, WifiOffIcon, } from "lucide-react"; import { cn, randomHex } from "~/lib/utils"; @@ -254,7 +255,13 @@ import { serverEnvironment, } from "../state/server"; import { terminalEnvironment } from "../state/terminal"; -import { threadEnvironment, useEnvironmentThread } from "../state/threads"; +import { threadEnvironment, useCodexGoal, useEnvironmentThread } from "../state/threads"; +import { + formatCodexGoalDescription, + formatCodexGoalError, + formatCodexGoalStatus, + parseCodexGoalCommand, +} from "@t3tools/client-runtime/state/threads"; import { requestOlderThreadTurns, threadHasOlderTurns, @@ -1297,6 +1304,11 @@ function ChatViewContent(props: ChatViewProps) { const revertThreadCheckpoint = useAtomCommand(threadEnvironment.revertCheckpoint, { 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(); @@ -1665,6 +1677,10 @@ function ChatViewContent(props: ChatViewProps) { [activeThreadEnvironmentId, activeThreadId], ); const activeThreadKey = activeThreadRef ? scopedThreadKey(activeThreadRef) : null; + const activeThreadKeyRef = useRef(activeThreadKey); + useLayoutEffect(() => { + activeThreadKeyRef.current = activeThreadKey; + }, [activeThreadKey]); const changeRequestSnapshotByKey = useAtomValue(threadChangeRequestSnapshotsAtom); const [timelineAnchor, setTimelineAnchor] = useState<{ readonly threadKey: string | null; @@ -2294,6 +2310,10 @@ function ChatViewContent(props: ChatViewProps) { selectedProviderByThreadId ?? threadProvider, ); const selectedProvider: ProviderDriverKind = lockedProvider ?? unlockedSelectedProvider; + const codexGoal = useCodexGoal( + isServerThread && selectedProvider === "codex" ? environmentId : null, + isServerThread && selectedProvider === "codex" ? activeThreadId : null, + ); const phase = derivePhase(activeThread?.session ?? null); const threadActivities = activeThread?.activities ?? EMPTY_ACTIVITIES; const workLogEntries = useMemo(() => deriveWorkLogEntries(threadActivities), [threadActivities]); @@ -4682,8 +4702,8 @@ function ChatViewContent(props: ChatViewProps) { // calm-styled live states flagged `urgent`, like update progress), then // background liveness — its Stop button is the only stop affordance for // settled turns, so a passive "update available" notice must not cover it — - // then calm system banners, the woke and branch-mismatch notices, and the - // informational parked-thread banner last — it must never cover another. + // then calm system banners, the woke and branch-mismatch notices, the parked + // banner, and the passive Goal banner last — it must never cover another. const parkedThreadBannerItem = useMemo(() => { if (!activeThreadSnoozed && !activeThreadSettled) { return null; @@ -4725,6 +4745,24 @@ function ChatViewContent(props: ChatViewProps) { isUnsnoozing, isUnsettling, ]); + 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); @@ -4741,6 +4779,7 @@ function ChatViewContent(props: ChatViewProps) { backgroundLivenessBannerItem === null ? [] : [backgroundLivenessBannerItem]; const wokeThreadItems = wokeThreadBannerItem === null ? [] : [wokeThreadBannerItem]; const parkedThreadItems = parkedThreadBannerItem === null ? [] : [parkedThreadBannerItem]; + const codexGoalItems = codexGoalBannerItem === null ? [] : [codexGoalBannerItem]; if (!localCheckoutBranchMismatch || !showBranchMismatchBanner || !activeBranchMismatchKey) { return [ ...urgentSystemItems, @@ -4748,6 +4787,7 @@ function ChatViewContent(props: ChatViewProps) { ...calmSystemItems, ...wokeThreadItems, ...parkedThreadItems, + ...codexGoalItems, ]; } return [ @@ -4795,10 +4835,12 @@ function ChatViewContent(props: ChatViewProps) { }, }, ...parkedThreadItems, + ...codexGoalItems, ]; }, [ activeBranchMismatchKey, backgroundLivenessBannerItem, + codexGoalBannerItem, handleRestoreThreadBranch, isRestoringThreadBranch, localCheckoutBranchMismatch, @@ -5159,6 +5201,20 @@ function ChatViewContent(props: ChatViewProps) { } const sendCtx = composerRef.current?.getSendContext(); if (!sendCtx?.providerAvailable) { + if ( + sendCtx !== undefined && + !directAnnotation && + sendCtx.images.length === 0 && + parseCodexGoalCommand(promptRef.current) !== null + ) { + toastManager.add( + stackedThreadToast({ + type: "info", + title: "Provider still loading", + description: "Wait for the thread's provider to load before running a Goal command.", + }), + ); + } notifyDirectAnnotationAttached(); return; } @@ -5209,15 +5265,15 @@ function ChatViewContent(props: ChatViewProps) { composerPreviewAnnotations.length + composerReviewComments.length, }); - const feedbackCommand = + const isUnadornedCodexCommand = ctxSelectedProvider === "codex" && + !directAnnotation && composerImages.length === 0 && sendableComposerTerminalContexts.length === 0 && composerElementContexts.length === 0 && composerPreviewAnnotations.length === 0 && - composerReviewComments.length === 0 - ? parseCodexFeedbackCommand(trimmed) - : null; + composerReviewComments.length === 0; + const feedbackCommand = isUnadornedCodexCommand ? parseCodexFeedbackCommand(trimmed) : null; if (feedbackCommand) { if (!isServerThread || activeThread.session === null) { toastManager.add( @@ -5304,6 +5360,103 @@ function ChatViewContent(props: ChatViewProps) { ); return; } + 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 stillOnSubmittedThread = () => activeThreadKeyRef.current === submittedThreadKey; + const clearSubmittedGoalCommandDraft = () => { + if (!stillOnSubmittedThread() || promptRef.current !== promptForSend) return; + promptRef.current = ""; + clearComposerDraftContent(composerDraftTarget); + composerRef.current?.resetCursorState(); + }; + sendInFlightRef.current = true; + 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: { + threadId: activeThreadId, + ...(codexGoalCommand.objective === undefined + ? {} + : { objective: codexGoalCommand.objective }), + ...(codexGoalCommand.status === undefined + ? {} + : { status: codexGoalCommand.status }), + }, + }); + 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 { + sendInFlightRef.current = false; + } + } if (!directAnnotation && showPlanFollowUpPrompt && activeProposedPlan) { const followUp = resolvePlanFollowUpSubmission({ draftText: trimmed, diff --git a/apps/web/src/state/threads.ts b/apps/web/src/state/threads.ts index fd936f99ff23..33217cdc0d27 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, ThreadId } from "@t3tools/contracts"; import * as Option from "effect/Option"; import { AsyncResult, Atom } from "effect/unstable/reactivity"; @@ -28,6 +28,21 @@ 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, +): CodexGoal | null { + const result = useAtomValue( + environmentId !== null && threadId !== null + ? threadEnvironment.codexGoal({ environmentId, input: { threadId } }) + : 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 f696e8877b2e..4d564876da56 100644 --- a/docs/user/providers-codex.md +++ b/docs/user/providers-codex.md @@ -40,6 +40,24 @@ When a Codex tool needs access to an app such as Safari, T3 Code shows the app n approval. You can approve, decline, or cancel the request from the desktop app, web app, or mobile app. Some tools also offer approval for the current session or permanent approval. +## Manage A Codex Goal + +After starting a Codex thread, use `/goal` in the composer to manage that thread's native Codex +Goal: + +```text +/goal create +/goal status +/goal steer +/goal pause +/goal resume +/goal clear +``` + +`/goal ` is shorthand for create, and `/goal reset` is an alias for clear. Goal status +and usage come directly from Codex and stay synchronized when Codex updates them in the background. +Goal commands are only available on Codex threads. + ## I Want Work And Personal Codex Accounts Use one real Codex home and one shadow home. diff --git a/packages/client-runtime/src/rpc/client.ts b/packages/client-runtime/src/rpc/client.ts index bfe57a6c0dd5..3f7721a79a63 100644 --- a/packages/client-runtime/src/rpc/client.ts +++ b/packages/client-runtime/src/rpc/client.ts @@ -50,6 +50,7 @@ export type EnvironmentSubscriptionRpcTag = | typeof WS_METHODS.subscribePreviewEvents | typeof WS_METHODS.subscribeDiscoveredLocalServers | typeof WS_METHODS.subscribeResourceTelemetry + | 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..2977a85819c1 --- /dev/null +++ b/packages/client-runtime/src/state/threadCommands.test.ts @@ -0,0 +1,145 @@ +import { ThreadId, type CodexGoal, type CodexGoalStreamEvent } from "@t3tools/contracts"; +import { describe, expect, it } from "@effect/vitest"; + +import { + applyCodexGoalStreamEvent, + formatCodexGoalDescription, + formatCodexGoalError, + formatCodexGoalStatus, + formatCodexGoalUsage, + parseCodexGoalCommand, +} 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("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 initial = { goal: null, hasNativeUpdate: false }; + const updated = applyCodexGoalStreamEvent(initial, { + type: "updated", + threadId, + goal: goal("Updated asynchronously"), + }); + expect(updated.goal?.objective).toBe("Updated asynchronously"); + expect(applyCodexGoalStreamEvent(updated, { type: "cleared", threadId }).goal).toBeNull(); + }); + + it("does not let a late snapshot overwrite a live native update", () => { + const updated = applyCodexGoalStreamEvent( + { goal: null, hasNativeUpdate: false }, + { + type: "updated", + threadId, + goal: goal("Live update"), + }, + ); + const lateSnapshot: CodexGoalStreamEvent = { + type: "snapshot", + threadId, + goal: goal("Stale snapshot"), + }; + expect(applyCodexGoalStreamEvent(updated, lateSnapshot).goal?.objective).toBe("Live update"); + }); +}); + +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 c540644289df..febc96f2ff13 100644 --- a/packages/client-runtime/src/state/threadCommands.ts +++ b/packages/client-runtime/src/state/threadCommands.ts @@ -1,11 +1,20 @@ +import { + type CodexGoal, + type CodexGoalSetInput, + type CodexGoalStatus, + type CodexGoalStreamEvent, + WS_METHODS, +} from "@t3tools/contracts"; import * as Crypto from "effect/Crypto"; -import { Atom } from "effect/unstable/reactivity"; -import { WS_METHODS } from "@t3tools/contracts"; +import * as Effect from "effect/Effect"; +import * as Stream from "effect/Stream"; +import { Atom, AtomRegistry } from "effect/unstable/reactivity"; import { createAtomCommandScheduler, createEnvironmentCommand, createEnvironmentRpcCommand, + createEnvironmentRpcSubscriptionAtomFamily, } from "./runtime.ts"; import { type ArchiveThreadInput, @@ -51,6 +60,97 @@ import { } from "../operations/commands.ts"; import type { EnvironmentRegistry } from "../connection/registry.ts"; +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 }; + + // Match Codex's `/goal ` shorthand. + return { action: "set", objective: argument, status: "active" }; +} + +interface CodexGoalProjection { + readonly goal: CodexGoal | null; + readonly hasNativeUpdate: boolean; +} + +export function applyCodexGoalStreamEvent( + current: CodexGoalProjection, + event: CodexGoalStreamEvent, +): CodexGoalProjection { + if (event.type === "snapshot") { + return current.hasNativeUpdate ? current : { goal: event.goal, hasNativeUpdate: false }; + } + if (event.type === "updated") return { goal: event.goal, hasNativeUpdate: true }; + return { goal: null, hasNativeUpdate: true }; +} + export type { ArchiveThreadInput, CreateThreadInput, @@ -83,7 +183,52 @@ 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, + transform: (events) => + events.pipe( + Stream.mapAccum( + (): CodexGoalProjection => ({ goal: null, hasNativeUpdate: false }), + (current, event) => { + const next = applyCodexGoalStreamEvent(current, event); + return [next, [next.goal]] as const; + }, + ), + ), + }); + const refreshCodexGoal = ( + target: Parameters[0], + registry: AtomRegistry.AtomRegistry, + ) => Effect.sync(() => registry.refresh(codexGoal(target))); + return { + codexGoal, + getCodexGoal: createEnvironmentRpcCommand(runtime, { + label: "environment-data:codex-goal:get", + tag: WS_METHODS.codexGoalGet, + scheduler, + concurrency, + onSuccess: refreshCodexGoal, + }), + setCodexGoal: createEnvironmentRpcCommand(runtime, { + label: "environment-data:codex-goal:set", + tag: WS_METHODS.codexGoalSet, + scheduler, + concurrency: { + mode: "serial", + key: ({ environmentId, input }: { environmentId: string; input: CodexGoalSetInput }) => + JSON.stringify([environmentId, input.threadId]), + }, + onSuccess: refreshCodexGoal, + }), + clearCodexGoal: createEnvironmentRpcCommand(runtime, { + label: "environment-data:codex-goal:clear", + tag: WS_METHODS.codexGoalClear, + scheduler, + concurrency, + onSuccess: refreshCodexGoal, + }), create: createEnvironmentCommand(runtime, { label: "environment-data:commands:thread:create", execute: (input: CreateThreadInput) => createThread(input), diff --git a/packages/contracts/src/codexGoal.ts b/packages/contracts/src/codexGoal.ts new file mode 100644 index 000000000000..9f41fe7d3a9c --- /dev/null +++ b/packages/contracts/src/codexGoal.ts @@ -0,0 +1,77 @@ +import * as Schema from "effect/Schema"; + +import { NonNegativeInt, ThreadId, TrimmedNonEmptyString } from "./baseSchemas.ts"; + +export const CodexGoalStatus = Schema.Literals([ + "active", + "paused", + "blocked", + "usageLimited", + "budgetLimited", + "complete", +]); +export type CodexGoalStatus = typeof CodexGoalStatus.Type; + +/** Native Codex App Server Goal state, excluding its provider-local thread id. */ +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 CodexGoalSetInput = Schema.Struct({ + threadId: ThreadId, + objective: Schema.optionalKey(TrimmedNonEmptyString), + status: Schema.optionalKey(CodexGoalStatus), + tokenBudget: Schema.optionalKey(Schema.NullOr(NonNegativeInt)), +}); +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 c6daef8687ba..8735db4a8ce5 100644 --- a/packages/contracts/src/index.ts +++ b/packages/contracts/src/index.ts @@ -11,6 +11,7 @@ export * from "./terminal.ts"; export * from "./provider.ts"; export * from "./providerInstance.ts"; export * from "./providerRuntime.ts"; +export * from "./codexGoal.ts"; export * from "./model.ts"; export * from "./keybindings.ts"; export * from "./server.ts"; diff --git a/packages/contracts/src/providerRuntime.ts b/packages/contracts/src/providerRuntime.ts index a734c797b17d..12d622259a7f 100644 --- a/packages/contracts/src/providerRuntime.ts +++ b/packages/contracts/src/providerRuntime.ts @@ -15,6 +15,7 @@ import { } from "./baseSchemas.ts"; import { ProviderInstanceId, ProviderDriverKind } from "./providerInstance.ts"; import { ProviderApprovalOption } from "./orchestration.ts"; +import { CodexGoal } from "./codexGoal.ts"; const TrimmedNonEmptyStringSchema = TrimmedNonEmptyString; const UnknownRecordSchema = Schema.Record(Schema.String, Schema.Unknown); @@ -156,6 +157,8 @@ const ProviderRuntimeEventType = Schema.Literals([ "thread.state.changed", "thread.metadata.updated", "thread.token-usage.updated", + "thread.goal.updated", + "thread.goal.cleared", "thread.realtime.started", "thread.realtime.item-added", "thread.realtime.audio.delta", @@ -207,6 +210,8 @@ const ThreadStartedType = Schema.Literal("thread.started"); const ThreadStateChangedType = Schema.Literal("thread.state.changed"); const ThreadMetadataUpdatedType = Schema.Literal("thread.metadata.updated"); const ThreadTokenUsageUpdatedType = Schema.Literal("thread.token-usage.updated"); +const ThreadGoalUpdatedType = Schema.Literal("thread.goal.updated"); +const ThreadGoalClearedType = Schema.Literal("thread.goal.cleared"); const ThreadRealtimeStartedType = Schema.Literal("thread.realtime.started"); const ThreadRealtimeItemAddedType = Schema.Literal("thread.realtime.item-added"); const ThreadRealtimeAudioDeltaType = Schema.Literal("thread.realtime.audio.delta"); @@ -332,6 +337,14 @@ const ThreadTokenUsageUpdatedPayload = Schema.Struct({ }); export type ThreadTokenUsageUpdatedPayload = typeof ThreadTokenUsageUpdatedPayload.Type; +const ThreadGoalUpdatedPayload = Schema.Struct({ + goal: CodexGoal, +}); +export type ThreadGoalUpdatedPayload = typeof ThreadGoalUpdatedPayload.Type; + +const ThreadGoalClearedPayload = Schema.Struct({}); +export type ThreadGoalClearedPayload = typeof ThreadGoalClearedPayload.Type; + const ThreadRealtimeStartedPayload = Schema.Struct({ realtimeSessionId: Schema.optional(TrimmedNonEmptyStringSchema), }); @@ -841,6 +854,22 @@ const ProviderRuntimeThreadTokenUsageUpdatedEvent = Schema.Struct({ export type ProviderRuntimeThreadTokenUsageUpdatedEvent = typeof ProviderRuntimeThreadTokenUsageUpdatedEvent.Type; +const ProviderRuntimeThreadGoalUpdatedEvent = Schema.Struct({ + ...ProviderRuntimeEventBase.fields, + type: ThreadGoalUpdatedType, + payload: ThreadGoalUpdatedPayload, +}); +export type ProviderRuntimeThreadGoalUpdatedEvent = + typeof ProviderRuntimeThreadGoalUpdatedEvent.Type; + +const ProviderRuntimeThreadGoalClearedEvent = Schema.Struct({ + ...ProviderRuntimeEventBase.fields, + type: ThreadGoalClearedType, + payload: ThreadGoalClearedPayload, +}); +export type ProviderRuntimeThreadGoalClearedEvent = + typeof ProviderRuntimeThreadGoalClearedEvent.Type; + const ProviderRuntimeThreadRealtimeStartedEvent = Schema.Struct({ ...ProviderRuntimeEventBase.fields, type: ThreadRealtimeStartedType, @@ -1149,6 +1178,8 @@ export const ProviderRuntimeEventV2 = Schema.Union([ ProviderRuntimeThreadStateChangedEvent, ProviderRuntimeThreadMetadataUpdatedEvent, ProviderRuntimeThreadTokenUsageUpdatedEvent, + ProviderRuntimeThreadGoalUpdatedEvent, + ProviderRuntimeThreadGoalClearedEvent, ProviderRuntimeThreadRealtimeStartedEvent, ProviderRuntimeThreadRealtimeItemAddedEvent, ProviderRuntimeThreadRealtimeAudioDeltaEvent, diff --git a/packages/contracts/src/rpc.ts b/packages/contracts/src/rpc.ts index 14363cfedff9..3af49bc4ba39 100644 --- a/packages/contracts/src/rpc.ts +++ b/packages/contracts/src/rpc.ts @@ -205,6 +205,14 @@ import { SourceControlRepositoryLookupInput, } from "./sourceControl.ts"; import { VcsError } from "./vcs.ts"; +import { + CodexGoal, + CodexGoalClearResult, + CodexGoalOperationError, + CodexGoalSetInput, + CodexGoalStreamEvent, + CodexGoalThreadInput, +} from "./codexGoal.ts"; export const WS_METHODS = { // Project registry methods @@ -229,6 +237,12 @@ export const WS_METHODS = { // Provider methods providerUploadFeedback: "provider.uploadFeedback", + // Codex native Goal methods + codexGoalGet: "codex.goal.get", + codexGoalSet: "codex.goal.set", + codexGoalClear: "codex.goal.clear", + subscribeCodexGoal: "codex.goal.subscribe", + // VCS methods vcsPull: "vcs.pull", vcsRefreshStatus: "vcs.refreshStatus", @@ -891,6 +905,31 @@ export const WsSubscribePreviewEventsRpc = Rpc.make(WS_METHODS.subscribePreviewE 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: CodexGoalThreadInput, + success: CodexGoalStreamEvent, + error: Schema.Union([CodexGoalOperationError, EnvironmentAuthorizationError]), + stream: true, +}); + export const WsSubscribeDiscoveredLocalServersRpc = Rpc.make( WS_METHODS.subscribeDiscoveredLocalServers, { @@ -1106,6 +1145,10 @@ export const WsRpcGroup = RpcGroup.make( WsPreviewAutomationRespondRpc, WsPreviewAutomationFocusHostRpc, WsSubscribePreviewEventsRpc, + WsCodexGoalGetRpc, + WsCodexGoalSetRpc, + WsCodexGoalClearRpc, + WsSubscribeCodexGoalRpc, WsSubscribeDiscoveredLocalServersRpc, WsSubscribeServerConfigRpc, WsSubscribeServerLifecycleRpc, From cbe146726ded13e4db39968f1ca72e780f53c8ed Mon Sep 17 00:00:00 2001 From: Fredrik Ekman Date: Sun, 23 Aug 2026 09:52:19 +0200 Subject: [PATCH 2/7] fix(codex): keep Goal operations thread scoped --- .../provider/Layers/CodexCollabWire.test.ts | 4 ++ .../provider/Layers/CodexSessionRuntime.ts | 4 ++ .../provider/Layers/ProviderService.test.ts | 21 +++++-- .../src/provider/Layers/ProviderService.ts | 62 ++++++++++++++++--- apps/web/src/components/ChatView.tsx | 9 ++- .../src/state/threadCommands.test.ts | 16 +++++ .../src/state/threadCommands.ts | 19 +++++- 7 files changed, 116 insertions(+), 19 deletions(-) diff --git a/apps/server/src/provider/Layers/CodexCollabWire.test.ts b/apps/server/src/provider/Layers/CodexCollabWire.test.ts index 50e5e819d1f0..87f6fdb11d11 100644 --- a/apps/server/src/provider/Layers/CodexCollabWire.test.ts +++ b/apps/server/src/provider/Layers/CodexCollabWire.test.ts @@ -135,6 +135,8 @@ describe("routeCodexChildNotification", () => { "item/commandExecution/outputDelta", "turn/plan/updated", "thread/name/updated", + "thread/goal/updated", + "thread/goal/cleared", ]) { assert.equal(routeCodexChildNotification(method), "drop", method); } @@ -155,6 +157,8 @@ describe("routeCodexChildNotification", () => { "thread/compacted", "thread/name/updated", "thread/tokenUsage/updated", + "thread/goal/updated", + "thread/goal/cleared", "turn/started", "turn/completed", "turn/plan/updated", diff --git a/apps/server/src/provider/Layers/CodexSessionRuntime.ts b/apps/server/src/provider/Layers/CodexSessionRuntime.ts index 5f854b41b5c0..ebd8ed65fc13 100644 --- a/apps/server/src/provider/Layers/CodexSessionRuntime.ts +++ b/apps/server/src/provider/Layers/CodexSessionRuntime.ts @@ -988,6 +988,8 @@ function shouldSuppressChildConversationNotification( method === "thread/compacted" || method === "thread/name/updated" || method === "thread/tokenUsage/updated" || + method === "thread/goal/updated" || + method === "thread/goal/cleared" || method === "turn/started" || method === "turn/completed" || method === "turn/plan/updated" || @@ -1037,6 +1039,8 @@ const CHILD_CHATTER_METHODS: ReadonlySet = new Set([ "turn/diff/updated", "thread/name/updated", "thread/settings/updated", + "thread/goal/updated", + "thread/goal/cleared", "rawResponseItem/completed", // Child-owned thread lifecycle: the parent adapter maps these onto the // PARENT thread (archived/compacted state), so a child compacting would diff --git a/apps/server/src/provider/Layers/ProviderService.test.ts b/apps/server/src/provider/Layers/ProviderService.test.ts index 61e464fd7132..1acf631471bd 100644 --- a/apps/server/src/provider/Layers/ProviderService.test.ts +++ b/apps/server/src/provider/Layers/ProviderService.test.ts @@ -1043,13 +1043,24 @@ routing.layer("ProviderServiceLive routing", (it) => { cwd: "/tmp/claude-goal-thread", runtimeMode: "full-access", }); + yield* provider.stopSession({ threadId }); + routing.claude.startSession.mockClear(); + routing.claude.stopSession.mockClear(); - const result = yield* provider.getCodexGoal(threadId).pipe(Effect.result); - assert.equal(result._tag, "Failure"); - if (result._tag === "Failure") { - assert.equal(result.failure._tag, "ProviderValidationError"); + const results = yield* Effect.all([ + provider.getCodexGoal(threadId).pipe(Effect.result), + provider + .setCodexGoal({ threadId, objective: "Unsupported Goal", status: "active" }) + .pipe(Effect.result), + provider.clearCodexGoal(threadId).pipe(Effect.result), + ]); + for (const result of results) { + assert.equal(result._tag, "Failure"); + if (result._tag === "Failure") { + assert.equal(result.failure._tag, "ProviderValidationError"); + } } - yield* provider.stopSession({ threadId }); + assert.equal(routing.claude.startSession.mock.calls.length, 0); routing.claude.startSession.mockClear(); routing.claude.stopSession.mockClear(); }), diff --git a/apps/server/src/provider/Layers/ProviderService.ts b/apps/server/src/provider/Layers/ProviderService.ts index 4d6a74b42877..0d05b59a9fc2 100644 --- a/apps/server/src/provider/Layers/ProviderService.ts +++ b/apps/server/src/provider/Layers/ProviderService.ts @@ -1221,12 +1221,12 @@ const makeProviderService = Effect.fn("makeProviderService")(function* ( const getCodexGoal: ProviderServiceMethod<"getCodexGoal"> = Effect.fn("getCodexGoal")( function* (threadId, options) { - const routed = yield* resolveRoutableSession({ + let routed = yield* resolveRoutableSession({ threadId, operation: "ProviderService.getCodexGoal", - allowRecovery: options?.allowRecovery ?? true, + allowRecovery: false, }); - const goal = routed.adapter.codexGoal; + let goal = routed.adapter.codexGoal; if (!goal) { return yield* toValidationError( "ProviderService.getCodexGoal", @@ -1234,7 +1234,21 @@ const makeProviderService = Effect.fn("makeProviderService")(function* ( ); } if (!routed.isActive) { - return null; + if (options?.allowRecovery === false) { + return null; + } + routed = yield* resolveRoutableSession({ + threadId, + operation: "ProviderService.getCodexGoal", + allowRecovery: true, + }); + goal = routed.adapter.codexGoal; + if (!goal) { + return yield* toValidationError( + "ProviderService.getCodexGoal", + `Provider '${routed.adapter.provider}' does not support native Codex Goals.`, + ); + } } return yield* goal.get(routed.threadId); }, @@ -1242,36 +1256,64 @@ const makeProviderService = Effect.fn("makeProviderService")(function* ( const setCodexGoal: ProviderServiceMethod<"setCodexGoal"> = Effect.fn("setCodexGoal")( function* (input) { - const routed = yield* resolveRoutableSession({ + let routed = yield* resolveRoutableSession({ threadId: input.threadId, operation: "ProviderService.setCodexGoal", - allowRecovery: true, + allowRecovery: false, }); - const goal = routed.adapter.codexGoal; + let goal = routed.adapter.codexGoal; if (!goal) { return yield* toValidationError( "ProviderService.setCodexGoal", `Provider '${routed.adapter.provider}' does not support native Codex Goals.`, ); } + if (!routed.isActive) { + routed = yield* resolveRoutableSession({ + threadId: input.threadId, + operation: "ProviderService.setCodexGoal", + allowRecovery: true, + }); + goal = routed.adapter.codexGoal; + if (!goal) { + return yield* toValidationError( + "ProviderService.setCodexGoal", + `Provider '${routed.adapter.provider}' does not support native Codex Goals.`, + ); + } + } return yield* goal.set(input); }, ); const clearCodexGoal: ProviderServiceMethod<"clearCodexGoal"> = Effect.fn("clearCodexGoal")( function* (threadId) { - const routed = yield* resolveRoutableSession({ + let routed = yield* resolveRoutableSession({ threadId, operation: "ProviderService.clearCodexGoal", - allowRecovery: true, + allowRecovery: false, }); - const goal = routed.adapter.codexGoal; + let goal = routed.adapter.codexGoal; if (!goal) { return yield* toValidationError( "ProviderService.clearCodexGoal", `Provider '${routed.adapter.provider}' does not support native Codex Goals.`, ); } + if (!routed.isActive) { + routed = yield* resolveRoutableSession({ + threadId, + operation: "ProviderService.clearCodexGoal", + allowRecovery: true, + }); + goal = routed.adapter.codexGoal; + if (!goal) { + return yield* toValidationError( + "ProviderService.clearCodexGoal", + `Provider '${routed.adapter.provider}' does not support native Codex Goals.`, + ); + } + } return yield* goal.clear(routed.threadId); }, ); diff --git a/apps/web/src/components/ChatView.tsx b/apps/web/src/components/ChatView.tsx index 715c89663a2a..cf7622c156ca 100644 --- a/apps/web/src/components/ChatView.tsx +++ b/apps/web/src/components/ChatView.tsx @@ -1484,6 +1484,7 @@ function ChatViewContent(props: ChatViewProps) { const attachmentPreviewPromotionInFlightByMessageIdRef = useRef>({}); const sendInFlightRef = useRef(false); const feedbackUploadsInFlightRef = useRef(new Set()); + const goalCommandsInFlightRef = useRef(new Set()); const terminalUiOpenByThreadRef = useRef>({}); useLayoutEffect(() => { @@ -5176,7 +5177,8 @@ function ChatViewContent(props: ChatViewProps) { isConnecting || threadDetailLoading || sendInFlightRef.current || - feedbackUploadsInFlightRef.current.has(routeThreadKey) + feedbackUploadsInFlightRef.current.has(routeThreadKey) || + goalCommandsInFlightRef.current.has(routeThreadKey) ) { notifyDirectAnnotationAttached(); return; @@ -5385,6 +5387,7 @@ function ChatViewContent(props: ChatViewProps) { 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; @@ -5392,7 +5395,7 @@ function ChatViewContent(props: ChatViewProps) { clearComposerDraftContent(composerDraftTarget); composerRef.current?.resetCursorState(); }; - sendInFlightRef.current = true; + goalCommandsInFlightRef.current.add(submittedGoalCommandThreadKey); try { if (codexGoalCommand.action === "status") { const result = await getCodexGoal(target); @@ -5454,7 +5457,7 @@ function ChatViewContent(props: ChatViewProps) { clearSubmittedGoalCommandDraft(); return; } finally { - sendInFlightRef.current = false; + goalCommandsInFlightRef.current.delete(submittedGoalCommandThreadKey); } } if (!directAnnotation && showPlanFollowUpPrompt && activeProposedPlan) { diff --git a/packages/client-runtime/src/state/threadCommands.test.ts b/packages/client-runtime/src/state/threadCommands.test.ts index 2977a85819c1..287b683448f7 100644 --- a/packages/client-runtime/src/state/threadCommands.test.ts +++ b/packages/client-runtime/src/state/threadCommands.test.ts @@ -8,6 +8,7 @@ import { formatCodexGoalStatus, formatCodexGoalUsage, parseCodexGoalCommand, + toCodexGoalSubscriptionTarget, } from "./threadCommands.ts"; const threadId = ThreadId.make("thread-1"); @@ -49,6 +50,21 @@ describe("parseCodexGoalCommand", () => { }); }); +describe("toCodexGoalSubscriptionTarget", () => { + it("keys Goal refreshes only by environment and thread", () => { + expect( + toCodexGoalSubscriptionTarget({ + environmentId: "environment-1", + input: { + threadId: "thread-1", + objective: "Do not leak into the subscription key", + status: "active", + }, + }), + ).toEqual({ environmentId: "environment-1", input: { threadId: "thread-1" } }); + }); +}); + describe("applyCodexGoalStreamEvent", () => { it("formats native usage consistently for clients", () => { expect(formatCodexGoalDescription(goal("Ship it"))).toBe( diff --git a/packages/client-runtime/src/state/threadCommands.ts b/packages/client-runtime/src/state/threadCommands.ts index febc96f2ff13..e46f293f1dfd 100644 --- a/packages/client-runtime/src/state/threadCommands.ts +++ b/packages/client-runtime/src/state/threadCommands.ts @@ -69,6 +69,22 @@ export type CodexGoalCommand = const GOAL_USAGE = "Usage: /goal [status | create | steer | pause | resume | clear | reset]"; +export function toCodexGoalSubscriptionTarget< + EnvironmentId, + GoalInput extends { readonly threadId: unknown }, +>(target: { + readonly environmentId: EnvironmentId; + readonly input: GoalInput; +}): { + readonly environmentId: EnvironmentId; + readonly input: { readonly threadId: GoalInput["threadId"] }; +} { + return { + environmentId: target.environmentId, + input: { threadId: target.input.threadId }, + }; +} + export function formatCodexGoalUsage(goal: CodexGoal): string { const budget = goal.tokenBudget == null ? "" : ` / ${goal.tokenBudget.toLocaleString()}`; return `${goal.tokensUsed.toLocaleString()} tokens${budget}, ${goal.timeUsedSeconds.toLocaleString()} seconds`; @@ -220,7 +236,8 @@ export function createThreadEnvironmentAtoms( key: ({ environmentId, input }: { environmentId: string; input: CodexGoalSetInput }) => JSON.stringify([environmentId, input.threadId]), }, - onSuccess: refreshCodexGoal, + onSuccess: (target, registry) => + refreshCodexGoal(toCodexGoalSubscriptionTarget(target), registry), }), clearCodexGoal: createEnvironmentRpcCommand(runtime, { label: "environment-data:codex-goal:clear", From fd4b4bb2e6cbaa5b9300a89bce65b977012054eb Mon Sep 17 00:00:00 2001 From: Fredrik Ekman Date: Sun, 23 Aug 2026 16:23:32 +0200 Subject: [PATCH 3/7] fix(codex): gate Goal commands on session --- apps/mobile/src/features/threads/ThreadDetailScreen.tsx | 7 +++++++ apps/web/src/components/ChatView.tsx | 2 +- 2 files changed, 8 insertions(+), 1 deletion(-) diff --git a/apps/mobile/src/features/threads/ThreadDetailScreen.tsx b/apps/mobile/src/features/threads/ThreadDetailScreen.tsx index 884ee11a8523..df5a860d6c9d 100644 --- a/apps/mobile/src/features/threads/ThreadDetailScreen.tsx +++ b/apps/mobile/src/features/threads/ThreadDetailScreen.tsx @@ -568,6 +568,13 @@ export const ThreadDetailScreen = memo(function ThreadDetailScreen(props: Thread Alert.alert("Invalid Goal command", goalCommand.message); return null; } + if (props.selectedThread.session === null) { + Alert.alert( + "Start the Codex thread first", + "Send a message before managing its native Goal.", + ); + return null; + } const target = { environmentId: props.environmentId, input: { threadId: props.selectedThread.id }, diff --git a/apps/web/src/components/ChatView.tsx b/apps/web/src/components/ChatView.tsx index cf7622c156ca..f074564c91b9 100644 --- a/apps/web/src/components/ChatView.tsx +++ b/apps/web/src/components/ChatView.tsx @@ -5374,7 +5374,7 @@ function ChatViewContent(props: ChatViewProps) { ); return; } - if (!isServerThread || activeThreadId === null) { + if (!isServerThread || activeThreadId === null || activeThread.session === null) { toastManager.add( stackedThreadToast({ type: "warning", From fab3ab65dbdeeb42d0c09f0d077774756745ec47 Mon Sep 17 00:00:00 2001 From: Fredrik Ekman Date: Sun, 23 Aug 2026 16:34:20 +0200 Subject: [PATCH 4/7] fix(clients): surface running Goal commands --- .../features/threads/ThreadDetailScreen.tsx | 1 + apps/web/src/components/ChatView.tsx | 22 ++++++++++++++++--- 2 files changed, 20 insertions(+), 3 deletions(-) diff --git a/apps/mobile/src/features/threads/ThreadDetailScreen.tsx b/apps/mobile/src/features/threads/ThreadDetailScreen.tsx index df5a860d6c9d..081efa766ed6 100644 --- a/apps/mobile/src/features/threads/ThreadDetailScreen.tsx +++ b/apps/mobile/src/features/threads/ThreadDetailScreen.tsx @@ -664,6 +664,7 @@ export const ThreadDetailScreen = memo(function ThreadDetailScreen(props: Thread props.onChangeDraftMessage, props.selectedThread.id, props.selectedThread.latestTurn, + props.selectedThread.session, props.selectedThreadQueueCount, selectedThreadFeed, selectedThreadKey, diff --git a/apps/web/src/components/ChatView.tsx b/apps/web/src/components/ChatView.tsx index f074564c91b9..e3c31524717f 100644 --- a/apps/web/src/components/ChatView.tsx +++ b/apps/web/src/components/ChatView.tsx @@ -1422,6 +1422,10 @@ function ChatViewContent(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< @@ -5396,6 +5400,11 @@ function ChatViewContent(props: ChatViewProps) { 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); @@ -5458,6 +5467,11 @@ function ChatViewContent(props: ChatViewProps) { return; } finally { goalCommandsInFlightRef.current.delete(submittedGoalCommandThreadKey); + setGoalCommandThreadKeysInFlight((current) => { + const next = new Set(current); + next.delete(submittedGoalCommandThreadKey); + return next; + }); } } if (!directAnnotation && showPlanFollowUpPrompt && activeProposedPlan) { @@ -6982,9 +6996,11 @@ function ChatViewContent(props: ChatViewProps) { sendDisabledReason={ feedbackUploading ? "Sending feedback" - : threadDetailLoading - ? "Messages loading" - : null + : goalCommandRunning + ? "Running Goal command" + : threadDetailLoading + ? "Messages loading" + : null } isPreparingWorktree={isPreparingWorktree} externalDrawerAttached={externalComposerDrawerAttached} From 1e5b206245bd3c2cd74c96ea08e546f0efb55f61 Mon Sep 17 00:00:00 2001 From: Fredrik Ekman Date: Sun, 23 Aug 2026 17:31:05 +0200 Subject: [PATCH 5/7] fix(server): bound Goal event buffering --- apps/server/src/server.test.ts | 31 ++++++++++++++++++++++++++++--- apps/server/src/ws.ts | 2 +- 2 files changed, 29 insertions(+), 4 deletions(-) diff --git a/apps/server/src/server.test.ts b/apps/server/src/server.test.ts index fee486087559..21dacf90d47e 100644 --- a/apps/server/src/server.test.ts +++ b/apps/server/src/server.test.ts @@ -4801,11 +4801,14 @@ it.layer(NodeServices.layer)("server router seam", (it) => { }); const cleared = yield* client[WS_METHODS.codexGoalClear]({ threadId }); const snapshotSeen = yield* Deferred.make(); + const updatedSeen = yield* Deferred.make(); const streamed = yield* client[WS_METHODS.subscribeCodexGoal]({ threadId }).pipe( Stream.tap((event) => event.type === "snapshot" ? Deferred.succeed(snapshotSeen, undefined).pipe(Effect.ignore) - : Effect.void, + : event.type === "updated" + ? Deferred.succeed(updatedSeen, undefined).pipe(Effect.ignore) + : Effect.void, ), Stream.take(3), Stream.runCollect, @@ -4820,6 +4823,7 @@ it.layer(NodeServices.layer)("server router seam", (it) => { threadId, payload: { goal: steeredGoal }, }); + yield* Deferred.await(updatedSeen); yield* PubSub.publish(events, { type: "thread.goal.cleared", eventId: EventId.make("goal-cleared-event"), @@ -4878,7 +4882,7 @@ it.layer(NodeServices.layer)("server router seam", (it) => { }).pipe(Effect.provide(NodeHttpServer.layerTest)), ); - it.effect("subscribeCodexGoal delivers goal updates published while the snapshot loads", () => + it.effect("subscribeCodexGoal buffers the latest update during snapshot loading", () => Effect.gen(function* () { const threadId = ThreadId.make("goal-race-thread"); const events = yield* PubSub.unbounded({ replay: 1 }); @@ -4899,6 +4903,11 @@ it.layer(NodeServices.layer)("server router seam", (it) => { objective: "Steered Goal", updatedAt: 1_777_000_020, }; + const finalGoal = { + ...initialGoal, + objective: "Final Goal", + updatedAt: 1_777_000_040, + }; yield* buildAppUnderTest({ layers: { @@ -4938,6 +4947,22 @@ it.layer(NodeServices.layer)("server router seam", (it) => { threadId, payload: { goal: steeredGoal }, }); + yield* PubSub.publish(events, { + type: "thread.goal.cleared", + eventId: EventId.make("goal-race-cleared-event"), + provider: ProviderDriverKind.make("codex"), + createdAt: "2026-01-01T00:00:01.000Z", + threadId, + payload: {}, + }); + yield* PubSub.publish(events, { + type: "thread.goal.updated", + eventId: EventId.make("goal-race-final-event"), + provider: ProviderDriverKind.make("codex"), + createdAt: "2026-01-01T00:00:02.000Z", + threadId, + payload: { goal: finalGoal }, + }); yield* Deferred.succeed(releaseSnapshot, undefined); return yield* Fiber.join(collected); }), @@ -4952,7 +4977,7 @@ it.layer(NodeServices.layer)("server router seam", (it) => { } assert.equal(updated?.type, "updated"); if (updated?.type === "updated") { - assert.equal(updated.goal.objective, "Steered Goal"); + assert.equal(updated.goal.objective, "Final Goal"); } }).pipe(Effect.provide(NodeHttpServer.layerTest)), ); diff --git a/apps/server/src/ws.ts b/apps/server/src/ws.ts index 19da5f40cd7d..644f44927550 100644 --- a/apps/server/src/ws.ts +++ b/apps/server/src/ws.ts @@ -2268,7 +2268,7 @@ const makeWsRpcLayer = ( return Result.failVoid; }), ), - { capacity: "unbounded" }, + { capacity: 1, strategy: "sliding" }, ); const goal = yield* providerService .getCodexGoal(input.threadId, { allowRecovery: false }) From 987657ad01fc20aa34d8570241437702190bd7d7 Mon Sep 17 00:00:00 2001 From: Fredrik Ekman Date: Mon, 24 Aug 2026 06:59:35 +0200 Subject: [PATCH 6/7] fix(clients): defer Goal subscription until session --- .../mobile/src/features/threads/ThreadDetailScreen.tsx | 6 ++++-- apps/web/src/components/ChatView.tsx | 10 ++++++++-- 2 files changed, 12 insertions(+), 4 deletions(-) diff --git a/apps/mobile/src/features/threads/ThreadDetailScreen.tsx b/apps/mobile/src/features/threads/ThreadDetailScreen.tsx index 081efa766ed6..9cd5cf06539f 100644 --- a/apps/mobile/src/features/threads/ThreadDetailScreen.tsx +++ b/apps/mobile/src/features/threads/ThreadDetailScreen.tsx @@ -470,9 +470,11 @@ export const ThreadDetailScreen = memo(function ThreadDetailScreen(props: Thread const selectedProvider = props.serverConfig?.providers.find( (provider) => provider.instanceId === selectedInstanceId, ); + const hasCodexGoalSession = + selectedProvider?.driver === "codex" && props.selectedThread.session !== null; const codexGoal = useCodexGoal( - selectedProvider?.driver === "codex" ? props.environmentId : null, - selectedProvider?.driver === "codex" ? props.selectedThread.id : null, + hasCodexGoalSession ? props.environmentId : null, + hasCodexGoalSession ? props.selectedThread.id : null, ); useStreamingHaptics(props.selectedThread.id, props.selectedThreadFeed); const selectedProviderSkills = useMemo( diff --git a/apps/web/src/components/ChatView.tsx b/apps/web/src/components/ChatView.tsx index e3c31524717f..123c869209bc 100644 --- a/apps/web/src/components/ChatView.tsx +++ b/apps/web/src/components/ChatView.tsx @@ -2315,9 +2315,15 @@ function ChatViewContent(props: ChatViewProps) { selectedProviderByThreadId ?? threadProvider, ); const selectedProvider: ProviderDriverKind = lockedProvider ?? unlockedSelectedProvider; + const hasCodexGoalSession = + isServerThread && + selectedProvider === "codex" && + activeThread !== null && + activeThread !== undefined && + activeThread.session !== null; const codexGoal = useCodexGoal( - isServerThread && selectedProvider === "codex" ? environmentId : null, - isServerThread && selectedProvider === "codex" ? activeThreadId : null, + hasCodexGoalSession ? environmentId : null, + hasCodexGoalSession ? activeThreadId : null, ); const phase = derivePhase(activeThread?.session ?? null); const threadActivities = activeThread?.activities ?? EMPTY_ACTIVITIES; From e46939314bc392b703c9417db53dac24f588103c Mon Sep 17 00:00:00 2001 From: Fredrik Ekman Date: Mon, 24 Aug 2026 09:20:00 +0200 Subject: [PATCH 7/7] fix(clients): refresh Goal after session resume --- .../mobile/src/features/threads/ThreadDetailScreen.tsx | 10 ++++++---- apps/web/src/components/ChatView.tsx | 9 +++++---- 2 files changed, 11 insertions(+), 8 deletions(-) diff --git a/apps/mobile/src/features/threads/ThreadDetailScreen.tsx b/apps/mobile/src/features/threads/ThreadDetailScreen.tsx index 9cd5cf06539f..f05dd86bb520 100644 --- a/apps/mobile/src/features/threads/ThreadDetailScreen.tsx +++ b/apps/mobile/src/features/threads/ThreadDetailScreen.tsx @@ -470,11 +470,13 @@ export const ThreadDetailScreen = memo(function ThreadDetailScreen(props: Thread const selectedProvider = props.serverConfig?.providers.find( (provider) => provider.instanceId === selectedInstanceId, ); - const hasCodexGoalSession = - selectedProvider?.driver === "codex" && props.selectedThread.session !== null; + const hasActiveCodexGoalSession = + selectedProvider?.driver === "codex" && + props.selectedThread.session !== null && + props.selectedThread.session.status !== "stopped"; const codexGoal = useCodexGoal( - hasCodexGoalSession ? props.environmentId : null, - hasCodexGoalSession ? props.selectedThread.id : null, + hasActiveCodexGoalSession ? props.environmentId : null, + hasActiveCodexGoalSession ? props.selectedThread.id : null, ); useStreamingHaptics(props.selectedThread.id, props.selectedThreadFeed); const selectedProviderSkills = useMemo( diff --git a/apps/web/src/components/ChatView.tsx b/apps/web/src/components/ChatView.tsx index 123c869209bc..a1fb60687ecc 100644 --- a/apps/web/src/components/ChatView.tsx +++ b/apps/web/src/components/ChatView.tsx @@ -2315,15 +2315,16 @@ function ChatViewContent(props: ChatViewProps) { selectedProviderByThreadId ?? threadProvider, ); const selectedProvider: ProviderDriverKind = lockedProvider ?? unlockedSelectedProvider; - const hasCodexGoalSession = + const hasActiveCodexGoalSession = isServerThread && selectedProvider === "codex" && activeThread !== null && activeThread !== undefined && - activeThread.session !== null; + activeThread.session !== null && + activeThread.session.status !== "stopped"; const codexGoal = useCodexGoal( - hasCodexGoalSession ? environmentId : null, - hasCodexGoalSession ? activeThreadId : null, + hasActiveCodexGoalSession ? environmentId : null, + hasActiveCodexGoalSession ? activeThreadId : null, ); const phase = derivePhase(activeThread?.session ?? null); const threadActivities = activeThread?.activities ?? EMPTY_ACTIVITIES;