From 6a1c8d1e34e8b3e74111072bdc1c841e5530c76e Mon Sep 17 00:00:00 2001 From: Julius Marminge Date: Mon, 26 Oct 2026 03:21:08 -0700 Subject: [PATCH 01/13] fix(mobile): mobile:install no longer requires --prepare The --prepare flag from #25390 had no default, so effect/cli made it required and every mobile:install failed with "Missing required flag". Co-Authored-By: Claude Opus 5.5 --- scripts/mobile-native-client.ts | 1 + 1 file changed, 1 insertion(+) diff --git a/scripts/mobile-native-client.ts b/scripts/mobile-native-client.ts index 99f9997db7c0a..74e914312b618 100644 --- a/scripts/mobile-native-client.ts +++ b/scripts/mobile-native-client.ts @@ -1048,6 +1048,7 @@ const main = Command.make( Flag.withDescription( "Fetch or build the client for this checkout without a device (Android: for --abi). Run it before leasing a device: a build can outlast the lease.", ), + Flag.withDefault(false), ), }, Effect.fn("nativeClient.main")(function* ({ From 3e47c1cabc463145e485f1ee02e092832cd45c34 Mon Sep 17 00:00:00 2001 From: Julius Marminge Date: Mon, 26 Oct 2026 03:26:33 -0700 Subject: [PATCH 02/13] fix(mobile): e2e landscape poses rotate for real on tablets and folds agent-device names rotations relative to the device's natural orientation, which is landscape on the Pixel Tablet and an unfolded Fold, so "landscape-left" turned them portrait. The runner now asks for a shape and tries each rotation until the screen has it. It also prepares each platform's dev client before leasing, rides out longer worker drops, and skips the Duo poses by default until agent-device can fold Simlock's simulators. Co-Authored-By: Claude Opus 5.5 --- scripts/mobile-e2e.ts | 16 +++++++++++++--- scripts/mobile-e2e/device.ts | 20 ++++++++++++++++++-- scripts/mobile-e2e/journey.ts | 9 +++------ scripts/mobile-e2e/poses.ts | 6 ++++++ 4 files changed, 40 insertions(+), 11 deletions(-) diff --git a/scripts/mobile-e2e.ts b/scripts/mobile-e2e.ts index dc9af4a5e3b2e..9b3466dd2a797 100644 --- a/scripts/mobile-e2e.ts +++ b/scripts/mobile-e2e.ts @@ -139,7 +139,10 @@ async function runPose( async function main() { const args = NodeProcess.argv.slice(2); if (args.includes("--list")) { - for (const pose of POSES) NodeProcess.stdout.write(`${pose.id}\t${pose.device}\n`); + for (const pose of POSES) + NodeProcess.stdout.write( + `${pose.id}\t${pose.device}${pose.pending ? `\t(pending: ${pose.pending})` : ""}\n`, + ); return; } const values = (flag: string) => @@ -150,7 +153,7 @@ async function main() { const platforms = values("--platform"); const poses = POSES.filter( (pose) => - (selected.length === 0 || selected.includes(pose.id)) && + (selected.length === 0 ? !pose.pending : selected.includes(pose.id)) && (platforms.length === 0 || platforms.includes(pose.platform)), ); const concurrencyFlag = values("--concurrency")[0]; @@ -221,8 +224,15 @@ async function main() { }, ); await waitForPort(metroPort, "Metro", 120_000); - // Warm each platform's bundle once, instead of on every device at once. + // Warm each platform's bundle once, instead of on every device at once, and have its dev + // client cached before any lease: a cache miss builds, which can outlast a lease. for (const platform of new Set(poses.map((pose) => pose.platform))) { + log("e2e", `preparing the ${platform} dev client`); + await run( + NodeProcess.execPath, + ["scripts/mobile-native-client.ts", "ensure", platform, "--prepare"], + { cwd: REPO_ROOT, timeoutMs: 1_800_000 }, + ); log("e2e", `bundling for ${platform}`); await run("curl", [ "-sf", diff --git a/scripts/mobile-e2e/device.ts b/scripts/mobile-e2e/device.ts index a2638b10d9a40..5a8175eab22e4 100644 --- a/scripts/mobile-e2e/device.ts +++ b/scripts/mobile-e2e/device.ts @@ -105,9 +105,9 @@ export class Device { try { return await this.agentOnce(args, options); } catch (error) { - if (attempt >= 3 || !(error instanceof AgentDeviceError) || !TRANSIENT.test(error.message)) + if (attempt >= 5 || !(error instanceof AgentDeviceError) || !TRANSIENT.test(error.message)) throw error; - await delay(3_000 * attempt); + await delay(5_000 * attempt); } } } @@ -174,6 +174,22 @@ export class Device { await this.acceptOpenPrompt(); } + /** + * agent-device names rotations relative to the device's natural orientation, which is + * landscape on tablets and unfolded folds, so try each name until the screen has the shape. + */ + async orient(shape: "landscape" | "portrait"): Promise { + let snapshot = await this.snapshot(); + for (const name of ["landscape-left", "portrait", "landscape-right"]) { + const { width, height } = snapshot.viewport; + if (width > height === (shape === "landscape")) return snapshot; + await this.agent(["orientation", name]); + await delay(2_000); + snapshot = await this.snapshot(); + } + return snapshot; + } + async openUrl(url: string) { await this.agent(["open", BUNDLE_ID, url]); await this.acceptOpenPrompt(); diff --git a/scripts/mobile-e2e/journey.ts b/scripts/mobile-e2e/journey.ts index 49e6dfb90fd4b..f3442d0a8c950 100644 --- a/scripts/mobile-e2e/journey.ts +++ b/scripts/mobile-e2e/journey.ts @@ -137,9 +137,8 @@ export async function runJourney( } if (!threads) throw new Error("the app did not show the fixture threads after pairing"); // Rotated only once the app is up: a rotation during launch is lost. - if (device.pose.landscape) await device.agent(["orientation", "landscape-left"]); - await delay(device.pose.landscape ? 2_000 : 1_000); - threads = await device.snapshot(); + await delay(1_000); + threads = await device.orient(device.pose.landscape ? "landscape" : "portrait"); const tabs = tabBarRect(threads); const wide = threads.viewport.width > threads.viewport.height; await capture("threads", threads, [ @@ -270,9 +269,7 @@ export async function runJourney( await capture("settings", await device.snapshot()); if (device.pose.form === "tablet" && !device.pose.fold) { - await device.agent(["orientation", device.pose.landscape ? "portrait" : "landscape-left"]); - await delay(2_000); - await capture("rotated", await device.snapshot()); + await capture("rotated", await device.orient(device.pose.landscape ? "portrait" : "landscape")); } if (device.pose.fold) { await device.agent(["fold", device.pose.fold === "open" ? "closed" : "open"]); diff --git a/scripts/mobile-e2e/poses.ts b/scripts/mobile-e2e/poses.ts index 0586d08e9eeef..9cc42b40a2609 100644 --- a/scripts/mobile-e2e/poses.ts +++ b/scripts/mobile-e2e/poses.ts @@ -9,6 +9,8 @@ export interface Pose { readonly landscape?: boolean; readonly fold?: "closed" | "open"; readonly textSize?: "accessibility-large"; + /** Why the pose is skipped unless named with --pose. */ + readonly pending?: string; } const pose = (value: Pose) => value; @@ -35,6 +37,8 @@ export const POSES: ReadonlyArray = [ form: "phone", appearance: "light", fold: "closed", + pending: + "agent-device fold refuses simulators outside the default device set, which Simlock uses", }), pose({ id: "iphone-duo-unfolded", @@ -43,6 +47,8 @@ export const POSES: ReadonlyArray = [ form: "tablet", appearance: "dark", fold: "open", + pending: + "agent-device fold refuses simulators outside the default device set, which Simlock uses", }), pose({ id: "ipad-13-portrait", From cc0b86ddf12a34bd65dce88d6f2ae1f069ef841e Mon Sep 17 00:00:00 2001 From: Theo Browne Date: Mon, 26 Oct 2026 03:28:17 -0700 Subject: [PATCH 03/13] fix(server): goals stop the running turn, clear while working, no echoes (#25426) Co-authored-by: Claude Opus 5.5 --- .../Adapters/ClaudeAdapterV2.test.ts | 49 ++++ .../Adapters/ClaudeAdapterV2.ts | 23 +- .../Adapters/CodexAdapterV2.test.ts | 230 +++++++++++++++++ .../Adapters/CodexAdapterV2.ts | 168 ++++++++----- .../src/orchestration-v2/CommandPolicy.ts | 22 ++ .../src/orchestration-v2/Orchestrator.ts | 232 ++++++++++++++++-- .../ProviderTurnStartService.ts | 21 +- .../SteeringCompletion.integration.test.ts | 217 +++++++++++++++- .../components/chat/ThreadActivityBanners.tsx | 5 +- docs/user/composer.md | 6 +- packages/contracts/src/orchestrationV2.ts | 31 +++ 11 files changed, 913 insertions(+), 91 deletions(-) diff --git a/apps/server/src/orchestration-v2/Adapters/ClaudeAdapterV2.test.ts b/apps/server/src/orchestration-v2/Adapters/ClaudeAdapterV2.test.ts index 0c61165924917..7fece9817ce01 100644 --- a/apps/server/src/orchestration-v2/Adapters/ClaudeAdapterV2.test.ts +++ b/apps/server/src/orchestration-v2/Adapters/ClaudeAdapterV2.test.ts @@ -14524,6 +14524,55 @@ describe("ClaudeAdapterV2 background wake turns", () => { ), ), ); + + it.effect("keeps the CLI's goal confirmation out of the conversation", () => + Effect.gen(function* () { + const harness = yield* makeWakeHarness; + const condition = "all tests pass"; + yield* harness.runtime.startTurn( + makeClaudeTestTurnInput({ + threadId: harness.threadId, + providerThread: { + ...harness.providerThread, + goal: { objective: condition, status: "active", checks: 1 }, + }, + now: yield* DateTime.now, + attemptId: RunAttemptId.make("goal-clear-attempt"), + text: "/goal clear", + attachments: [], + }), + ); + yield* Queue.offer( + harness.sdkMessages, + syntheticFrame("goal-clear", `Goal cleared: ${condition}`), + ); + yield* Queue.offer( + harness.sdkMessages, + makeResultFrame({ + uuid: "goal-clear-result", + result: `Goal cleared: ${condition}\n`, + numTurns: 0, + }), + ); + yield* Queue.take(harness.terminalReceipts); + assert.isTrue( + harness.events.some( + (event) => + event.type === "provider_thread.updated" && event.providerThread.goal === null, + ), + ); + assert.isFalse( + harness.events.some( + (event) => event.type === "message.updated" && event.message.role === "assistant", + ), + ); + }).pipe( + Effect.scoped, + Effect.provide( + Layer.mergeAll(IdAllocator.layer, McpProviderSessions.layer, NodeServices.layer), + ), + ), + ); }); }); diff --git a/apps/server/src/orchestration-v2/Adapters/ClaudeAdapterV2.ts b/apps/server/src/orchestration-v2/Adapters/ClaudeAdapterV2.ts index 1a9bf4a2bbbe0..baf9a0b0a690b 100644 --- a/apps/server/src/orchestration-v2/Adapters/ClaudeAdapterV2.ts +++ b/apps/server/src/orchestration-v2/Adapters/ClaudeAdapterV2.ts @@ -1723,6 +1723,19 @@ function nextClaudeGoal( }; } +/** The CLI confirms a goal change as a synthetic reply; the goal row shows that change instead. */ +function isClaudeGoalAcknowledgement(message: SDKMessage): boolean { + if ( + message.type !== "assistant" || + message.parent_tool_use_id !== null || + message.message.model !== "" + ) { + return false; + } + const text = textFromClaudeContent(message.message.content).trim(); + return text.startsWith(CLAUDE_GOAL_SET_PREFIX) || text.startsWith("Goal cleared: "); +} + const providerGoalsEqual = Schema.toEquivalence(Schema.NullOr(OrchestrationV2ProviderGoal)); function normalizeClaudeResultText(text: string): string { @@ -8895,6 +8908,13 @@ export const makeClaudeAdapterV2 = Effect.fn("makeClaudeAdapterV2")(function* ( }); return; } + if (isClaudeGoalAcknowledgement(message)) { + // The CLI repeats a local command's output as the turn's result text. + if (assistantText !== null) { + context.assistant.noticeTexts.add(normalizeClaudeResultText(assistantText.text)); + } + return; + } // Claude Code echoes /compact's stdout ("Not enough messages to compact.") as a // synthetic assistant message. The compaction item reports that outcome; it is not a // reply. @@ -8967,7 +8987,8 @@ export const makeClaudeAdapterV2 = Effect.fn("makeClaudeAdapterV2")(function* ( context.assistant.fallbackText.length === 0 && resultText !== null && resultText.text.length > 0 && - !context.assistant.noticeTexts.has(resultText.text) + !context.assistant.noticeTexts.has(resultText.text) && + !context.assistant.noticeTexts.has(normalizeClaudeResultText(resultText.text)) ) { context.assistant.fallbackText = resultText.text; context.assistant.fallbackNativeItemId = resultText.nativeItemId; diff --git a/apps/server/src/orchestration-v2/Adapters/CodexAdapterV2.test.ts b/apps/server/src/orchestration-v2/Adapters/CodexAdapterV2.test.ts index bd7b6f4208763..602e1a4d56f29 100644 --- a/apps/server/src/orchestration-v2/Adapters/CodexAdapterV2.test.ts +++ b/apps/server/src/orchestration-v2/Adapters/CodexAdapterV2.test.ts @@ -5958,6 +5958,7 @@ describe("CodexAdapterV2 post-settle continuation", () => { nativeThreadId, nativeTurnId: "goal-turn-1", prompt: objective, + sentPrompt: `Start working toward this goal: ${objective}`, }).slice(5), 5, ), @@ -6022,6 +6023,8 @@ describe("CodexAdapterV2 post-settle continuation", () => { [providerTurnId("goal-turn-2"), RunAttemptId.make("goal-attempt"), 2], ], ); + // The goal is set paused until its first turn starts; only the result shows. + assert.notInclude(providerGoals(harness.events), "paused"); assert.equal(providerGoals(harness.events).at(-1), "complete"); }).pipe( Effect.scoped, @@ -6204,6 +6207,233 @@ describe("CodexAdapterV2 post-settle continuation", () => { ), ); + it.effect("pauses a goal Codex does not continue before its run settles", () => + Effect.gen(function* () { + const prompt = "Keep going"; + const transcript = makeCodexReplayTranscript({ + scenario: "goal-continuation-expired", + entries: [ + ...codexReplayPreamble({ nativeThreadId, nativeTurnId: "goal-turn", prompt }), + notification("goal active", "thread/goal/updated", { + threadId: nativeThreadId, + turnId: "goal-turn", + goal: codexGoal("active", 10), + }), + notification("turn done", "turn/completed", { + threadId: nativeThreadId, + turn: makeCodexReplayTurn({ id: "goal-turn", status: "completed" }), + }), + notification("goal accounted", "thread/goal/updated", { + threadId: nativeThreadId, + turnId: null, + goal: codexGoal("active", 11), + }), + ...request( + 4, + "thread/goal/set", + { threadId: nativeThreadId, status: "paused" }, + { goal: codexGoal("paused", 11) }, + ), + ], + }); + const harness = yield* makeCodexReplayHarness(transcript); + yield* harness.runtime.startTurn( + makeCodexTestTurnInput({ + threadId: harness.threadId, + providerThread: harness.providerThread, + now: yield* DateTime.now, + attemptId: RunAttemptId.make("goal-expired-attempt"), + text: prompt, + }), + ); + yield* harness.until(() => + harness.events.some( + (event) => + event.type === "provider_thread.updated" && + event.providerThread.goal?.tokensUsed === 11, + ), + ); + assert.lengthOf(harness.terminalEvents(), 0); + yield* TestClock.adjust("5 seconds"); + yield* harness.firstTerminal; + + assert.deepEqual(providerGoals(harness.events), ["active", "active", "paused"]); + assert.deepEqual( + harness.terminalEvents().map((event) => event.status), + ["completed"], + ); + }).pipe( + Effect.scoped, + Effect.provide( + Layer.mergeAll(IdAllocator.layer, McpProviderSessions.layer, NodeServices.layer), + ), + ), + ); + + it.effect("stops a turn Codex starts while the expiry pause is in flight", () => + Effect.gen(function* () { + const prompt = "Keep going"; + const transcript = makeCodexReplayTranscript({ + scenario: "goal-continuation-during-expiry", + entries: [ + ...codexReplayPreamble({ nativeThreadId, nativeTurnId: "goal-turn", prompt }), + notification("goal active", "thread/goal/updated", { + threadId: nativeThreadId, + turnId: "goal-turn", + goal: codexGoal("active", 10), + }), + notification("turn done", "turn/completed", { + threadId: nativeThreadId, + turn: makeCodexReplayTurn({ id: "goal-turn", status: "completed" }), + }), + notification("goal accounted", "thread/goal/updated", { + threadId: nativeThreadId, + turnId: null, + goal: codexGoal("active", 11), + }), + { + type: "expect_outbound", + label: "pause", + frame: { + id: 4, + method: "thread/goal/set", + params: { threadId: nativeThreadId, status: "paused" }, + }, + }, + notification("goal paused", "thread/goal/updated", { + threadId: nativeThreadId, + turnId: null, + goal: codexGoal("paused", 11), + }), + notification("late continuation", "turn/started", { + threadId: nativeThreadId, + turn: makeCodexReplayTurn({ id: "goal-turn-2", status: "inProgress" }), + }), + ...request( + 5, + "thread/goal/set", + { threadId: nativeThreadId, status: "paused" }, + { goal: codexGoal("paused", 11) }, + ), + ...request( + 6, + "turn/interrupt", + { threadId: nativeThreadId, turnId: "goal-turn-2" }, + {}, + ), + { + type: "emit_inbound", + label: "paused", + frame: { id: 4, result: { goal: codexGoal("paused", 11) } }, + }, + ], + }); + const harness = yield* makeCodexReplayHarness(transcript); + yield* harness.runtime.startTurn( + makeCodexTestTurnInput({ + threadId: harness.threadId, + providerThread: harness.providerThread, + now: yield* DateTime.now, + attemptId: RunAttemptId.make("goal-race-attempt"), + text: prompt, + }), + ); + yield* harness.until(() => + harness.events.some( + (event) => + event.type === "provider_thread.updated" && + event.providerThread.goal?.tokensUsed === 11, + ), + ); + yield* TestClock.adjust("5 seconds"); + yield* harness.firstTerminal; + + assert.deepEqual( + harness.terminalEvents().map((event) => [event.providerTurnId, event.status]), + [ + [ + (yield* IdAllocator.IdAllocatorV2).derive.providerTurn({ + driver: CodexAdapterV2.CODEX_DRIVER_KIND, + nativeTurnId: "goal-turn", + }), + "completed", + ], + ], + ); + assert.equal(providerGoals(harness.events).at(-1), "paused"); + }).pipe( + Effect.scoped, + Effect.provide( + Layer.mergeAll(IdAllocator.layer, McpProviderSessions.layer, NodeServices.layer), + ), + ), + ); + + it.effect("settles the run when Codex never answers the expiry pause", () => + Effect.gen(function* () { + const prompt = "Keep going"; + const transcript = makeCodexReplayTranscript({ + scenario: "goal-expiry-pause-unanswered", + entries: [ + ...codexReplayPreamble({ nativeThreadId, nativeTurnId: "goal-turn", prompt }), + notification("goal active", "thread/goal/updated", { + threadId: nativeThreadId, + turnId: "goal-turn", + goal: codexGoal("active", 10), + }), + notification("turn done", "turn/completed", { + threadId: nativeThreadId, + turn: makeCodexReplayTurn({ id: "goal-turn", status: "completed" }), + }), + notification("goal accounted", "thread/goal/updated", { + threadId: nativeThreadId, + turnId: null, + goal: codexGoal("active", 11), + }), + { + type: "expect_outbound", + label: "pause", + frame: { + id: 4, + method: "thread/goal/set", + params: { threadId: nativeThreadId, status: "paused" }, + }, + }, + ], + }); + const harness = yield* makeCodexReplayHarness(transcript); + yield* harness.runtime.startTurn( + makeCodexTestTurnInput({ + threadId: harness.threadId, + providerThread: harness.providerThread, + now: yield* DateTime.now, + attemptId: RunAttemptId.make("goal-unanswered-attempt"), + text: prompt, + }), + ); + yield* harness.until(() => + harness.events.some( + (event) => + event.type === "provider_thread.updated" && + event.providerThread.goal?.tokensUsed === 11, + ), + ); + yield* TestClock.adjust("5 seconds"); + yield* TestClock.adjust("10 seconds"); + yield* harness.firstTerminal; + + assert.deepEqual( + harness.terminalEvents().map((event) => event.status), + ["completed"], + ); + }).pipe( + Effect.scoped, + Effect.provide( + Layer.mergeAll(IdAllocator.layer, McpProviderSessions.layer, NodeServices.layer), + ), + ), + ); + it.effect("pauses an active goal before Stop interrupts its turn", () => Effect.gen(function* () { const prompt = "Keep going"; diff --git a/apps/server/src/orchestration-v2/Adapters/CodexAdapterV2.ts b/apps/server/src/orchestration-v2/Adapters/CodexAdapterV2.ts index c98a391de6812..6dba639b63b01 100644 --- a/apps/server/src/orchestration-v2/Adapters/CodexAdapterV2.ts +++ b/apps/server/src/orchestration-v2/Adapters/CodexAdapterV2.ts @@ -31,7 +31,9 @@ import { CodexSettings, defaultInstanceIdForDriver, isOrchestrationV2WorkActive, + type OrchestrationV2GoalCommand, OrchestrationV2ProviderGoal, + parseGoalCommand, ProviderDriverKind, type ProviderSetupError, } from "@t3tools/contracts"; @@ -882,32 +884,6 @@ const isTerminalProviderTurn = (turn: OrchestrationV2ProviderTurn): boolean => turn.status === "failed" || turn.status === "cancelled"); -export type CodexGoalCommand = - | { readonly type: "show" } - | { readonly type: "clear" } - | { readonly type: "pause" } - | { readonly type: "resume" } - | { readonly type: "set"; readonly objective: string }; - -function parseCodexGoalCommand(text: string): CodexGoalCommand | null { - const match = /^\/goal(?:\s+([\s\S]*))?$/u.exec(text.trim()); - if (match === null) return null; - const argument = (match[1] ?? "").trim(); - switch (argument.toLowerCase()) { - case "": - case "edit": - return { type: "show" }; - case "clear": - return { type: "clear" }; - case "pause": - return { type: "pause" }; - case "resume": - return { type: "resume" }; - default: - return { type: "set", objective: argument }; - } -} - type CodexThreadGoal = CodexSchema.V2ThreadGoalUpdatedNotification["goal"]; const CODEX_GOAL_STATUSES = { @@ -1849,9 +1825,17 @@ export const makeCodexAdapterV2 = Effect.fn("makeCodexAdapterV2")(function* ( const emitProviderEvent = (event: ProviderAdapter.ProviderAdapterV2Event) => Queue.offer(events, event).pipe(Effect.asVoid); + // Setting a goal passes through cleared and paused states; publish only the result. + const goalUpdatesHeld = new Set(); const emitGoalUpdate = Effect.fnUntraced(function* (nativeThreadId: string) { const providerThread = rootProviderThreads.get(nativeThreadId); - if (providerThread === undefined || !goalsByNativeThread.has(nativeThreadId)) return; + if ( + providerThread === undefined || + !goalsByNativeThread.has(nativeThreadId) || + goalUpdatesHeld.has(nativeThreadId) + ) { + return; + } const goal = goalsByNativeThread.get(nativeThreadId) ?? null; if (providerGoalsEqual(providerThread.goal ?? null, goal)) return; const updated = { ...providerThread, goal, updatedAt: yield* DateTime.now }; @@ -1872,6 +1856,12 @@ export const makeCodexAdapterV2 = Effect.fn("makeCodexAdapterV2")(function* ( yield* emitGoalUpdate(nativeThreadId); }); + const releaseGoalUpdates = (nativeThreadId: string) => + Effect.suspend(() => { + goalUpdatesHeld.delete(nativeThreadId); + return emitGoalUpdate(nativeThreadId); + }); + // Only new model output proves recovery; local item/completed can arrive during upstream retries. const completeProviderRetry = Effect.fn("CodexAdapterV2.completeProviderRetry")(function* ( context: ActiveCodexTurnContext, @@ -5695,10 +5685,12 @@ export const makeCodexAdapterV2 = Effect.fn("makeCodexAdapterV2")(function* ( next: yield* Deferred.make(), }); yield* Effect.sleep(CODEX_GOAL_CONTINUATION_GRACE).pipe( + Effect.andThen(pauseUncontinuedGoal(nativeThreadId, event)), Effect.andThen( turnTerminalizationPermit.withPermits(1)(releaseGoalHold(nativeThreadId, event)), ), - Effect.forkIn(scope), + // The grace window starts when the turn completes, not when the fiber is scheduled. + Effect.forkIn(scope, { startImmediately: true }), ); return; } @@ -5706,6 +5698,51 @@ export const makeCodexAdapterV2 = Effect.fn("makeCodexAdapterV2")(function* ( }, ); + // A goal must not read as active once its run ends. Claiming the hold first + // makes a turn Codex starts during the pause an orphan, which is stopped. + const pauseUncontinuedGoal = Effect.fnUntraced(function* ( + nativeThreadId: string, + expected: CodexRootTerminalEvent, + ) { + const hold = goalHolds.get(nativeThreadId); + if ( + hold?.event !== expected || + goalStops.has(nativeThreadId) || + goalsByNativeThread.get(nativeThreadId)?.status !== "active" + ) { + return; + } + goalHolds.delete(nativeThreadId); + // Keeps a turn Codex starts before settlement an orphan once the goal reads paused. + goalStops.set(nativeThreadId, (goalStops.get(nativeThreadId) ?? 0) + 1); + yield* Effect.logWarning("orchestration-v2.codex-goal-continuation-expired", { + nativeThreadId, + }); + yield* client + .request("thread/goal/set", { threadId: nativeThreadId, status: "paused" }) + .pipe( + // The hold is already claimed, so an unanswered pause must not keep the run open. + Effect.timeout("10 seconds"), + Effect.flatMap(({ goal }) => { + goalsByNativeThread.set(nativeThreadId, providerGoalFromCodex(goal)); + return emitGoalUpdate(nativeThreadId); + }), + Effect.catch((cause) => + Effect.logWarning("orchestration-v2.codex-goal-pause-failed", { cause }), + ), + Effect.ensuring( + turnTerminalizationPermit.withPermits(1)(settleGoalHold(hold, "completed")), + ), + Effect.ensuring( + Effect.sync(() => { + const stops = (goalStops.get(nativeThreadId) ?? 1) - 1; + if (stops > 0) goalStops.set(nativeThreadId, stops); + else goalStops.delete(nativeThreadId); + }), + ), + ); + }); + /** Hold the terminalization permit and remove the goalHolds entry before calling. */ const settleGoalHold = Effect.fnUntraced(function* ( hold: CodexGoalHold, @@ -6084,9 +6121,10 @@ export const makeCodexAdapterV2 = Effect.fn("makeCodexAdapterV2")(function* ( }), ); + /** Settles a /goal turn that never reaches Codex. State changes reply through the goal row. */ const completeGoalCommandTurn = Effect.fnUntraced(function* ( turnInput: ProviderAdapter.ProviderAdapterV2TurnInput, - reply: string, + reply: string | null, ) { const now = yield* DateTime.now; const nativeId = `goal-command:${turnInput.attemptId}`; @@ -6131,26 +6169,28 @@ export const makeCodexAdapterV2 = Effect.fn("makeCodexAdapterV2")(function* ( providerTurn, }); yield* rememberRootProviderThread(turnInput.providerThread); - const artifacts = yield* buildAgentMessageArtifacts( - context, - { id: nativeId, text: reply }, - true, - ); - yield* emitProviderEvent({ - type: "node.updated", - driver: CODEX_PROVIDER, - node: artifacts.node, - }); - yield* emitProviderEvent({ - type: "message.updated", - driver: CODEX_PROVIDER, - message: artifacts.message, - }); - yield* emitProviderEvent({ - type: "turn_item.updated", - driver: CODEX_PROVIDER, - turnItem: artifacts.turnItem, - }); + if (reply !== null) { + const artifacts = yield* buildAgentMessageArtifacts( + context, + { id: nativeId, text: reply }, + true, + ); + yield* emitProviderEvent({ + type: "node.updated", + driver: CODEX_PROVIDER, + node: artifacts.node, + }); + yield* emitProviderEvent({ + type: "message.updated", + driver: CODEX_PROVIDER, + message: artifacts.message, + }); + yield* emitProviderEvent({ + type: "turn_item.updated", + driver: CODEX_PROVIDER, + turnItem: artifacts.turnItem, + }); + } yield* emitProviderEvent({ type: "provider_turn.updated", driver: CODEX_PROVIDER, @@ -6308,7 +6348,7 @@ export const makeCodexAdapterV2 = Effect.fn("makeCodexAdapterV2")(function* ( const runGoalCommand = Effect.fn("CodexAdapterV2.runGoalCommand")( function* ( turnInput: ProviderAdapter.ProviderAdapterV2TurnInput, - command: CodexGoalCommand, + command: OrchestrationV2GoalCommand, ) { const threadId = yield* getNativeThreadId(turnInput.providerThread); rootProviderThreads.set(threadId, turnInput.providerThread); @@ -6330,10 +6370,7 @@ export const makeCodexAdapterV2 = Effect.fn("makeCodexAdapterV2")(function* ( if (command.type === "clear") { const { cleared } = yield* client.request("thread/goal/clear", { threadId }); goalsByNativeThread.set(threadId, null); - return yield* completeGoalCommandTurn( - turnInput, - cleared ? "Goal cleared." : "No goal to clear.", - ); + return yield* completeGoalCommandTurn(turnInput, cleared ? null : "No goal is set."); } if (current === null && command.type !== "set") { return yield* completeGoalCommandTurn(turnInput, "No goal is set."); @@ -6344,12 +6381,10 @@ export const makeCodexAdapterV2 = Effect.fn("makeCodexAdapterV2")(function* ( status: "paused", }); goalsByNativeThread.set(threadId, providerGoalFromCodex(goal)); - return yield* completeGoalCommandTurn( - turnInput, - "Goal paused. Send /goal resume to continue.", - ); + return yield* completeGoalCommandTurn(turnInput, null); } if (command.type === "set") { + goalUpdatesHeld.add(threadId); if (current !== null) yield* client.request("thread/goal/clear", { threadId }); // Keep paused until our turn starts so Codex cannot start one first. const { goal } = yield* client.request("thread/goal/set", { @@ -6363,12 +6398,16 @@ export const makeCodexAdapterV2 = Effect.fn("makeCodexAdapterV2")(function* ( const activation = { stopped: false }; goalActivations.set(threadId, activation); yield* Effect.gen(function* () { + // A bare objective reads like a statement, so the model echoes it back. yield* startNativeTurn( turnInput, command.type === "set" ? yield* toCodexInput({ ...turnInput, - message: { ...turnInput.message, text: command.objective }, + message: { + ...turnInput.message, + text: `Start working toward this goal: ${command.objective}`, + }, }) : [], ); @@ -6402,6 +6441,14 @@ export const makeCodexAdapterV2 = Effect.fn("makeCodexAdapterV2")(function* ( }, (effect, turnInput) => effect.pipe( + Effect.ensuring( + Effect.suspend(() => { + const threadId = turnInput.providerThread.nativeThreadRef?.nativeId; + return threadId == null || !goalUpdatesHeld.has(threadId) + ? Effect.void + : releaseGoalUpdates(threadId); + }), + ), Effect.mapError( (cause) => new ProviderAdapter.ProviderAdapterTurnStartError({ @@ -6690,9 +6737,8 @@ export const makeCodexAdapterV2 = Effect.fn("makeCodexAdapterV2")(function* ( ), startTurn: (turnInput) => { const goalCommand = - turnInput.message.attachments.length === 0 && turnInput.restartContinuationOfRunId === undefined - ? parseCodexGoalCommand(turnInput.message.text) + ? parseGoalCommand(turnInput.message) : null; if (goalCommand !== null) return runGoalCommand(turnInput, goalCommand); return Effect.gen(function* () { diff --git a/apps/server/src/orchestration-v2/CommandPolicy.ts b/apps/server/src/orchestration-v2/CommandPolicy.ts index 6c07a410a1b36..8a8ef82554f63 100644 --- a/apps/server/src/orchestration-v2/CommandPolicy.ts +++ b/apps/server/src/orchestration-v2/CommandPolicy.ts @@ -4,6 +4,8 @@ import { ModelSelection, type OrchestrationV2Command, OrchestrationV2ProviderCapabilities, + type OrchestrationV2GoalCommand, + type OrchestrationV2ProviderGoalStatus, type OrchestrationV2Run, OrchestrationV2ThreadProjection, ProviderInstanceId, @@ -128,6 +130,26 @@ export function isNativeMaintenanceCommand(message: { ); } +/** + * A goal command sent into active work stops that work when it starts or + * replaces goal work, or ends a goal that is driving the run. + */ +export function goalCommandStopsActiveRun( + command: OrchestrationV2GoalCommand, + goalStatus: OrchestrationV2ProviderGoalStatus | null, +): boolean { + switch (command.type) { + case "set": + case "resume": + return true; + case "clear": + case "pause": + return goalStatus === "active"; + case "show": + return false; + } +} + export function resolveMessageDispatchIntent( projection: OrchestrationV2ThreadProjection, requestedMode: MessageDispatchMode, diff --git a/apps/server/src/orchestration-v2/Orchestrator.ts b/apps/server/src/orchestration-v2/Orchestrator.ts index 1541efe578a67..0c6ca268f1e04 100644 --- a/apps/server/src/orchestration-v2/Orchestrator.ts +++ b/apps/server/src/orchestration-v2/Orchestrator.ts @@ -55,6 +55,7 @@ import { type OrchestrationV2ThreadProjection, type OrchestrationV2TurnItem, latestProviderTurnForAttempt, + parseGoalCommand, orchestrationV2RunWorkStartedAt, ProviderInstanceId, ProviderInteractionMode, @@ -93,6 +94,7 @@ import { import { CheckpointServiceV2 } from "./CheckpointService.ts"; import { CommandPolicyV2, + goalCommandStopsActiveRun, isNativeMaintenanceCommand, resolveMessageDispatchIntent, } from "./CommandPolicy.ts"; @@ -452,13 +454,6 @@ function wakeWorkStartedAt( return previous === undefined ? {} : { workStartedAt: orchestrationV2RunWorkStartedAt(previous) }; } -function isGoalCommand(message: { - readonly text: string; - readonly attachments: ReadonlyArray; -}): boolean { - return message.attachments.length === 0 && /^\/goal(?:\s|$)/u.test(message.text.trim()); -} - const threadPullRequestLinksEqual = Schema.toEquivalence(Schema.NullOr(ThreadLinkedPullRequest)); function commandThreadId(command: OrchestrationV2ServerCommand): ThreadId { @@ -622,10 +617,12 @@ function isBlockingRun(run: OrchestrationV2Run): boolean { } /** Waiting runs only drain post-terminal output; their agent turn is over and still needs a wake. */ +function isLiveRun(run: OrchestrationV2Run): boolean { + return run.status === "preparing" || run.status === "starting" || run.status === "running"; +} + function hasLiveRun(projection: Pick): boolean { - return projection.runs.some( - (run) => run.status === "preparing" || run.status === "starting" || run.status === "running", - ); + return projection.runs.some(isLiveRun); } function withPullRequestWatch( @@ -987,6 +984,35 @@ const makeOrchestrator = Effect.fn("orchestrationV2.Orchestrator.layer")(functio const threadForkService = yield* ThreadForkServiceV2; const threadDispatch = yield* ThreadCommandExecutor.ThreadCommandExecutor; + /** The driver when this provider runs `/goal` natively, else undefined. */ + const goalDriverFor = Effect.fn("Orchestrator.goalDriverFor")(function* ( + command: OrchestrationV2ServerCommand, + providerInstanceId: ProviderInstanceId, + ) { + const { driver } = yield* providerAdapters + .get(providerInstanceId) + .pipe(mapDispatchError(command)); + return driver === "codex" || driver === "claudeAgent" ? driver : undefined; + }); + + /** Claude Code reads `/goal pause` and `/goal resume` as new goals named "pause" and "resume". */ + const ensureGoalCommandSupported = Effect.fn("Orchestrator.ensureGoalCommandSupported")( + function* ( + command: OrchestrationV2ServerCommand, + message: { readonly text: string; readonly attachments: ReadonlyArray }, + providerInstanceId: ProviderInstanceId, + ) { + const goalCommand = parseGoalCommand(message); + if (goalCommand?.type !== "pause" && goalCommand?.type !== "resume") return; + if ((yield* goalDriverFor(command, providerInstanceId)) !== "claudeAgent") return; + return yield* new OrchestratorDispatchError({ + commandId: command.commandId, + commandType: command.type, + cause: "Claude goals cannot be paused or resumed. Send /goal clear to remove the goal.", + }); + }, + ); + const mapDispatchError = (command: OrchestrationV2ServerCommand) => (effect: Effect.Effect): Effect.Effect => @@ -4278,7 +4304,7 @@ const makeOrchestrator = Effect.fn("orchestrationV2.Orchestrator.layer")(functio : "Signing out must run as a separate turn. Queue it or wait for the active turn to finish.", }); } - if (isGoalCommand(input)) { + if (parseGoalCommand(input) !== null) { return yield* new OrchestratorDispatchError({ commandId: input.command.commandId, commandType: input.command.type, @@ -5202,6 +5228,45 @@ const makeOrchestrator = Effect.fn("orchestrationV2.Orchestrator.layer")(functio ) { dispatchMode = { type: "queue_after_active" }; } + // Codex and Claude own /goal; elsewhere it is ordinary text that cannot steer. + const parsedGoalCommand = parseGoalCommand(command); + const goalCommand = + parsedGoalCommand === null || + (yield* goalDriverFor(command, modelSelection.instanceId)) === undefined + ? null + : parsedGoalCommand; + let goalStopRunId: RunId | undefined; + if (goalCommand !== null) { + yield* ensureGoalCommandSupported(command, command, modelSelection.instanceId); + // A goal command never joins a running turn. One that starts or ends goal + // work stops the run and goes first in the queue; the rest wait. A send + // that asked to be queued (no delivery intent) only queues, and a run + // that is only finishing or running background work is never stopped. + const resolvedMode = dispatchMode; + const target = + resolvedMode.type === "steer_active" || resolvedMode.type === "restart_active" + ? projection.runs.find((run) => run.id === resolvedMode.targetRunId) + : resolvedMode.type === "queue_after_active" && command.deliveryIntent !== undefined + ? projection.runs.findLast(isLiveRun) + : undefined; + if (target !== undefined) { + const goal = projection.providerThreads.find( + (candidate) => candidate.id === target.providerThreadId, + )?.goal; + const targetMessage = projection.messages.find( + (message) => message.id === target.userMessageId, + ); + if ( + isLiveRun(target) && + goalCommandStopsActiveRun(goalCommand, goal?.status ?? null) && + (targetMessage === undefined || !isNativeMaintenanceCommand(targetMessage)) && + !(yield* stopReachedRun(command, command.threadId, target.id)) + ) { + goalStopRunId = target.id; + } + dispatchMode = { type: "queue_after_active" }; + } + } if ( command.notification !== undefined && (command.createdBy !== "agent" || @@ -5547,12 +5612,14 @@ const makeOrchestrator = Effect.fn("orchestrationV2.Orchestrator.layer")(functio ? { queueHeld: true } : {}), queuePosition: - Math.max( - 0, - ...projection.runs - .filter((candidate) => candidate.status === "queued") - .map((candidate) => candidate.queuePosition ?? candidate.ordinal), - ) + 1, + goalStopRunId !== undefined + ? 1 + : Math.max( + 0, + ...projection.runs + .filter((candidate) => candidate.status === "queued") + .map((candidate) => candidate.queuePosition ?? candidate.ordinal), + ) + 1, requestedAt: now, startedAt: null, completedAt: null, @@ -5705,6 +5772,17 @@ const makeOrchestrator = Effect.fn("orchestrationV2.Orchestrator.layer")(functio }, }); } + if (goalStopRunId !== undefined) { + return yield* stopRunForGoalCommand({ + command, + events, + effects, + projection: yield* getProjectionWithPendingEvents(command.threadId, events), + queuedRunId: runId, + targetRunId: goalStopRunId, + now, + }); + } return; } const pendingForkTransfer = pendingForkTransferForThread(projection); @@ -8270,6 +8348,46 @@ const makeOrchestrator = Effect.fn("orchestrationV2.Orchestrator.layer")(functio const now = yield* DateTime.now; const emitEvent = emit(events, command); + const parsedGoalCommand = parseGoalCommand(queuedMessage); + const goalCommand = + parsedGoalCommand === null || + (yield* goalDriverFor(command, queuedRun.providerInstanceId)) === undefined + ? null + : parsedGoalCommand; + if (goalCommand !== null) { + yield* ensureGoalCommandSupported(command, queuedMessage, queuedRun.providerInstanceId); + const target = projection.runs.find((run) => run.id === command.targetRunId); + const goal = projection.providerThreads.find( + (candidate) => candidate.id === target?.providerThreadId, + )?.goal; + const targetMessage = projection.messages.find( + (message) => message.id === target?.userMessageId, + ); + yield* emitEvent({ + type: "run.updated", + threadId: command.threadId, + runId: queuedRun.id, + nodeId: queuedRootNode.id, + providerInstanceId: queuedRun.providerInstanceId, + occurredAt: now, + payload: { ...queuedRun, queuePosition: 1 }, + }); + const stops = + target !== undefined && + isLiveRun(target) && + goalCommandStopsActiveRun(goalCommand, goal?.status ?? null) && + (targetMessage === undefined || !isNativeMaintenanceCommand(targetMessage)) && + !(yield* stopReachedRun(command, command.threadId, target.id)); + return yield* stopRunForGoalCommand({ + command, + events, + effects, + projection, + queuedRunId: queuedRun.id, + targetRunId: stops ? target.id : undefined, + now, + }); + } yield* emitEvent({ type: "run.updated", threadId: command.threadId, @@ -8580,6 +8698,11 @@ const makeOrchestrator = Effect.fn("orchestrationV2.Orchestrator.layer")(functio cause: "A guest's message cannot be edited. Cancel it instead.", }); } + yield* ensureGoalCommandSupported( + command, + { text: command.text, attachments: command.attachments ?? queuedMessage.attachments }, + queuedRun.providerInstanceId, + ); const queuedTurnItem = projection.turnItems.find( (candidate) => candidate.type === "user_message" && candidate.messageId === queuedMessage.id, ); @@ -10371,6 +10494,73 @@ const makeOrchestrator = Effect.fn("orchestrationV2.Orchestrator.layer")(functio ), ); + /** Puts a queued goal command first and, given a target, stops the run in its way. */ + const stopRunForGoalCommand = Effect.fn("Orchestrator.stopRunForGoalCommand")(function* (input: { + readonly command: Extract< + OrchestrationV2Command, + { readonly type: "message.dispatch" | "queued-message.promote-to-steer" } + >; + readonly events: Ref.Ref>; + readonly effects: Ref.Ref>; + readonly projection: Pick; + readonly queuedRunId: RunId; + readonly targetRunId: RunId | undefined; + readonly now: DateTime.Utc; + }) { + const { command, events, effects, projection, now, targetRunId } = input; + const others = queuedRunsInDeliveryOrder(projection).filter( + (run) => run.id !== input.queuedRunId && !isAutomaticCompletionRun(projection, run), + ); + for (const [index, run] of others.entries()) { + const queuePosition = index + 2; + if (run.queuePosition === queuePosition) continue; + yield* emit( + events, + command, + )({ + type: "run.updated", + threadId: command.threadId, + runId: run.id, + ...(run.rootNodeId === null ? {} : { nodeId: run.rootNodeId }), + providerInstanceId: run.providerInstanceId, + occurredAt: now, + payload: { ...run, queuePosition }, + }); + } + if (targetRunId === undefined) return undefined; + const interruptEvents = yield* Ref.make>([]); + const interruptEffects = yield* Ref.make>([]); + const interrupted = yield* Effect.exit( + dispatchRunInterrupt( + { + type: "run.interrupt", + commandId: command.commandId, + threadId: command.threadId, + runId: targetRunId, + reason: "Stopped to apply /goal", + }, + interruptEvents, + interruptEffects, + ), + ); + // A failed interrupt may have written part of its events, so they are kept apart. + // The goal command then waits for the turn to end. + if (Exit.isFailure(interrupted)) { + yield* Effect.logWarning("goal command could not stop the running turn", { + threadId: command.threadId, + runId: targetRunId, + cause: interrupted.cause, + }); + return undefined; + } + const written = yield* Ref.get(interruptEvents); + const enqueued = yield* Ref.get(interruptEffects); + yield* Ref.update(events, (existing) => [...existing, ...written]); + yield* Ref.update(effects, (existing) => [...existing, ...enqueued]); + // Like Stop: a run stopped before its provider started also abandons that start. + return interrupted.value; + }); + const dispatchThreadStop = Effect.fn("Orchestrator.dispatchThreadStop")(function* ( command: Extract, events: Ref.Ref>, @@ -11709,7 +11899,7 @@ const makeOrchestrator = Effect.fn("orchestrationV2.Orchestrator.layer")(functio text: guestSend.text, context: { version: 1 as const, records: [], guest: guestSend.guest }, }; - yield* dispatchMessage(dispatched, events, effects); + cancelUnsettledEffects = yield* dispatchMessage(dispatched, events, effects); yield* recordSubagentFollowup(dispatched, events).pipe(mapDispatchError(command)); break; } @@ -11769,7 +11959,11 @@ const makeOrchestrator = Effect.fn("orchestrationV2.Orchestrator.layer")(functio cancelUnsettledEffects = yield* dispatchThreadStop(command, events, effects); break; case "queued-message.promote-to-steer": - yield* dispatchQueuedMessagePromoteToSteer(command, events, effects); + cancelUnsettledEffects = yield* dispatchQueuedMessagePromoteToSteer( + command, + events, + effects, + ); break; case "queued-run.start": case "queue.resume": { diff --git a/apps/server/src/orchestration-v2/ProviderTurnStartService.ts b/apps/server/src/orchestration-v2/ProviderTurnStartService.ts index a6fdcabc0b3a4..97fdb7c85ca71 100644 --- a/apps/server/src/orchestration-v2/ProviderTurnStartService.ts +++ b/apps/server/src/orchestration-v2/ProviderTurnStartService.ts @@ -9,6 +9,8 @@ import { type OrchestrationV2Run, type OrchestrationV2RunAttempt, type OrchestrationV2TurnItem, + isCompactionCommandMessage, + parseGoalCommand, RunId, ThreadId, } from "@t3tools/contracts"; @@ -282,11 +284,12 @@ export const layer: Layer.Layer< source.status === "interrupted" || (source.status === "completed" && handoff.delivery === undefined && + // Compaction and /goal turns defer history to the next turn. projection.messages.some( (message) => message.id === source.userMessageId && - message.attachments.length === 0 && - message.text.trim().toLowerCase() === "/compact", + (isCompactionCommandMessage(message) || + parseGoalCommand(message) !== null), ))), ))), ); @@ -1036,6 +1039,9 @@ export const layer: Layer.Layer< text: message.text, records: message.context?.records ?? [], }); + // Codex and Claude parse /goal from the prompt, so it goes verbatim and history waits a turn. + const goalDriver = session.driver === "codex" || session.driver === "claudeAgent"; + const goalCommand = goalDriver && parseGoalCommand(message) !== null; // The provider turn marks cancellation work delivered; restart continuations use their own text or resume natively. const noteContinuation = isRestartNoteContinuation( run, @@ -1050,8 +1056,8 @@ export const layer: Layer.Layer< projection.messages .filter( (candidate) => - candidate.attachments.length === 0 && - candidate.text.trim().toLowerCase() === "/compact", + isCompactionCommandMessage(candidate) || + (goalDriver && parseGoalCommand(candidate) !== null), ) .map((candidate) => candidate.id), ), @@ -1199,7 +1205,7 @@ export const layer: Layer.Layer< ]; const delivery = yield* deliverContextHandoffs({ handoffs: [...effectiveHandoffs, ...retryHandoff], - deferInline: compact, + deferInline: compact || goalCommand, providerThread: runningProviderThread, budget: Effect.gen(function* () { return handoffBudget({ @@ -1253,7 +1259,10 @@ export const layer: Layer.Layer< ...(noteContinuation ? promptedInput : turnInput), message: { ...turnInput.message, - text: context === "" ? userText : `${context}\n\nUser message:\n${userText}`, + text: + context === "" || goalCommand + ? userText + : `${context}\n\nUser message:\n${userText}`, }, }).pipe( // Clear pending handoff markers for refused prompts so the next turn retains the native session. diff --git a/apps/server/src/orchestration-v2/SteeringCompletion.integration.test.ts b/apps/server/src/orchestration-v2/SteeringCompletion.integration.test.ts index 1a5ee50872706..2cc6ee6f825b2 100644 --- a/apps/server/src/orchestration-v2/SteeringCompletion.integration.test.ts +++ b/apps/server/src/orchestration-v2/SteeringCompletion.integration.test.ts @@ -547,7 +547,7 @@ const nextTurnSelectionHarness = Effect.fn("nextTurnSelectionHarness")(function* yield* Fiber.join(running); return threadId; }); - return { events, started, steered, layer, startFirstTurn }; + return { cwd, events, started, steered, layer, startFirstTurn }; }); it.effect("sends a selected held message once and preserves the remaining queue order", () => @@ -1048,3 +1048,218 @@ it.effect("queues a plain send behind the stopped queue and releases its hold", }), ), ); + +const goalHarness = Effect.fn("goalHarness")(function* (name: string) { + const harness = yield* nextTurnSelectionHarness(name); + const send = (id: string, text: string, delivery: "queue" | "auto") => + Effect.gen(function* () { + const orchestrator = yield* Orchestrator.OrchestratorV2; + yield* orchestrator.dispatch({ + type: "message.dispatch", + commandId: CommandId.make(id), + threadId: ThreadId.make(`thread:${name}`), + messageId: MessageId.make(`message:${id}`), + text, + attachments: [], + modelSelection: runSelection, + ...(delivery === "queue" + ? { dispatchMode: { type: "queue_after_active" } } + : { dispatchMode: { type: "start_immediately" }, deliveryIntent: "auto" }), + createdBy: "user", + creationSource: "web", + }); + }); + const read = (threadId: ThreadId) => + Effect.gen(function* () { + const projection = yield* (yield* Orchestrator.OrchestratorV2).getThreadProjection(threadId); + const textOf = (run: (typeof projection.runs)[number]) => + projection.messages.find((message) => message.id === run.userMessageId)?.text; + return { + queue: projection.runs + .filter((run) => run.status === "queued") + .toSorted((left, right) => (left.queuePosition ?? 0) - (right.queuePosition ?? 0)) + .map(textOf), + started: harness.started.map((turn) => turn.message.text), + stopRequested: projection.turnItems.some( + (item) => + item.type === "run_interrupt_request" && item.runId === harness.started[0]?.runId, + ), + }; + }); + return { ...harness, send, read }; +}); + +it.effect("stops the running turn and starts a goal sent into it next", () => + Effect.scoped( + Effect.gen(function* () { + const { steered, layer, startFirstTurn, send, read } = yield* goalHarness("goal-steer"); + yield* Effect.gen(function* () { + const worker = yield* EffectWorker.OrchestrationEffectWorkerV2; + const threadId = yield* startFirstTurn; + yield* send("queued", "queued", "queue"); + yield* send("goal", "/goal all tests pass", "auto"); + yield* worker.drain(); + + assert.deepEqual(steered, []); + assert.deepEqual(yield* read(threadId), { + queue: ["queued"], + started: ["first", "/goal all tests pass"], + stopRequested: true, + }); + }).pipe(Effect.provide(layer)); + }), + ), +); + +it.effect("queues a goal clear without stopping work when no goal drives it", () => + Effect.scoped( + Effect.gen(function* () { + const { steered, layer, startFirstTurn, send, read } = yield* goalHarness("goal-clear"); + yield* Effect.gen(function* () { + const worker = yield* EffectWorker.OrchestrationEffectWorkerV2; + const threadId = yield* startFirstTurn; + yield* send("queued", "queued", "queue"); + yield* send("clear", "/goal clear", "auto"); + yield* worker.drain(); + + assert.deepEqual(steered, []); + assert.deepEqual(yield* read(threadId), { + queue: ["queued", "/goal clear"], + started: ["first"], + stopRequested: false, + }); + }).pipe(Effect.provide(layer)); + }), + ), +); + +it.effect("promotes a queued goal by stopping the running turn", () => + Effect.scoped( + Effect.gen(function* () { + const { started, steered, layer, startFirstTurn, send, read } = + yield* goalHarness("goal-promote"); + yield* Effect.gen(function* () { + const orchestrator = yield* Orchestrator.OrchestratorV2; + const worker = yield* EffectWorker.OrchestrationEffectWorkerV2; + const threadId = yield* startFirstTurn; + yield* send("queued", "queued", "queue"); + yield* send("goal", "/goal ship it", "queue"); + const goalRun = (yield* orchestrator.getThreadProjection(threadId)).runs.find( + (run) => run.userMessageId === MessageId.make("message:goal"), + )!; + yield* orchestrator.dispatch({ + type: "queued-message.promote-to-steer", + commandId: CommandId.make("promote"), + threadId, + queuedRunId: goalRun.id, + targetRunId: started[0]!.runId, + }); + yield* worker.drain(); + + assert.deepEqual(steered, []); + assert.deepEqual(yield* read(threadId), { + queue: ["queued"], + started: ["first", "/goal ship it"], + stopRequested: true, + }); + }).pipe(Effect.provide(layer)); + }), + ), +); + +it.effect("stops a turn that is still starting when a goal is sent into it", () => + Effect.scoped( + Effect.gen(function* () { + const { cwd, layer, send, started } = yield* goalHarness("goal-during-start"); + yield* Effect.gen(function* () { + const orchestrator = yield* Orchestrator.OrchestratorV2; + const worker = yield* EffectWorker.OrchestrationEffectWorkerV2; + const threadId = ThreadId.make("thread:goal-during-start"); + yield* orchestrator.dispatch({ + type: "thread.create", + commandId: CommandId.make("create"), + threadId, + projectId: ProjectId.make("project:goal-during-start"), + title: "Goal during start", + modelSelection: runSelection, + runtimeMode: "full-access", + interactionMode: "default", + branch: null, + worktreePath: cwd, + createdBy: "user", + creationSource: "web", + }); + // The stopped run's end starts the queue, which then needs one more drain. + const goalStarting = yield* orchestrator.streamDomainEvents.pipe( + Stream.filter( + (event) => event.type === "run.updated" && event.payload.status === "starting", + ), + Stream.take(1), + Stream.runDrain, + Effect.forkScoped({ startImmediately: true }), + ); + yield* send("first", "first", "auto"); + yield* send("goal", "/goal all tests pass", "auto"); + yield* worker.drain(); + yield* Fiber.join(goalStarting); + yield* worker.drain(); + + const projection = yield* orchestrator.getThreadProjection(threadId); + const statusOf = (text: string) => + projection.runs.find( + (run) => + projection.messages.find((message) => message.id === run.userMessageId)?.text === + text, + )?.status; + assert.equal(statusOf("first"), "interrupted"); + assert.deepEqual( + started.map((turn) => turn.message.text), + ["/goal all tests pass"], + ); + }).pipe(Effect.provide(layer)); + }), + ), +); + +it.effect("stops the running turn when a goal clear ends the goal driving it", () => + Effect.scoped( + Effect.gen(function* () { + const { events, layer, startFirstTurn, send, read, steered } = + yield* goalHarness("goal-clear-active"); + yield* Effect.gen(function* () { + const orchestrator = yield* Orchestrator.OrchestratorV2; + const worker = yield* EffectWorker.OrchestrationEffectWorkerV2; + const threadId = yield* startFirstTurn; + const providerThread = (yield* orchestrator.getThreadProjection(threadId)) + .providerThreads[0]!; + const goalSeen = yield* orchestrator.streamDomainEvents.pipe( + Stream.filter( + (event) => + event.type === "provider-thread.updated" && event.payload.goal?.status === "active", + ), + Stream.take(1), + Stream.runDrain, + Effect.forkScoped({ startImmediately: true }), + ); + yield* Queue.offer(events, { + type: "provider_thread.updated", + driver, + providerThread: { + ...providerThread, + goal: { objective: "all tests pass", status: "active" }, + }, + }); + yield* Fiber.join(goalSeen); + yield* send("clear", "/goal clear", "auto"); + yield* worker.drain(); + + assert.deepEqual(steered, []); + assert.deepEqual(yield* read(threadId), { + queue: [], + started: ["first", "/goal clear"], + stopRequested: true, + }); + }).pipe(Effect.provide(layer)); + }), + ), +); diff --git a/apps/web/src/components/chat/ThreadActivityBanners.tsx b/apps/web/src/components/chat/ThreadActivityBanners.tsx index b9de23227e7c9..39a60ed27ce3e 100644 --- a/apps/web/src/components/chat/ThreadActivityBanners.tsx +++ b/apps/web/src/components/chat/ThreadActivityBanners.tsx @@ -68,9 +68,10 @@ export function goalBannerItem(input: { ? presentation.title : `${presentation.title} · ${presentation.usage}`, description: presentation.objective, - actions: isWorking ? undefined : ( + // Clear stops goal work; Resume would stop unrelated work, so it waits. + actions: ( <> - {presentation.canResume ? ( + {presentation.canResume && !isWorking ? (