From 02329327c64f524df1b6ea8d2d1817637e7f797e Mon Sep 17 00:00:00 2001 From: Julius Marminge Date: Tue, 27 Oct 2026 12:21:16 -0700 Subject: [PATCH 01/19] fix(clients): prompt links find projects cloned from git.cafe (#26042) Co-authored-by: Claude Opus 5.5 --- packages/client-runtime/src/promptLinks.test.ts | 13 +++++++++++++ packages/client-runtime/src/promptLinks.ts | 10 ++++++---- 2 files changed, 19 insertions(+), 4 deletions(-) diff --git a/packages/client-runtime/src/promptLinks.test.ts b/packages/client-runtime/src/promptLinks.test.ts index 213392f6c61f..12a73ddf9153 100644 --- a/packages/client-runtime/src/promptLinks.test.ts +++ b/packages/client-runtime/src/promptLinks.test.ts @@ -39,6 +39,19 @@ describe("findPromptLinkProjects", () => { expect(ids("me", "t3code")).toEqual(["fork"]); expect(ids("gitlab", "x")).toEqual([]); }); + + it("matches a checkout of the repository's GitCafe mirror", () => { + const projects = [ + { id: "gitcafe", repositoryIdentity: identity("git.cafe/pingdotgg/t3code") }, + { id: "staging", repositoryIdentity: identity("staging.git.cafe/pingdotgg/t3code") }, + { id: "gitlab", repositoryIdentity: identity("gitlab.com/pingdotgg/t3code") }, + ]; + expect( + findPromptLinkProjects(projects, { owner: "pingdotgg", name: "t3code" }).map( + (project) => project.id, + ), + ).toEqual(["gitcafe", "staging"]); + }); }); describe("pickPromptLinkPullRequestThread", () => { diff --git a/packages/client-runtime/src/promptLinks.ts b/packages/client-runtime/src/promptLinks.ts index 135b1c13f110..2c72280745b4 100644 --- a/packages/client-runtime/src/promptLinks.ts +++ b/packages/client-runtime/src/promptLinks.ts @@ -18,8 +18,9 @@ import { visibleThreadPullRequests, } from "./threadPullRequests.ts"; -const repositoryKey = (repository: PromptLinkRepository) => - `github.com/${repository.owner}/${repository.name}`.toLowerCase(); +// A link's `owner/name` is GitHub's. GitCafe syncs a repository two-way with GitHub under the +// same path, so a checkout cloned from git.cafe counts as the linked repository. +const PROMPT_LINK_REPOSITORY_HOSTS = ["github.com", "git.cafe", "staging.git.cafe"]; /** * The repository's projects in the order given. Callers pass projects most recently active @@ -28,11 +29,12 @@ const repositoryKey = (repository: PromptLinkRepository) => export function findPromptLinkProjects< P extends { readonly repositoryIdentity?: RepositoryIdentity | null | undefined }, >(projects: ReadonlyArray

, repository: PromptLinkRepository): P[] { - const key = repositoryKey(repository); + const path = `${repository.owner}/${repository.name}`.toLowerCase(); + const keys = new Set(PROMPT_LINK_REPOSITORY_HOSTS.map((host) => `${host}/${path}`)); return projects.filter( (project) => project.repositoryIdentity != null && - repositoryGroupingKeyOf(project.repositoryIdentity).toLowerCase() === key, + keys.has(repositoryGroupingKeyOf(project.repositoryIdentity).toLowerCase()), ); } From 740de0e642d630a08e37d0780fa99c178c27808e Mon Sep 17 00:00:00 2001 From: Yash Singh Date: Tue, 27 Oct 2026 14:36:56 -0500 Subject: [PATCH 02/19] fix(server): retry Codex capacity failures with empty input (#25611) Co-authored-by: maria-rcks --- .../Adapters/CodexAdapterV2.test.ts | 779 ++++++++++++++++++ .../Adapters/CodexAdapterV2.ts | 382 +++++++-- .../ProviderTurnControlService.ts | 6 +- .../SteeringCompletion.integration.test.ts | 26 +- 4 files changed, 1124 insertions(+), 69 deletions(-) diff --git a/apps/server/src/orchestration-v2/Adapters/CodexAdapterV2.test.ts b/apps/server/src/orchestration-v2/Adapters/CodexAdapterV2.test.ts index 602e1a4d56f2..1e762543a13a 100644 --- a/apps/server/src/orchestration-v2/Adapters/CodexAdapterV2.test.ts +++ b/apps/server/src/orchestration-v2/Adapters/CodexAdapterV2.test.ts @@ -2214,6 +2214,436 @@ describe("CodexAdapterV2 post-settle continuation", () => { ), ); + it.effect.each([ + "recover", + "completion-only", + "exhaust", + "stop", + "stop-continued", + "stop-starting", + "started-first", + "completed-first", + "new-run", + "steer", + "start-fails", + ] as const)("handles terminal Codex capacity failures when retries %s", (outcome) => + Effect.scoped( + Effect.gen(function* () { + const scenario = `capacity-${outcome}`; + const nativeThreadId = `native-${scenario}`; + const nativeTurnId = (attempt: number) => `${nativeThreadId}-${attempt}`; + const prompt = "Continue the task."; + const capacityError = { + message: "Selected model is at capacity. Please try a different model.", + codexErrorInfo: "serverOverloaded", + }; + const failureEntries = (attempt: number): Array => [ + { + type: "emit_inbound", + frame: { + method: "error", + params: { + threadId: nativeThreadId, + turnId: nativeTurnId(attempt), + willRetry: false, + error: capacityError, + }, + }, + }, + { + type: "emit_inbound", + frame: { + method: "turn/completed", + params: { + threadId: nativeThreadId, + turn: { + ...makeCodexReplayTurn({ id: nativeTurnId(attempt), status: "failed" }), + error: capacityError, + }, + }, + }, + }, + ]; + const entries = codexReplayPreamble({ + nativeThreadId, + nativeTurnId: nativeTurnId(0), + prompt, + }); + entries.push( + ...(outcome === "completion-only" ? failureEntries(0).slice(1) : failureEntries(0)), + ); + if (outcome === "completion-only") entries.push(...failureEntries(0).slice(1)); + const retries = outcome === "exhaust" ? 5 : outcome === "stop" ? 0 : 1; + const steeredText = "Use the smaller change."; + for (let attempt = 1; attempt <= retries; attempt++) { + const start = codexReplayPreamble({ + nativeThreadId, + nativeTurnId: nativeTurnId(attempt), + prompt, + }).slice(-3); + if ( + outcome === "started-first" || + outcome === "stop-starting" || + outcome === "completed-first" || + outcome === "new-run" + ) { + start.splice(1, 0, ...start.splice(2, 1)); + } + for (const entry of start) { + if (entry.type === "expect_outbound") { + entries.push({ + ...entry, + frame: { + id: 3 + attempt, + method: "turn/start", + params: { + threadId: nativeThreadId, + input: [], + cwd: "/workspace", + model: "gpt-5.4", + approvalPolicy: "never", + approvalsReviewer: "user", + sandboxPolicy: { type: "dangerFullAccess" }, + summary: "detailed", + collaborationMode: { mode: "default", settings: { model: "gpt-5.4" } }, + }, + }, + }); + } else if (entry.type === "emit_inbound" && entry.label === "turn/start") { + if (outcome === "completed-first" || outcome === "new-run") + entries.push({ + type: "emit_inbound", + frame: { + method: "turn/completed", + params: { + threadId: nativeThreadId, + turn: makeCodexReplayTurn({ id: nativeTurnId(attempt), status: "completed" }), + }, + }, + }); + if (outcome === "new-run") { + entries.push( + ...codexReplayPreamble({ + nativeThreadId, + nativeTurnId: nativeTurnId(2), + prompt: "Next task.", + }) + .slice(-3) + .map((next, index) => (index < 2 ? withReplayRequestId(next, 5) : next)), + ...failureEntries(2), + ); + } + entries.push( + outcome === "start-fails" + ? { + type: "emit_inbound", + frame: { + id: 3 + attempt, + error: { code: -32603, message: "Start failed" }, + }, + } + : withReplayRequestId(entry, 3 + attempt), + ); + } else if (outcome !== "start-fails") { + entries.push( + outcome === "stop-starting" && entry.type === "emit_inbound" + ? { ...entry, afterMs: 1_000 } + : entry, + ); + if (outcome === "stop-starting") { + entries.push( + { + type: "expect_outbound", + frame: { + id: 5, + method: "turn/interrupt", + params: { threadId: nativeThreadId, turnId: nativeTurnId(attempt) }, + }, + }, + { type: "emit_inbound", frame: { id: 5, result: {} } }, + ); + } + } + } + if (outcome === "exhaust") entries.push(...failureEntries(attempt)); + if ( + outcome === "recover" || + outcome === "completion-only" || + outcome === "started-first" || + outcome === "stop-starting" + ) { + entries.push({ + type: "emit_inbound", + frame: { + method: "turn/completed", + params: { + threadId: nativeThreadId, + turn: makeCodexReplayTurn({ + id: nativeTurnId(attempt), + status: outcome === "stop-starting" ? "interrupted" : "completed", + }), + }, + }, + }); + } + if (outcome === "stop-continued" || outcome === "steer") { + const steering = outcome === "steer"; + entries.push( + { + type: "expect_outbound", + frame: { + id: 5, + method: steering ? "turn/steer" : "turn/interrupt", + params: steering + ? { + threadId: nativeThreadId, + expectedTurnId: nativeTurnId(attempt), + input: [{ type: "text", text: steeredText }], + } + : { threadId: nativeThreadId, turnId: nativeTurnId(attempt) }, + }, + }, + { + type: "emit_inbound", + frame: { id: 5, result: steering ? { turnId: nativeTurnId(attempt) } : {} }, + }, + { + type: "emit_inbound", + frame: { + method: "turn/completed", + params: { + threadId: nativeThreadId, + turn: makeCodexReplayTurn({ + id: nativeTurnId(attempt), + status: steering ? "completed" : "interrupted", + }), + }, + }, + }, + ); + } + } + const milestones = yield* Queue.unbounded(); + const starts: Array = []; + const retryStarted = yield* Deferred.make(); + const harness = yield* makeCodexReplayHarness( + makeCodexReplayTranscript({ scenario, entries }), + (event) => + event.type === "turn.terminal" || + (event.type === "turn_item.updated" && + event.turnItem.type === "error" && + event.turnItem.status === "running") + ? Queue.offer(milestones, event) + : Effect.void, + (method, params) => + Effect.gen(function* () { + if (method === "turn/start") { + starts.push(params); + if (starts.length === 2) yield* Deferred.succeed(retryStarted, undefined); + } + }), + ); + const turnInput = makeCodexTestTurnInput({ + threadId: harness.threadId, + providerThread: harness.providerThread, + now: yield* DateTime.now, + attemptId: RunAttemptId.make(scenario), + text: prompt, + }); + yield* harness.runtime.startTurn(turnInput); + for (let attempt = 1; attempt <= Math.max(retries, 1); attempt++) { + const milestone = yield* Queue.take(milestones); + assert.equal(milestone.type, "turn_item.updated"); + if (milestone.type !== "turn_item.updated" || milestone.turnItem.type !== "error") return; + const providerTurnId = milestone.turnItem.providerTurnId; + assert.isNotNull(providerTurnId); + if (providerTurnId === null) return; + assert.deepEqual(milestone.turnItem.retry, { + attempt, + maxAttempts: 5, + retryDelayMs: Math.min(2_000 * 2 ** (attempt - 1), 30_000), + }); + assert.isEmpty(harness.terminalEvents()); + if (outcome === "stop") { + yield* harness.runtime.interruptTurn({ + providerThread: harness.providerThread, + providerTurnId, + }); + } + const steering = + outcome === "steer" + ? yield* harness.runtime + .steerTurn({ + ...turnInput, + providerTurnId, + message: { ...turnInput.message, text: steeredText }, + }) + .pipe(Effect.forkScoped) + : undefined; + yield* TestClock.adjust(Math.min(2_000 * 2 ** (attempt - 1), 30_000)); + if (outcome === "stop-starting") { + yield* Deferred.await(retryStarted); + const stopping = yield* harness.runtime + .interruptTurn({ providerThread: harness.providerThread, providerTurnId }) + .pipe(Effect.forkScoped); + yield* TestClock.adjust("1 second"); + yield* Fiber.join(stopping); + } + if (steering !== undefined) yield* Fiber.join(steering); + if (outcome === "stop-continued") { + yield* harness.until(() => + harness.events.some( + (event) => + event.type === "provider_turn.updated" && + event.providerTurn.nativeTurnRef?.nativeId === nativeTurnId(1), + ), + ); + yield* harness.runtime.interruptTurn({ + providerThread: harness.providerThread, + providerTurnId, + }); + } + } + yield* harness.awaitTerminal(1); + if (outcome === "new-run") { + assert.equal((yield* Queue.take(milestones)).type, "turn.terminal"); + yield* harness.runtime.startTurn({ + ...turnInput, + attemptId: RunAttemptId.make("capacity-next-task"), + message: { ...turnInput.message, text: "Next task." }, + }); + const nextRetry = yield* Queue.take(milestones); + assert.equal(nextRetry.type, "turn_item.updated"); + if (nextRetry.type !== "turn_item.updated" || nextRetry.turnItem.type !== "error") return; + assert.equal(nextRetry.turnItem.retry?.attempt, 1); + return; + } + const terminal = harness.terminalEvents()[0]; + assert.equal( + terminal?.status, + outcome === "recover" || + outcome === "completion-only" || + outcome === "steer" || + outcome === "started-first" || + outcome === "completed-first" + ? "completed" + : outcome === "stop" || outcome === "stop-continued" || outcome === "stop-starting" + ? "interrupted" + : "failed", + ); + assert.lengthOf(starts, 1 + retries); + assert.lengthOf(harness.terminalEvents(), 1); + assert.lengthOf( + harness.events.filter( + (event) => + event.type === "provider_turn.updated" && event.providerTurn.status === "running", + ), + outcome === "start-fails" ? 1 : 1 + retries, + ); + const retryItems = harness.events.flatMap((event) => + event.type === "turn_item.updated" && event.turnItem.type === "error" + ? [event.turnItem] + : [], + ); + assert.isTrue( + ["completed", "failed", "interrupted"].includes(retryItems.at(-1)?.status ?? ""), + ); + assert.lengthOf(new Set(retryItems.map((item) => item.id)), 1); + + if (outcome === "recover") { + const fs = yield* FileSystem.FileSystem; + const cwd = yield* fs.makeTempDirectoryScoped({ prefix: "t3-capacity-retry-" }); + const localTranscript = yield* decodeReplayTranscriptJson( + (yield* encodeReplayTranscriptJson( + makeCodexReplayTranscript({ scenario, entries }), + )).replaceAll(yield* encodeStringJson("/workspace"), yield* encodeStringJson(cwd)), + ); + const driver = yield* CodexReplay.makeReplayDriver(localTranscript); + yield* Effect.gen(function* () { + const orchestrator = yield* Orchestrator.OrchestratorV2; + const worker = yield* EffectWorker.OrchestrationEffectWorkerV2; + const threadId = ThreadId.make("thread:capacity-retry"); + yield* orchestrator.dispatch({ + type: "thread.create", + commandId: CommandId.make("create-capacity-retry"), + threadId, + projectId: ProjectId.make("project:capacity-retry"), + title: "Capacity retry", + modelSelection: CODEX_TEST_MODEL_SELECTION, + runtimeMode: "full-access", + interactionMode: "default", + branch: null, + worktreePath: cwd, + createdBy: "user", + creationSource: "web", + }); + const retrying = yield* orchestrator.streamDomainEvents.pipe( + Stream.filter( + (event) => + event.type === "turn-item.updated" && + event.payload.type === "error" && + event.payload.retry != null && + event.payload.status === "running", + ), + Stream.runHead, + Effect.forkChild({ startImmediately: true }), + ); + yield* orchestrator.dispatch({ + type: "message.dispatch", + commandId: CommandId.make("start-capacity-retry"), + threadId, + messageId: MessageId.make("message:capacity-retry"), + text: prompt, + attachments: [], + createdBy: "user", + creationSource: "web", + dispatchMode: { type: "start_immediately" }, + }); + yield* worker.drain(); + yield* Fiber.join(retrying); + assert.equal( + (yield* orchestrator.getThreadProjection(threadId)).runs.at(-1)?.status, + "running", + ); + const completed = yield* orchestrator.streamDomainEvents.pipe( + Stream.filter( + (event) => event.type === "run.updated" && event.payload.status === "waiting", + ), + Stream.runHead, + Effect.forkChild({ startImmediately: true }), + ); + yield* TestClock.adjust("2 seconds"); + yield* Fiber.join(completed); + yield* worker.drain(); + const projection = yield* orchestrator.getThreadProjection(threadId); + assert.equal(projection.runs.at(-1)?.status, "completed"); + assert.deepEqual( + projection.providerTurns.map((turn) => turn.status), + ["failed", "completed"], + ); + assert.lengthOf( + projection.messages.filter((message) => message.role === "user"), + 1, + ); + assert.isNull((yield* Ref.get(driver.state)).failure); + }).pipe( + Effect.provide( + ProviderReplayHarness.layerWithRegistry( + { name: "codex-capacity-retry", runtimePolicyOverride: { cwd } }, + CodexAdapterV2Testkit.layer({ transcript: localTranscript, driver }), + { runEffectWorker: false }, + ), + ), + ); + } + }).pipe( + Effect.provide( + Layer.mergeAll(IdAllocator.layer, McpProviderSessions.layer, NodeServices.layer), + ), + ), + ), + ); + it.effect("resolves retryable app-server errors on resumed provider activity", () => Effect.scoped( Effect.gen(function* () { @@ -5007,6 +5437,265 @@ describe("CodexAdapterV2 post-settle continuation", () => { }; }; + it.effect.each(["completed", "failed", "interrupted", "starting", "replied-last"] as const)( + "handles a late native child after the capacity retry %s", + (parentStatus) => + Effect.scoped( + Effect.gen(function* () { + const retriedTurnId = "native-capacity-recovered-parent"; + const retrying = yield* Deferred.make(); + const entries: Array = [ + ...lateChildPrefix().spawned, + { + type: "emit_inbound", + frame: { + method: "turn/completed", + params: { + threadId: RESUME_NATIVE_THREAD, + turn: { + ...makeCodexReplayTurn({ id: RESUME_NATIVE_TURN, status: "failed" }), + error: { + message: "Selected model is at capacity.", + codexErrorInfo: "serverOverloaded", + }, + }, + }, + }, + }, + { + type: "expect_outbound", + frame: { + id: 4, + method: "turn/start", + params: { + threadId: RESUME_NATIVE_THREAD, + input: [], + cwd: "/workspace", + model: "gpt-5.4", + approvalPolicy: "never", + approvalsReviewer: "user", + sandboxPolicy: { type: "dangerFullAccess" }, + summary: "detailed", + collaborationMode: { mode: "default", settings: { model: "gpt-5.4" } }, + }, + }, + }, + ...(parentStatus === "starting" ? [childTurnCompleted(RESUME_CHILD_TURN_1)] : []), + { + type: "emit_inbound", + frame: { + id: 4, + result: { + turn: makeCodexReplayTurn({ id: retriedTurnId, status: "inProgress" }), + }, + }, + }, + { + type: "emit_inbound", + frame: { + method: "turn/started", + params: { + threadId: RESUME_NATIVE_THREAD, + turn: makeCodexReplayTurn({ id: retriedTurnId, status: "inProgress" }), + }, + }, + }, + { + type: "emit_inbound", + frame: { + method: "turn/completed", + params: { + threadId: RESUME_NATIVE_THREAD, + turn: { + ...makeCodexReplayTurn({ + id: retriedTurnId, + status: + parentStatus === "starting" || parentStatus === "replied-last" + ? "completed" + : parentStatus, + }), + error: + parentStatus === "failed" + ? { message: "Recovery failed.", codexErrorInfo: "internalServerError" } + : null, + }, + }, + }, + }, + childAgentMessage({ + id: "child-after-capacity", + text: "CAPACITY_CHILD_DONE", + turnId: RESUME_CHILD_TURN_1, + completedAtMs: 1782622480000, + afterMs: 60_000, + }), + childTurnCompleted(RESUME_CHILD_TURN_1), + ]; + if (parentStatus === "replied-last") { + entries.push(...entries.splice(lateChildPrefix().spawned.length + 2, 1)); + } + const harness = yield* makeCodexReplayHarness( + makeCodexReplayTranscript({ scenario: "capacity-late-child", entries }), + (event) => + event.type === "turn_item.updated" && + event.turnItem.type === "error" && + event.turnItem.status === "running" + ? Deferred.succeed(retrying, undefined) + : Effect.void, + ); + yield* harness.runtime.startTurn( + makeCodexTestTurnInput({ + threadId: harness.threadId, + providerThread: harness.providerThread, + now: yield* DateTime.now, + attemptId: RunAttemptId.make("capacity-late-child"), + text: RESUME_PROMPT, + }), + ); + yield* Deferred.await(retrying); + yield* TestClock.adjust("2 seconds"); + if (parentStatus === "interrupted") assert.isEmpty(harness.terminalEvents()); + else { + yield* harness.awaitTerminal(1); + const terminal = harness.terminalEvents()[0]; + if (terminal?.status === "failed") assert.isTrue(terminal.childTurnsContinue); + } + yield* TestClock.adjust("60 seconds"); + yield* harness.until( + () => harness.subagentUpdates().at(-1)?.subagent.status === "completed", + ); + yield* harness.awaitTerminal(1); + assert.lengthOf( + harness.continuationRequests, + parentStatus === "completed" || parentStatus === "replied-last" ? 1 : 0, + ); + if (parentStatus === "completed" || parentStatus === "replied-last") + assert.include(harness.continuationRequests[0]?.detail, "CAPACITY_CHILD_DONE"); + }).pipe( + Effect.provide( + Layer.mergeAll(IdAllocator.layer, McpProviderSessions.layer, NodeServices.layer), + ), + ), + ), + ); + + it.effect("stops native children when capacity retry startup fails", () => + Effect.scoped( + Effect.gen(function* () { + const retrying = yield* Deferred.make(); + const starting = yield* Deferred.make(); + const release = yield* Deferred.make(); + const interrupts: unknown[] = []; + let starts = 0; + const harness = yield* makeCodexReplayHarness( + makeCodexReplayTranscript({ + scenario: "capacity-start-fails-stop-child", + entries: [ + ...lateChildPrefix().spawned, + { + type: "emit_inbound", + frame: { + method: "turn/completed", + params: { + threadId: RESUME_NATIVE_THREAD, + turn: { + ...makeCodexReplayTurn({ id: RESUME_NATIVE_TURN, status: "failed" }), + error: { message: "At capacity", codexErrorInfo: "serverOverloaded" }, + }, + }, + }, + }, + { + type: "expect_outbound", + frame: { + id: 4, + method: "turn/start", + params: { + threadId: RESUME_NATIVE_THREAD, + input: [], + cwd: "/workspace", + model: "gpt-5.4", + approvalPolicy: "never", + approvalsReviewer: "user", + sandboxPolicy: { type: "dangerFullAccess" }, + summary: "detailed", + collaborationMode: { mode: "default", settings: { model: "gpt-5.4" } }, + }, + }, + }, + { + type: "emit_inbound", + frame: { id: 4, error: { code: -32603, message: "Start failed" } }, + }, + { + type: "expect_outbound", + frame: { + id: 5, + method: "turn/interrupt", + params: { threadId: RESUME_CHILD_THREAD, turnId: RESUME_CHILD_TURN_1 }, + }, + }, + { type: "emit_inbound", frame: { id: 5, result: {} } }, + { + type: "emit_inbound", + frame: { + method: "turn/completed", + params: { + threadId: RESUME_CHILD_THREAD, + turn: makeCodexReplayTurn({ id: RESUME_CHILD_TURN_1, status: "interrupted" }), + }, + }, + }, + ], + }), + (event) => + event.type === "turn_item.updated" && + event.turnItem.type === "error" && + event.turnItem.status === "running" + ? Deferred.succeed(retrying, undefined) + : Effect.void, + (method, params) => + Effect.gen(function* () { + if (method === "turn/interrupt") interrupts.push(params); + if (method === "turn/start" && ++starts === 2) { + yield* Deferred.succeed(starting, undefined); + yield* Deferred.await(release); + } + }), + ); + yield* harness.runtime.startTurn( + makeCodexTestTurnInput({ + threadId: harness.threadId, + providerThread: harness.providerThread, + now: yield* DateTime.now, + attemptId: RunAttemptId.make("capacity-start-fails-stop-child"), + text: RESUME_PROMPT, + }), + ); + yield* Deferred.await(retrying); + const root = harness.events.find((event) => event.type === "provider_turn.updated"); + assert.equal(root?.type, "provider_turn.updated"); + if (root?.type !== "provider_turn.updated") return; + yield* TestClock.adjust("2 seconds"); + yield* Deferred.await(starting); + const stopping = yield* harness.runtime + .interruptTurn({ + providerThread: harness.providerThread, + providerTurnId: root.providerTurn.id, + }) + .pipe(Effect.forkChild({ startImmediately: true })); + yield* Deferred.succeed(release, undefined); + yield* Fiber.join(stopping); + assert.lengthOf(interrupts, 1); + assert.equal(harness.subagentUpdates().at(-1)?.subagent.status, "interrupted"); + }).pipe( + Effect.provide( + Layer.mergeAll(IdAllocator.layer, McpProviderSessions.layer, NodeServices.layer), + ), + ), + ), + ); + it.effect("names a late child spawned without a prompt in its wake", () => Effect.scoped( Effect.gen(function* () { @@ -6434,6 +7123,96 @@ describe("CodexAdapterV2 post-settle continuation", () => { ), ); + it.effect("Stop prevents a capacity retry while pausing its goal", () => + Effect.gen(function* () { + const nativeTurnId = "goal-capacity-stop"; + const prompt = "Keep going"; + const retrying = yield* Deferred.make(); + const pausing = yield* Deferred.make(); + const requests: Array = []; + const harness = yield* makeCodexReplayHarness( + makeCodexReplayTranscript({ + scenario: "goal-capacity-stop", + entries: [ + ...codexReplayPreamble({ nativeThreadId, nativeTurnId, prompt }), + notification("goal active", "thread/goal/updated", { + threadId: nativeThreadId, + turnId: nativeTurnId, + goal: codexGoal("active", 10), + }), + notification("capacity failure", "turn/completed", { + threadId: nativeThreadId, + turn: { + ...makeCodexReplayTurn({ id: nativeTurnId, status: "failed" }), + error: { + message: "Selected model is at capacity.", + codexErrorInfo: "serverOverloaded", + }, + }, + }), + { + type: "expect_outbound", + frame: { + id: 4, + method: "thread/goal/set", + params: { threadId: nativeThreadId, status: "paused" }, + }, + }, + { + type: "emit_inbound", + afterMs: 3_000, + frame: { id: 4, result: { goal: codexGoal("paused", 10) } }, + }, + ], + }), + (event) => + event.type === "turn_item.updated" && + event.turnItem.type === "error" && + event.turnItem.status === "running" + ? Deferred.succeed(retrying, undefined) + : Effect.void, + (method) => + Effect.gen(function* () { + requests.push(method); + if (method === "thread/goal/set") yield* Deferred.succeed(pausing, undefined); + }), + ); + yield* harness.runtime.startTurn( + makeCodexTestTurnInput({ + threadId: harness.threadId, + providerThread: harness.providerThread, + now: yield* DateTime.now, + attemptId: RunAttemptId.make("goal-capacity-stop"), + text: prompt, + }), + ); + yield* Deferred.await(retrying); + const stopping = yield* harness.runtime + .interruptTurn({ + providerThread: harness.providerThread, + providerTurnId: (yield* IdAllocator.IdAllocatorV2).derive.providerTurn({ + driver: CodexAdapterV2.CODEX_DRIVER_KIND, + nativeTurnId, + }), + }) + .pipe(Effect.forkScoped); + yield* Deferred.await(pausing); + yield* TestClock.adjust("3 seconds"); + assert.lengthOf( + requests.filter((method) => method === "turn/start"), + 1, + ); + yield* Fiber.join(stopping); + yield* harness.awaitTerminal(1); + assert.equal(harness.terminalEvents()[0]?.status, "interrupted"); + }).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 6dba639b63b0..74c4b46646e3 100644 --- a/apps/server/src/orchestration-v2/Adapters/CodexAdapterV2.ts +++ b/apps/server/src/orchestration-v2/Adapters/CodexAdapterV2.ts @@ -86,6 +86,7 @@ import * as Deferred from "effect/Deferred"; import * as Effect from "effect/Effect"; import * as Fiber from "effect/Fiber"; import * as FileSystem from "effect/FileSystem"; +import * as Function from "effect/Function"; import * as Layer from "effect/Layer"; import * as Option from "effect/Option"; import * as Path from "effect/Path"; @@ -152,6 +153,7 @@ import * as IdAllocator from "@t3tools/provider-core/server/IdAllocator"; import * as ProviderAdapter from "@t3tools/provider-core/server/ProviderAdapter"; const CODEX_PROVIDER = ProviderDriverKind.make("codex"); +const CODEX_CAPACITY_MAX_RETRIES = 5; export const CODEX_DRIVER_KIND = CODEX_PROVIDER; export const CODEX_DEFAULT_INSTANCE_ID = defaultInstanceIdForDriver(CODEX_DRIVER_KIND); @@ -1109,6 +1111,8 @@ function withCodexManagedFailure( } interface ActiveCodexTurnContext { + capacityRetryAttempt?: number; + capacityRetryPending?: Deferred.Deferred; closedCommandItemIds?: ReadonlySet; latestProviderFailure?: { readonly nativeMessage: string; @@ -1141,6 +1145,7 @@ interface ActiveCodexTurnContext { } interface ActiveCodexProviderRetry { + readonly context?: ActiveCodexTurnContext; readonly nativeMessage: string; readonly nativeCode: string | null; readonly retry: OrchestrationV2ProviderRetry; @@ -1149,6 +1154,30 @@ interface ActiveCodexProviderRetry { readonly itemOrdinal: number; } +function codexRetryTurnItem( + activeContext: ActiveCodexTurnContext, + state: ActiveCodexProviderRetry, + status: Parameters[0]["status"], + updatedAt: DateTime.Utc, +) { + // A continuation keeps the retry row attached to the native turn that failed. + const context = state.context ?? activeContext; + return makeProviderRetryTurnItem({ + driver: CODEX_PROVIDER, + threadId: context.projectionThreadId, + runId: context.projectionRunId, + nodeId: context.providerNodeId, + providerThreadId: context.providerThread.id, + providerTurnId: context.providerTurnId, + itemOrdinal: state.itemOrdinal, + failure: state.failure, + retry: state.retry, + status, + startedAt: state.startedAt, + updatedAt, + }); +} + interface TrackedRunningCommandItem { readonly id: string; readonly command: string; @@ -1815,12 +1844,23 @@ export const makeCodexAdapterV2 = Effect.fn("makeCodexAdapterV2")(function* ( const goalsByNativeThread = new Map(); const rootProviderThreads = new Map(); const goalHolds = new Map(); - const goalRuns = new Map>(); + const continuedRuns = new Map>(); + const capacityContinuations = new Map< + string, + { readonly context: ActiveCodexTurnContext; readonly retry: ActiveCodexProviderRetry } + >(); const goalActivations = new Map(); // Held runs must not complete while Stop resolves its target. const goalStops = new Map(); - const latestGoalTurnId = (providerTurnId: ProviderTurnId) => - goalRuns.get(providerTurnId)?.at(-1)?.providerTurnId ?? providerTurnId; + const latestRunTurnId = (providerTurnId: ProviderTurnId) => + continuedRuns.get(providerTurnId)?.at(-1)?.providerTurnId ?? providerTurnId; + const isContinuedRunDescendant = ( + candidate: ActiveCodexTurnContext, + context: ActiveCodexTurnContext, + ) => + (continuedRuns.get(context.providerTurnId) ?? [context]).some((ancestor) => + isDescendantCodexTurn(candidate, ancestor), + ); const emitProviderEvent = (event: ProviderAdapter.ProviderAdapterV2Event) => Queue.offer(events, event).pipe(Effect.asVoid); @@ -1882,20 +1922,7 @@ export const makeCodexAdapterV2 = Effect.fn("makeCodexAdapterV2")(function* ( yield* emitProviderEvent({ type: "turn_item.updated", driver: CODEX_PROVIDER, - turnItem: makeProviderRetryTurnItem({ - driver: CODEX_PROVIDER, - threadId: context.projectionThreadId, - runId: context.projectionRunId, - nodeId: context.providerNodeId, - providerThreadId: context.providerThread.id, - providerTurnId: context.providerTurnId, - itemOrdinal: providerRetry.itemOrdinal, - failure: providerRetry.failure, - retry: providerRetry.retry, - status: "completed", - startedAt: providerRetry.startedAt, - updatedAt, - }), + turnItem: codexRetryTurnItem(context, providerRetry, "completed", updatedAt), }); }); @@ -1934,6 +1961,24 @@ export const makeCodexAdapterV2 = Effect.fn("makeCodexAdapterV2")(function* ( fileChangeKindsByItem: new Map(), runningCompactionItemIds: new Set(), }; + const nativeThreadId = input.turnInput.providerThread.nativeThreadRef?.nativeId; + const continuation = + nativeThreadId == null ? undefined : capacityContinuations.get(nativeThreadId); + if (continuation !== undefined) { + context.capacityRetryAttempt = continuation.context.capacityRetryAttempt ?? 0; + yield* Ref.update(providerRetries, (current) => + new Map(current).set(providerTurnId, continuation.retry), + ); + const turns = continuedRuns.get(continuation.context.providerTurnId) ?? [ + continuation.context, + ]; + turns.push(context); + continuedRuns.set(continuation.context.providerTurnId, turns); + continuedRuns.set(providerTurnId, turns); + if (continuation.context.capacityRetryPending !== undefined) { + yield* Deferred.succeed(continuation.context.capacityRetryPending, context); + } + } yield* Ref.update(limitedTurnItems, (current) => { const next = new Map(current); next.delete(context.providerThread.id); @@ -1946,6 +1991,9 @@ export const makeCodexAdapterV2 = Effect.fn("makeCodexAdapterV2")(function* ( updated.set(input.nativeTurnId, context); return updated; }); + if (continuation !== undefined && nativeThreadId != null) { + capacityContinuations.delete(nativeThreadId); + } yield* emitProviderEvent({ type: "provider_turn.updated", driver: CODEX_PROVIDER, @@ -4380,10 +4428,12 @@ export const makeCodexAdapterV2 = Effect.fn("makeCodexAdapterV2")(function* ( nativeTurnId: payload.turn.id, startedAt: codexTimestamp(payload.turn.startedAt), }); - const goalRun = goalRuns.get(goalHold.context.providerTurnId) ?? [goalHold.context]; + const goalRun = continuedRuns.get(goalHold.context.providerTurnId) ?? [ + goalHold.context, + ]; goalRun.push(next); - goalRuns.set(goalHold.context.providerTurnId, goalRun); - goalRuns.set(next.providerTurnId, goalRun); + continuedRuns.set(goalHold.context.providerTurnId, goalRun); + continuedRuns.set(next.providerTurnId, goalRun); yield* Deferred.succeed(goalHold.next, next); return; } @@ -4493,6 +4543,7 @@ export const makeCodexAdapterV2 = Effect.fn("makeCodexAdapterV2")(function* ( previous?.itemOrdinal ?? (yield* resolveItemOrdinal(context, `terminal-failure:${context.providerTurnId}`)); const state: ActiveCodexProviderRetry = { + ...(previous?.context === undefined ? {} : { context: previous.context }), nativeMessage: payload.error.message, nativeCode: code, retry, @@ -4508,20 +4559,7 @@ export const makeCodexAdapterV2 = Effect.fn("makeCodexAdapterV2")(function* ( yield* emitProviderEvent({ type: "turn_item.updated", driver: CODEX_PROVIDER, - turnItem: makeProviderRetryTurnItem({ - driver: CODEX_PROVIDER, - threadId: context.projectionThreadId, - runId: context.projectionRunId, - nodeId: context.providerNodeId, - providerThreadId: context.providerThread.id, - providerTurnId: context.providerTurnId, - itemOrdinal, - failure, - retry, - status: "running", - startedAt: state.startedAt, - updatedAt, - }), + turnItem: codexRetryTurnItem(context, state, "running", updatedAt), }); }).pipe(Effect.orDie), ); @@ -5617,8 +5655,16 @@ export const makeCodexAdapterV2 = Effect.fn("makeCodexAdapterV2")(function* ( } : event; yield* emitProviderEvent(current); - for (const turn of goalRuns.get(context.providerTurnId) ?? []) { - goalRuns.delete(turn.providerTurnId); + const continued = continuedRuns.get(context.providerTurnId) ?? [context]; + if (current.status !== "completed") { + yield* Ref.update( + terminalizedNonCompletedNativeTurns, + (nativeTurns) => + new Set([...nativeTurns, ...continued.map((turn) => turn.nativeTurnId)]), + ); + } + for (const turn of continued) { + continuedRuns.delete(turn.providerTurnId); } const nativeThreadId = context.providerThread.nativeThreadRef?.nativeId; const rootProviderThread = @@ -5656,7 +5702,7 @@ export const makeCodexAdapterV2 = Effect.fn("makeCodexAdapterV2")(function* ( }) { const event = yield* makeRootTerminalEvent(input); const hasActiveDescendants = Array.from((yield* Ref.get(activeTurns)).values()).some( - (candidate) => isDescendantCodexTurn(candidate, input.context), + (candidate) => isContinuedRunDescendant(candidate, input.context), ); // Failed roots leave native children working and must settle without them. // Interrupted roots wait for descendants because Stop interrupts the whole lineage. @@ -5781,7 +5827,7 @@ export const makeCodexAdapterV2 = Effect.fn("makeCodexAdapterV2")(function* ( for (const [nativeTurnId, deferred] of current) { if ( !activeTurnContexts.some((candidate) => - isDescendantCodexTurn(candidate, deferred.context), + isContinuedRunDescendant(candidate, deferred.context), ) ) { updated.delete(nativeTurnId); @@ -5803,8 +5849,10 @@ export const makeCodexAdapterV2 = Effect.fn("makeCodexAdapterV2")(function* ( readonly completedAt: DateTime.Utc; readonly failureMessage?: string; readonly failureErrorInfo?: CodexErrorInfo; + readonly continueRun?: boolean; }) => - turnTerminalizationPermit.withPermits(1)( + // Capacity continuation already holds the permit across finalization and restart. + (input.continueRun ? Function.identity : turnTerminalizationPermit.withPermits(1))( Effect.gen(function* () { const current = (yield* Ref.get(activeTurns)).get(input.nativeTurnId); if (current !== input.context) { @@ -5822,31 +5870,33 @@ export const makeCodexAdapterV2 = Effect.fn("makeCodexAdapterV2")(function* ( }); if ( providerRetry !== undefined && - (input.status !== "failed" || input.context.subagent !== null) + !input.continueRun && + (providerRetry.context !== undefined || + input.status !== "failed" || + input.context.subagent !== null) ) { yield* emitProviderEvent({ type: "turn_item.updated", driver: CODEX_PROVIDER, - turnItem: makeProviderRetryTurnItem({ - driver: CODEX_PROVIDER, - threadId: input.context.projectionThreadId, - runId: input.context.projectionRunId, - nodeId: input.context.providerNodeId, - providerThreadId: input.context.providerThread.id, - providerTurnId: input.context.providerTurnId, - itemOrdinal: providerRetry.itemOrdinal, - failure: providerRetry.failure, - retry: providerRetry.retry, - status: providerTurnStatusToTerminal(input.status), - startedAt: providerRetry.startedAt, - updatedAt: input.completedAt, - }), + turnItem: codexRetryTurnItem( + input.context, + providerRetry, + input.status === "failed" && + providerRetry.context !== input.context && + providerRetry.context !== undefined + ? "interrupted" + : providerTurnStatusToTerminal(input.status), + input.completedAt, + ), }); } - if (input.status !== "completed") { + if (input.status !== "completed" && !input.continueRun) { yield* Ref.update(terminalizedNonCompletedNativeTurns, (current) => { const updated = new Set(current); - updated.add(input.nativeTurnId); + for (const turn of continuedRuns.get(input.context.providerTurnId) ?? [ + input.context, + ]) + updated.add(turn.nativeTurnId); return updated; }); } @@ -5967,11 +6017,14 @@ export const makeCodexAdapterV2 = Effect.fn("makeCodexAdapterV2")(function* ( // Only direct children wake an idle root turn; grandchildren report to their // own parent, and running root turns receive results natively. const parent = input.context.subagent.parentContext; + const parentNativeThreadId = parent.providerThread.nativeThreadRef?.nativeId; if ( continuationRequests !== undefined && parent.subagent === null && (input.status === "completed" || input.status === "failed") && !(yield* Ref.get(terminalizedNonCompletedNativeTurns)).has(parent.nativeTurnId) && + (parentNativeThreadId == null || + !capacityContinuations.has(parentNativeThreadId)) && ![...(yield* Ref.get(activeTurns)).values()].some( (active) => active.subagent === null && @@ -6017,7 +6070,7 @@ export const makeCodexAdapterV2 = Effect.fn("makeCodexAdapterV2")(function* ( input.completedAt, input.status === "interrupted" || input.status === "failed", ); - if (input.context.subagent === null) { + if (input.context.subagent === null && !input.continueRun) { yield* emitOrDeferRootTerminal({ ...input, ...(providerRetry === undefined ? {} : { providerRetry }), @@ -6051,6 +6104,9 @@ export const makeCodexAdapterV2 = Effect.fn("makeCodexAdapterV2")(function* ( if (input.context.nativeStartReady !== undefined) { yield* Deferred.succeed(input.context.nativeStartReady, undefined); } + if (!input.continueRun && input.context.capacityRetryPending !== undefined) { + yield* Deferred.succeed(input.context.capacityRetryPending, undefined); + } yield* flushReadyRootTerminals(); if (!retainSettledContext && !interruptInProgress) { yield* Ref.update(runningCommandItemsByTurn, (current) => { @@ -6092,6 +6148,141 @@ export const makeCodexAdapterV2 = Effect.fn("makeCodexAdapterV2")(function* ( }), ); + // Codex can end an overloaded turn without retry advice. Continue its + // persisted history with empty input instead of resending the user message. + const retryCapacityFailure = Effect.fn("CodexAdapterV2.retryCapacityFailure")(function* ( + terminal: Parameters[0], + ) { + const context = terminal.context; + const code = + terminal.failureErrorInfo == null + ? context.latestProviderFailure?.nativeCode + : codexErrorInfoCode(terminal.failureErrorInfo); + if ( + terminal.status !== "failed" || + code !== "serverOverloaded" || + context.subagent !== null || + (context.input.message.attachments.length === 0 && + context.input.message.text.trim().toLowerCase() === "/compact") + ) + return false; + + return yield* turnTerminalizationPermit.withPermits(1)( + Effect.gen(function* () { + if ((yield* Ref.get(activeTurns)).get(terminal.nativeTurnId) !== context) return true; + if (context.capacityRetryPending !== undefined) return true; + if ( + (context.capacityRetryAttempt ?? 0) >= CODEX_CAPACITY_MAX_RETRIES || + (yield* Ref.get(interruptingNativeTurns)).has(terminal.nativeTurnId) + ) + return false; + + const attempt = (context.capacityRetryAttempt ?? 0) + 1; + context.capacityRetryAttempt = attempt; + const pending = yield* Deferred.make(); + context.capacityRetryPending = pending; + const retryDelayMs = Math.min(2_000 * 2 ** (attempt - 1), 30_000); + const now = yield* DateTime.now; + const previous = (yield* Ref.get(providerRetries)).get(context.providerTurnId); + const state: ActiveCodexProviderRetry = { + context: previous?.context ?? context, + nativeMessage: + terminal.failureMessage ?? + context.latestProviderFailure?.nativeMessage ?? + "Selected model is at capacity.", + nativeCode: code, + failure: makeProviderFailure({ + message: + context.latestProviderFailure?.failure.message ?? terminal.failureMessage, + code, + class: "provider_error", + retryable: true, + }), + retry: { attempt, maxAttempts: CODEX_CAPACITY_MAX_RETRIES, retryDelayMs }, + itemOrdinal: + previous?.itemOrdinal ?? + (yield* resolveItemOrdinal( + context, + `terminal-failure:${context.providerTurnId}`, + )), + startedAt: previous?.startedAt ?? now, + }; + yield* Ref.update(providerRetries, (current) => + new Map(current).set(context.providerTurnId, state), + ); + yield* emitProviderEvent({ + type: "turn_item.updated", + driver: CODEX_PROVIDER, + turnItem: codexRetryTurnItem(context, state, "running", now), + }); + yield* Effect.sleep(retryDelayMs).pipe( + Effect.andThen( + Effect.gen(function* () { + const nativeThreadId = yield* getNativeThreadId(context.providerThread); + const claimed = yield* turnTerminalizationPermit.withPermits(1)( + Effect.gen(function* () { + if ( + (yield* Ref.get(activeTurns)).get(terminal.nativeTurnId) !== context || + goalStops.has(nativeThreadId) || + (yield* Ref.get(interruptingNativeTurns)).has(terminal.nativeTurnId) + ) + return false; + capacityContinuations.set(nativeThreadId, { context, retry: state }); + yield* finalizeCodexTurn({ ...terminal, continueRun: true }); + return true; + }), + ); + if (!claimed) return; + // Notifications before the RPC response also take the terminalization permit. + yield* startNativeTurn( + { + ...context.input, + providerTurnOrdinal: context.providerTurnOrdinal + 1, + }, + [], + ).pipe( + Effect.catch((cause) => + Effect.gen(function* () { + if (yield* Deferred.isDone(pending)) return; + yield* Effect.logWarning( + "orchestration-v2.codex-capacity-retry-start-failed", + { + providerTurnId: context.providerTurnId, + cause, + }, + ); + yield* emitProviderEvent({ + type: "turn_item.updated", + driver: CODEX_PROVIDER, + turnItem: codexRetryTurnItem( + context, + state, + "failed", + yield* DateTime.now, + ), + }); + yield* Deferred.succeed(pending, undefined); + yield* turnTerminalizationPermit.withPermits(1)( + emitOrDeferRootTerminal({ ...terminal, providerRetry: state }), + ); + }), + ), + Effect.ensuring( + Effect.sync(() => { + if (capacityContinuations.get(nativeThreadId)?.context === context) + capacityContinuations.delete(nativeThreadId); + }), + ), + ); + }), + ), + Effect.forkIn(scope, { startImmediately: true }), + ); + return true; + }), + ); + }); + yield* client.handleServerNotification("turn/completed", (payload) => Effect.gen(function* () { const context = (yield* Ref.get(activeTurns)).get(payload.turn.id); @@ -6104,7 +6295,7 @@ export const makeCodexAdapterV2 = Effect.fn("makeCodexAdapterV2")(function* ( (yield* Ref.get(interruptingNativeTurns)).has(payload.turn.id) ? "interrupted" : nativeStatus; - yield* finalizeCodexTurn({ + const terminal = { context, nativeTurnId: payload.turn.id, status, @@ -6117,7 +6308,8 @@ export const makeCodexAdapterV2 = Effect.fn("makeCodexAdapterV2")(function* ( ? {} : { failureErrorInfo: payload.turn.error.codexErrorInfo }), }), - }); + }; + if (!(yield* retryCapacityFailure(terminal))) yield* finalizeCodexTurn(terminal); }), ); @@ -6231,10 +6423,10 @@ export const makeCodexAdapterV2 = Effect.fn("makeCodexAdapterV2")(function* ( goalStops.set(goalThreadId, (goalStops.get(goalThreadId) ?? 0) + 1); return yield* Effect.gen(function* () { // The run can settle while the pause is in flight, which drops its record. - const requestedTurnId = latestGoalTurnId(turnInput.providerTurnId); + const requestedTurnId = latestRunTurnId(turnInput.providerTurnId); const heldBefore = goalHolds.get(goalThreadId)?.context; const turnsBefore = [ - ...(goalRuns.get(requestedTurnId) ?? []), + ...(continuedRuns.get(requestedTurnId) ?? []), ...Array.from((yield* Ref.get(activeTurns)).values()).filter( (context) => context.providerTurnId === requestedTurnId, ), @@ -6258,13 +6450,13 @@ export const makeCodexAdapterV2 = Effect.fn("makeCodexAdapterV2")(function* ( } return yield* turnTerminalizationPermit.withPermits(1)( Effect.gen(function* () { - const providerTurnId = latestGoalTurnId(turnInput.providerTurnId); + const providerTurnId = latestRunTurnId(turnInput.providerTurnId); const hold = goalHolds.get(goalThreadId); const held = hold !== undefined && hold.context.providerTurnId === providerTurnId; const goalTurns: ReadonlyArray = Array.from( new Set([ ...turnsBefore, - ...(goalRuns.get(providerTurnId) ?? []), + ...(continuedRuns.get(providerTurnId) ?? []), ...(held ? [hold.context] : []), ]), ); @@ -6319,6 +6511,8 @@ export const makeCodexAdapterV2 = Effect.fn("makeCodexAdapterV2")(function* ( return next; }); const started = yield* client.request("turn/start", turnStartParams); + // A turn/started notification may have registered and completed this turn already. + if ((yield* Ref.get(pendingRootTurns)).get(threadId) !== turnInput) return; yield* registerRootTurn({ turnInput, nativeTurnId: started.turn.id, @@ -6331,6 +6525,7 @@ export const makeCodexAdapterV2 = Effect.fn("makeCodexAdapterV2")(function* ( Effect.ensuring( Effect.flatMap(getNativeThreadId(turnInput.providerThread), (threadId) => Ref.update(pendingRootTurns, (current) => { + if (current.get(threadId) !== turnInput) return current; const updated = new Map(current); updated.delete(threadId); return updated; @@ -6764,15 +6959,22 @@ export const makeCodexAdapterV2 = Effect.fn("makeCodexAdapterV2")(function* ( steerTurn: Effect.fn("CodexAdapterV2.steerTurn")( function* (turnInput) { const threadId = yield* getNativeThreadId(turnInput.providerThread); - const latestTurnId = latestGoalTurnId(turnInput.providerTurnId); + const latestTurnId = latestRunTurnId(turnInput.providerTurnId); const held = heldGoalTurn(turnInput.providerThread, latestTurnId); const providerTurnId = held === undefined ? latestTurnId : ((yield* Deferred.await(held.hold.next))?.providerTurnId ?? latestTurnId); - const activeTurn = Array.from((yield* Ref.get(activeTurns)).values()).find( + let activeTurn = Array.from((yield* Ref.get(activeTurns)).values()).find( (candidate) => candidate.providerTurnId === providerTurnId, ); + const startingRetry = capacityContinuations.get(threadId)?.context; + if (activeTurn === undefined && startingRetry?.providerTurnId === providerTurnId) { + activeTurn = startingRetry; + } + while (activeTurn?.capacityRetryPending !== undefined) { + activeTurn = yield* Deferred.await(activeTurn.capacityRetryPending); + } if (activeTurn === undefined) { return yield* toProtocolError( `Provider turn ${turnInput.providerTurnId} is not active and cannot be steered.`, @@ -6822,6 +7024,24 @@ export const makeCodexAdapterV2 = Effect.fn("makeCodexAdapterV2")(function* ( ), interruptTurn: Effect.fn("CodexAdapterV2.interruptTurn")( function* (requestedInput) { + const nativeThreadId = requestedInput.providerThread.nativeThreadRef?.nativeId; + const capacityRetry = + nativeThreadId == null + ? undefined + : capacityContinuations.get(nativeThreadId)?.context; + const startingRetry = + capacityRetry?.providerTurnId === latestRunTurnId(requestedInput.providerTurnId) + ? capacityRetry + : undefined; + if (startingRetry?.capacityRetryPending !== undefined) { + const started = yield* Deferred.await(startingRetry.capacityRetryPending).pipe( + Effect.timeoutOption("10 seconds"), + ); + if (Option.isNone(started)) + return yield* toProtocolError( + "Codex did not start the capacity retry within 10 seconds; Stop could not be delivered.", + ); + } const { turnInput, goalTurns } = yield* resolveGoalStopTarget(requestedInput); const [activeTurnContexts, settledTurnContexts] = yield* turnTerminalizationPermit.withPermits(1)( @@ -6842,7 +7062,11 @@ export const makeCodexAdapterV2 = Effect.fn("makeCodexAdapterV2")(function* ( ) : undefined); const lineageRoots = Array.from( - new Set([...(activeTurn === undefined ? [] : [activeTurn]), ...goalTurns]), + new Set([ + ...(activeTurn === undefined ? [] : [activeTurn]), + ...(startingRetry === undefined ? [] : [startingRetry]), + ...goalTurns, + ]), ); const inLineage = (context: ActiveCodexTurnContext) => lineageRoots.some( @@ -7053,6 +7277,15 @@ export const makeCodexAdapterV2 = Effect.fn("makeCodexAdapterV2")(function* ( yield* Effect.gen(function* () { for (const target of interruptTargets) { const context = target.context; + if (context.capacityRetryPending !== undefined) { + yield* finalizeCodexTurn({ + context, + nativeTurnId: context.nativeTurnId, + status: "interrupted", + completedAt: yield* DateTime.now, + }); + continue; + } // Null start timestamps ACK queued turns; interruption must wait for turn/started. if (context.nativeStartReady !== undefined) { const ready = yield* Deferred.await(context.nativeStartReady).pipe( @@ -7193,7 +7426,22 @@ export const makeCodexAdapterV2 = Effect.fn("makeCodexAdapterV2")(function* ( ); }, (effect, requestedInput) => - effect.pipe( + Effect.acquireUseRelease( + Effect.sync(() => { + const nativeThreadId = requestedInput.providerThread.nativeThreadRef?.nativeId; + if (nativeThreadId != null) + goalStops.set(nativeThreadId, (goalStops.get(nativeThreadId) ?? 0) + 1); + return nativeThreadId; + }), + () => effect, + (nativeThreadId) => + Effect.sync(() => { + if (nativeThreadId == null) return; + const stops = (goalStops.get(nativeThreadId) ?? 1) - 1; + if (stops > 0) goalStops.set(nativeThreadId, stops); + else goalStops.delete(nativeThreadId); + }), + ).pipe( Effect.mapError( (cause) => new ProviderAdapter.ProviderAdapterInterruptError({ diff --git a/apps/server/src/orchestration-v2/ProviderTurnControlService.ts b/apps/server/src/orchestration-v2/ProviderTurnControlService.ts index 3a5bc04ecba2..e1dbac21bc52 100644 --- a/apps/server/src/orchestration-v2/ProviderTurnControlService.ts +++ b/apps/server/src/orchestration-v2/ProviderTurnControlService.ts @@ -127,7 +127,11 @@ export const layer: Layer.Layer< const interruptProviderThread = targetsRecordedSession ? providerThread : { ...providerThread, providerSessionId: input.providerSessionId }; - if (providerTurn.status !== "running") { + const continuingRun = + input.operation === "steer" && + context.run?.status === "running" && + context.run.activeAttemptId === providerTurn.runAttemptId; + if (providerTurn.status !== "running" && !continuingRun) { if (input.operation === "steer") { return yield* new ProviderTurnControlError({ threadId: input.threadId, diff --git a/apps/server/src/orchestration-v2/SteeringCompletion.integration.test.ts b/apps/server/src/orchestration-v2/SteeringCompletion.integration.test.ts index 4c2d6712c5ad..7481dcaa41b2 100644 --- a/apps/server/src/orchestration-v2/SteeringCompletion.integration.test.ts +++ b/apps/server/src/orchestration-v2/SteeringCompletion.integration.test.ts @@ -42,6 +42,7 @@ it.effect.each( [true, "without native steering"], [true, "with interrupting native steering"], [true, "settled only"], + [false, "during continuation"], ] as const ).map(([mailbox, timing]) => ({ mailbox, @@ -135,7 +136,7 @@ it.effect.each( steerTurn: (turn) => Effect.gen(function* () { steerCalls += 1; - if (timing === "after delivery") return; + if (timing === "after delivery" || timing === "during continuation") return; yield* Deferred.succeed(steerEntered, undefined); yield* Deferred.await(rejectSteer); return yield* new ProviderAdapter.ProviderAdapterSteerRunError({ @@ -284,6 +285,29 @@ it.effect.each( : {}), }); if (timing !== "before dispatch") yield* dispatchSteer; + if (timing === "during continuation") { + const projection = yield* orchestrator.getThreadProjection(threadId); + const turn = projection.providerTurns[0]!; + const failed = yield* watch( + (event) => event.type === "provider-turn.updated" && event.payload.status === "failed", + ); + yield* Queue.offer(events, { + type: "provider_turn.updated", + driver, + providerTurn: { ...turn, status: "failed", completedAt: yield* DateTime.now }, + }); + yield* Fiber.join(failed); + yield* worker.drain(); + assert.equal(steerCalls, 1); + assert.lengthOf(started, 1); + const continued = yield* orchestrator.getThreadProjection(threadId); + assert.equal(continued.runs[0]?.status, "running"); + assert.equal( + continued.messages.find((message) => message.id === messageId)?.runId, + first.runId, + ); + return; + } if (timing === "after delivery") yield* worker.drain(); const delivery = timing === "during delivery" ? yield* worker.runOnce.pipe(Effect.forkScoped) : null; From 07242abff3cf2004a703ae631eae1416574a5f30 Mon Sep 17 00:00:00 2001 From: Yash Singh Date: Tue, 27 Oct 2026 14:44:59 -0500 Subject: [PATCH 03/19] fix(server): recover handoffs when target context is exhausted (#25610) --- .../Adapters/AcpAdapterV2.test.ts | 112 +++- .../Adapters/ClaudeAdapterV2.test.ts | 44 +- .../Adapters/ClaudeAdapterV2.ts | 16 +- .../ProviderTurnStartService.ts | 600 ++++++++++-------- .../ProviderSwitch.integration.test.ts | 181 +++++- docs/user/portable-handoffs.md | 10 +- packages/provider-acp/src/server/adapter.ts | 153 +++-- .../src/server/ProviderAdapter.ts | 4 +- .../provider-core/src/server/failure.test.ts | 22 +- packages/provider-core/src/server/failure.ts | 16 +- 10 files changed, 816 insertions(+), 342 deletions(-) diff --git a/apps/server/src/orchestration-v2/Adapters/AcpAdapterV2.test.ts b/apps/server/src/orchestration-v2/Adapters/AcpAdapterV2.test.ts index a919726c9a56..92eb5b1047ff 100644 --- a/apps/server/src/orchestration-v2/Adapters/AcpAdapterV2.test.ts +++ b/apps/server/src/orchestration-v2/Adapters/AcpAdapterV2.test.ts @@ -909,7 +909,11 @@ describe("AcpAdapterV2", () => { flavor: { driver: ACP_TEST_DRIVER, capabilities: AcpProviderCapabilitiesV2, - makeRuntime: makeMockRuntime({ childProcessSpawner, mockAgentPath, protocolEvents }), + makeRuntime: makeMockRuntime({ + childProcessSpawner, + mockAgentPath, + protocolEvents, + }), }, selfInvocation, }); @@ -2406,6 +2410,99 @@ describe("AcpAdapterV2", () => { }).pipe(Effect.provide(layerTest), Effect.scoped), ); + it.live("creates a fresh ACP identity when recovering an exhausted target", () => + Effect.gen(function* () { + const childProcessSpawner = yield* ChildProcessSpawner.ChildProcessSpawner; + const path = yield* Path.Path; + const selfInvocation = yield* resolveSelfInvocation(); + const mockAgentPath = yield* path.fromFileUrl( + new URL("../../../scripts/acp-mock-agent.ts", import.meta.url), + ); + const protocolEvents = yield* Queue.unbounded(); + const instanceId = ProviderInstanceId.make("acp-test-exhausted-target"); + const adapter = yield* makeAcpAdapterV2({ + instanceId, + flavor: { + driver: ACP_TEST_DRIVER, + capabilities: AcpProviderCapabilitiesV2, + makeRuntime: makeMockRuntime({ + childProcessSpawner, + mockAgentPath, + protocolEvents, + wrapRuntime: (runtime, ordinal) => ({ + ...runtime, + start: () => + runtime.start().pipe( + Effect.map((started) => ({ + ...started, + sessionId: `mock-recovery-session-${ordinal}`, + })), + ), + }), + }), + }, + selfInvocation, + }); + const threadId = ThreadId.make("thread-acp-exhausted-target"); + const runtimePolicy = ProviderAdapter.ProviderAdapterV2RuntimePolicy.make({ + runtimeMode: "full-access", + interactionMode: "default", + cwd: process.cwd(), + }); + const modelSelection = { instanceId, model: "default" } as const; + const runtime = yield* adapter.openSession({ + threadId, + providerSessionId: ProviderSessionId.make("provider-session-acp-exhausted-target"), + modelSelection, + runtimePolicy, + }); + const original = yield* runtime.ensureThread({ threadId, modelSelection, runtimePolicy }); + yield* runtime.startTurn( + makeTurnInput({ + threadId, + providerThread: original, + instanceId, + runtimePolicy, + now: yield* DateTime.now, + ordinal: 1, + }), + ); + yield* runtime.events.pipe( + Stream.takeUntil((event) => event.type === "turn.terminal"), + Stream.runDrain, + ); + const replacement = yield* runtime.ensureThread({ + threadId, + modelSelection, + runtimePolicy, + existingProviderThread: { ...original, nativeThreadRef: null }, + forceNewNativeThread: true, + }); + assert.notEqual(replacement.nativeThreadRef?.nativeId, original.nativeThreadRef?.nativeId); + assert.equal(replacement.id, original.id); + assert.equal(replacement.nativeMetadata?.itemIdentityVersion, 2); + yield* runtime.startTurn( + makeTurnInput({ + threadId, + providerThread: replacement, + instanceId, + runtimePolicy, + now: yield* DateTime.now, + ordinal: 2, + }), + ); + const events = yield* runtime.events.pipe( + Stream.takeUntil((event) => event.type === "turn.terminal"), + Stream.runCollect, + ); + assert.isTrue( + events.some((event) => event.type === "turn.terminal" && event.status === "completed"), + ); + const methods = yield* pollProtocolMethods(protocolEvents); + assert.equal(methods.filter((method) => method === "session/new").length, 2); + }).pipe(Effect.provide(layerTest), Effect.scoped), + ); + it.live("preserves new-session fallback when an eager ACP session load is stale", () => Effect.gen(function* () { const childProcessSpawner = yield* ChildProcessSpawner.ChildProcessSpawner; @@ -5240,6 +5337,19 @@ describe("AcpAdapterV2", () => { yield* hasPendingBackgroundWork, "carryover live subagent must pin hasPendingBackgroundWork after root settle", ); + const refusedReplacement = yield* runtime + .ensureThread({ + threadId, + modelSelection, + runtimePolicy, + existingProviderThread: { ...providerThread, nativeThreadRef: null }, + forceNewNativeThread: true, + }) + .pipe(Effect.flip); + assert.equal(refusedReplacement._tag, "ProviderAdapterEnsureThreadError"); + const preserved = yield* runtime.ensureThread({ threadId, modelSelection, runtimePolicy }); + assert.equal(preserved.nativeThreadRef?.nativeId, providerThread.nativeThreadRef?.nativeId); + assert.isTrue(yield* hasPendingBackgroundWork); assert.isDefined(sessionUpdateHandler, "session update handler must be wired"); subagentPhase = "complete"; diff --git a/apps/server/src/orchestration-v2/Adapters/ClaudeAdapterV2.test.ts b/apps/server/src/orchestration-v2/Adapters/ClaudeAdapterV2.test.ts index 8de0f78af88c..470e04a384dd 100644 --- a/apps/server/src/orchestration-v2/Adapters/ClaudeAdapterV2.test.ts +++ b/apps/server/src/orchestration-v2/Adapters/ClaudeAdapterV2.test.ts @@ -7784,7 +7784,7 @@ describe("ClaudeAdapterV2 background wake turns", () => { ), ); - it.effect("stops background work after the turn settled", () => + it.effect("preserves settled background work until an explicit stop", () => Effect.scoped( Effect.gen(function* () { let closes = 0; @@ -7813,6 +7813,29 @@ describe("ClaudeAdapterV2 background wake turns", () => { assert.equal(settledThread?.pendingBackgroundTasks?.[0]?.taskId, WAKE_TASK_ID); assert.isTrue(yield* harness.hasPendingBackgroundWork); + const replacementAttempt = yield* harness.runtime + .startTurn( + makeClaudeTestTurnInput({ + threadId: harness.threadId, + providerThread: { + ...harness.providerThread, + nativeThreadRef: { + ...harness.providerThread.nativeThreadRef!, + nativeId: "replacement-session", + }, + }, + now, + attemptId: RunAttemptId.make("attempt-claude-background-replacement"), + text: "Continue in a fresh session.", + attachments: [], + providerTurnOrdinal: 2, + }), + ) + .pipe(Effect.result); + assert.equal(replacementAttempt._tag, "Failure"); + assert.equal(closes, 0); + assert.isTrue(yield* harness.hasPendingBackgroundWork); + yield* harness.runtime.interruptTurn({ providerThread: settledThread ?? harness.providerThread, providerTurnId: harness.terminalEvents()[0]!.providerTurnId, @@ -12240,6 +12263,25 @@ describe("ClaudeAdapterV2 background wake turns", () => { yield* changes.until(() => terminals().length === 1); const settledTurn = terminals()[0]!; + assert.isFalse(yield* runtime.hasPendingBackgroundWorkForThread!(providerThread)); + assert.isTrue( + yield* runtime.hasPendingBackgroundWorkForThread!(providerThread, { + includeSubagents: true, + }), + ); + assert.isFalse( + yield* runtime.hasPendingBackgroundWorkForThread!( + { + ...providerThread, + nativeThreadRef: { + ...providerThread.nativeThreadRef!, + nativeId: "sibling-native-session", + }, + }, + { includeSubagents: true }, + ), + ); + const alternateModel = CLAUDE_TEST_RESTART_MODEL_SELECTION; const switchTurn = (attempt: string) => runtime.startTurn( diff --git a/apps/server/src/orchestration-v2/Adapters/ClaudeAdapterV2.ts b/apps/server/src/orchestration-v2/Adapters/ClaudeAdapterV2.ts index b66e85e14d7c..e1ff63efe044 100644 --- a/apps/server/src/orchestration-v2/Adapters/ClaudeAdapterV2.ts +++ b/apps/server/src/orchestration-v2/Adapters/ClaudeAdapterV2.ts @@ -9692,10 +9692,11 @@ export const makeClaudeAdapterV2 = Effect.fn("makeClaudeAdapterV2")(function* ( return existing; } - // Selection changes kill CLI background work; refuse until it finishes or Stop closes the process. + // Replacing this thread's process kills its background work, even with a new native identity. if ( existing !== null && - existing.nativeThreadId === nativeThreadId && + (existing.nativeThreadId === nativeThreadId || + existing.lastTurn?.input.threadId === turnInput.threadId) && !existing.stopping && (yield* liveProcessRunsBackgroundWork(existing)) ) { @@ -10560,11 +10561,20 @@ export const makeClaudeAdapterV2 = Effect.fn("makeClaudeAdapterV2")(function* ( }), hasPendingBackgroundWorkForThread: Effect.fn( "ClaudeAdapterV2.hasPendingBackgroundWorkForThread", - )(function* (providerThread) { + )(function* (providerThread, options) { const nativeThreadId = providerThread.nativeThreadRef?.nativeId; if (nativeThreadId === undefined || nativeThreadId === null) { return false; } + if (options?.includeSubagents) { + const live = yield* Ref.get(queryContext); + if ( + live?.nativeThreadId === nativeThreadId && + !live.stopping && + (yield* liveProcessRunsBackgroundWork(live)) + ) + return true; + } return ( rosterForNativeThread( yield* Ref.get(pendingBackgroundTasksByNativeThread), diff --git a/apps/server/src/orchestration-v2/ProviderTurnStartService.ts b/apps/server/src/orchestration-v2/ProviderTurnStartService.ts index 049f8e9a692d..2e0eee3828ea 100644 --- a/apps/server/src/orchestration-v2/ProviderTurnStartService.ts +++ b/apps/server/src/orchestration-v2/ProviderTurnStartService.ts @@ -4,6 +4,7 @@ import { CommandId, latestProviderTurnForAttempt, type OrchestrationV2DomainEvent, + type OrchestrationV2ContextHandoff, type OrchestrationV2ExecutionNode, type OrchestrationV2ProviderThread, type OrchestrationV2Run, @@ -48,6 +49,7 @@ import * as IdAllocator from "@t3tools/provider-core/server/IdAllocator"; import * as ProjectionStore from "./ProjectionStore.ts"; import * as ProviderSessionManager from "./ProviderSessionManager.ts"; import { + ContextHandoffBackgroundWorkError, knownProviderFailureMessage, makeProviderFailure, } from "@t3tools/provider-core/server/failure"; @@ -754,6 +756,279 @@ export const layer: Layer.Layer< }); let effectiveHandoffs = handoffs; let resumeFellBack = false; + const userText = projectComposerContextForProvider({ + 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, + projection.runs, + projection.providerTurns, + projection.attempts, + ); + const restartCancelledWork = pendingRestartCancelledBackgroundWork({ + runs: projection.runs, + providerTurns: projection.providerTurns, + compactionMessageIds: new Set( + projection.messages + .filter( + (candidate) => + isCompactionCommandMessage(candidate) || + (goalDriver && parseGoalCommand(candidate) !== null), + ) + .map((candidate) => candidate.id), + ), + run, + attempts: projection.attempts, + }); + const restartNote = + restartCancelledWork.length === 0 + ? "" + : restartCancelledBackgroundWorkNote(restartCancelledWork); + const tokenCap = yield* handoffTokenCapConfig.pipe( + Effect.orElseSucceed(() => DEFAULT_HANDOFF_TOKEN_CAP), + ); + const handoffContextForThread = (loadedProviderThread: OrchestrationV2ProviderThread) => { + const measuredContext = latestNativeContextUsage(projection, providerThread); + const previousSelection = + measuredContext?.modelSelection ?? + projection.runs.findLast( + (source) => + source.ordinal < run.ordinal && + source.startedAt !== null && + source.providerThreadId === providerThread.id, + )?.modelSelection; + const sameSelection = + previousSelection === undefined || + modelSelectionsEqual(previousSelection, run.modelSelection); + const sameNativeThread = + loadedProviderThread.nativeThreadRef?.nativeId === + providerThread.nativeThreadRef?.nativeId; + const threadUsage = loadedProviderThread.contextUsage ?? providerThread.contextUsage; + const previousUsage = measuredContext + ? { ...threadUsage, ...measuredContext.usage } + : threadUsage; + const reuseTelemetry = + sameSelection || + (previousSelection !== undefined && + session.canReuseContextUsage?.(previousSelection, run.modelSelection) === true); + const knownModelWindow = session.getModelContextWindow?.( + run.modelSelection, + resolvedRuntimePolicy.cwd, + ); + const handoffUsage = contextUsageForHandoff({ + sameNativeThread, + sameSelection, + reuseTelemetry, + previousUsage, + knownModelWindow, + }); + const settledHandoffs = projection.contextHandoffs.filter( + (handoff) => + handoff.toProviderThreadId === providerThread.id && + handoff.delivery?.nativeThreadId === loadedProviderThread.nativeThreadRef?.nativeId && + handoff.delivery?.status !== "pending", + ); + const deliveredItemIds = new Set( + settledHandoffs.flatMap((handoff) => handoff.delivery?.itemIds ?? []), + ); + const coveredItemIds = new Set([ + ...deliveredItemIds, + ...settledHandoffs.flatMap((handoff) => handoff.delivery?.omittedItemIds ?? []), + ]); + const deliveredAttemptIds = new Set( + projection.providerTurns.map((turn) => turn.runAttemptId), + ); + const acceptedAttempts = projection.attempts.filter( + (source) => + source.providerThreadId === providerThread.id && deliveredAttemptIds.has(source.id), + ); + const nativeInputRunIds = new Set( + acceptedAttempts + .filter( + (source) => + source.nativeThreadId !== undefined && + source.nativeThreadId === loadedProviderThread.nativeThreadRef?.nativeId, + ) + .map((source) => source.runId), + ); + const legacyInputRunIds = new Set( + acceptedAttempts + .filter((source) => source.nativeThreadId === undefined) + .map((source) => source.runId), + ); + const legacyRecoveredRunIds = new Set( + projection.runs + .filter( + (source) => + source.providerThreadId === providerThread.id && + settledHandoffs.some( + (handoff) => + handoff.strategy === "full_thread_summary" && + handoff.fromProviderThreadIds.includes(providerThread.id) && + source.ordinal >= handoff.coveredRunOrdinals.from && + source.ordinal <= handoff.coveredRunOrdinals.to, + ), + ) + .map((source) => source.id), + ); + // Row identity survives recovery; only accepted native input and delivered history occupy the new session. + const nativeContextEstimate = Effect.gen(function* () { + return sameNativeThread + ? (yield* projectionStore.getTurnStartHistory(input.threadId)).reduce((sum, item) => { + const acceptedNatively = + item.providerThreadId === providerThread.id && + item.runId !== null && + (nativeInputRunIds.has(item.runId) || + (legacyInputRunIds.has(item.runId) && + !coveredItemIds.has(item.id) && + !legacyRecoveredRunIds.has(item.runId))); + if (item.runId === run.id || (!acceptedNatively && !deliveredItemIds.has(item.id))) + return sum; + const historical = historicalMessage(item); + const nativeAttachments = + item.type === "user_message" && acceptedNatively + ? attachmentTokenAllowance(item.attachments) + : 0; + return ( + sum + + (historical === null ? 0 : Buffer.byteLength(historical.text)) + + nativeAttachments + ); + }, 0) + : 0; + }); + const modelContextWindow = + knownModelWindow ?? + (handoffUsage !== null || reuseTelemetry ? previousUsage?.maxTokens : undefined); + const budgetProviderThread = { + ...loadedProviderThread, + contextUsage: handoffUsage, + }; + // The resume fallback summary already covers every earlier run in order. A + // retry copy after it would rank older missed requests as the newest history. + const missedRuns = projection.runs.filter( + (source) => + !resumeFellBack && + source.ordinal < run.ordinal && + source.providerThreadId === providerThread.id && + (source.status === "failed" || source.status === "interrupted") && + !deliveredAttemptIds.has(source.activeAttemptId), + ); + const missedRunIds = new Set(missedRuns.map((source) => source.id)); + return { + sameNativeThread, + handoffUsage, + deliveredItemIds, + coveredItemIds, + acceptedAttempts, + nativeInputRunIds, + legacyInputRunIds, + missedRuns, + missedRunIds, + budget: Effect.gen(function* () { + return handoffBudget({ + tokenCap, + modelContextWindow, + userText: restartNote === "" ? userText : `${restartNote}\n\n${userText}`, + attachments: message.attachments, + providerThread: budgetProviderThread, + nativeContextEstimate: + budgetProviderThread.contextUsage?.usedTokens === undefined + ? yield* nativeContextEstimate + : 0, + }); + }), + }; + }; + const prepareRetryHandoff = Effect.fn("ProviderTurnStartService.prepareRetryHandoff")( + function* (context: ReturnType) { + const { missedRunIds, coveredItemIds, missedRuns } = context; + const missedItems = + missedRunIds.size === 0 + ? [] + : (yield* projectionStore.getTurnStartHistory(input.threadId, [ + ...missedRunIds, + ])).filter( + (item) => + item.runId !== null && + missedRunIds.has(item.runId) && + !coveredItemIds.has(item.id) && + historicalMessage(item) !== null, + ); + const retryHandoff = + missedItems.length === 0 + ? [] + : [ + yield* contextHandoffService.prepareProviderHandoff({ + threadId: projection.thread.id, + targetRunId: run.id, + transferId: null, + fromProviderThreadIds: [providerThread.id], + toProviderThreadId: providerThread.id, + fromProviderInstanceId: run.providerInstanceId, + toProviderInstanceId: run.providerInstanceId, + coveredRunOrdinals: { + from: missedRuns[0]!.ordinal, + to: missedRuns.at(-1)!.ordinal, + }, + strategy: "delta_since_target_last_seen", + items: missedItems, + runs: projection.runs, + createdAt: yield* DateTime.now, + }), + ]; + return retryHandoff; + }, + ); + const prepareRecoveryHandoff = Effect.fn("ProviderTurnStartService.prepareRecoveryHandoff")( + function* () { + const transferId = yield* idAllocator.allocate.contextTransfer({ + sourceThreadId: projection.thread.id, + targetThreadId: projection.thread.id, + type: "provider_resume_fallback", + }); + const createdAt = yield* DateTime.now; + const handoff = yield* contextHandoffService.prepareProviderHandoff({ + threadId: projection.thread.id, + targetRunId: run.id, + transferId, + fromProviderThreadIds: [providerThread.id], + toProviderThreadId: providerThread.id, + fromProviderInstanceId: providerThread.providerInstanceId, + toProviderInstanceId: run.providerInstanceId, + coveredRunOrdinals: { from: 1, to: Math.max(1, run.ordinal - 1) }, + strategy: "full_thread_summary", + runs: projection.runs, + items: (yield* projectionStore.getTurnStartHistory(input.threadId)).filter( + (item) => + item.runId === null || + projection.runs.some( + (source) => source.id === item.runId && source.ordinal < run.ordinal, + ), + ), + createdAt, + }); + return { handoff, transferId, createdAt }; + }, + ); + const recoveryHandoff = yield* Effect.cached(prepareRecoveryHandoff()); + const withRecoveryHandoff = (handoff: OrchestrationV2ContextHandoff) => { + // Delivery ranks later handoffs as newer history. Pending handoffs from + // earlier runs predate this summary; handoffs for this run (provider-switch + // delta, merge-back) are newer and must stay last. + return [ + ...effectiveHandoffs.filter((candidate) => candidate.targetRunId !== run.id), + handoff, + ...effectiveHandoffs.filter((candidate) => candidate.targetRunId === run.id), + ]; + }; + let resumedRetryHandoff: ReadonlyArray | undefined; + let resumedHandoffContext: ReturnType | undefined; const loadedProviderThread = yield* Effect.gen(function* () { if (nativeForkTransfer !== undefined) { const sourceProjection = yield* projectionStore.getThreadRecords( @@ -831,15 +1106,71 @@ export const layer: Layer.Layer< }), ); if (resumed._tag === "Success") { - return resumed.success; + const context = handoffContextForThread(resumed.success); + context.budget = yield* Effect.cached(context.budget); + resumedHandoffContext = context; + resumedRetryHandoff = yield* prepareRetryHandoff(context); + const fits = yield* deliverContextHandoffs({ + handoffs: [...handoffs, ...resumedRetryHandoff], + providerThread: { ...resumed.success, id: providerThread.id }, + budget: context.budget, + deferInline: isCompactionCommandMessage(message) || goalCommand, + alreadyDeliveredItemIds: context.deliveredItemIds, + persist: () => Effect.void, + }).pipe( + Effect.as(true), + Effect.catchTags({ ContextHandoffBudgetError: () => Effect.succeed(false) }), + ); + if (fits) return resumed.success; + const freshThread = { ...resumed.success, nativeThreadRef: null }; + const freshContext = handoffContextForThread(freshThread); + const { handoff } = yield* recoveryHandoff; + const fitsFresh = yield* deliverContextHandoffs({ + handoffs: withRecoveryHandoff(handoff), + providerThread: freshThread, + budget: freshContext.budget, + alreadyDeliveredItemIds: freshContext.deliveredItemIds, + persist: () => Effect.void, + }).pipe( + Effect.as(true), + Effect.catchTags({ ContextHandoffBudgetError: () => Effect.succeed(false) }), + ); + if (!fitsFresh) return resumed.success; + const hasPendingWork = yield* ( + session.hasPendingBackgroundWorkForThread?.(resumed.success, { + includeSubagents: true, + }) ?? + session.hasPendingBackgroundWork ?? + Effect.succeed(false) + ); + if (hasPendingWork) { + return yield* loadFromProvider( + Effect.fail( + new ProviderAdapter.ProviderAdapterTurnStartError({ + driver: session.driver, + threadId: projection.thread.id, + providerThreadId: providerThread.id, + runId, + cause: new ContextHandoffBackgroundWorkError(), + }), + ), + ); + } } + resumedHandoffContext = undefined; + resumedRetryHandoff = []; + if (!(yield* isCurrentAttemptInStatus("starting"))) return undefined; - yield* Effect.logWarning("Provider resume failed; attempting a fresh native session", { + yield* Effect.logWarning("Recovering provider context in a fresh native session", { driver: session.driver, providerThreadId: providerThread.id, runId, - reason: uncertainDelivery ? "uncertain_history_delivery" : "resume_failed", - errorTag: resumed.failure._tag, + reason: uncertainDelivery + ? "uncertain_history_delivery" + : resumed._tag === "Success" + ? "handoff_budget_exhausted" + : "resume_failed", + errorTag: resumed._tag === "Failure" ? resumed.failure._tag : "ContextHandoffBudgetError", }); const replacement = yield* loadFromProvider( session.ensureThread({ @@ -850,43 +1181,12 @@ export const layer: Layer.Layer< providerSessionId, // Clear the native ref to force a fresh session while preserving row identity. existingProviderThread: { ...providerThread, nativeThreadRef: null }, + forceNewNativeThread: true, }), ); if (replacement === undefined) return undefined; - const transferId = yield* idAllocator.allocate.contextTransfer({ - sourceThreadId: projection.thread.id, - targetThreadId: projection.thread.id, - type: "provider_resume_fallback", - }); - const createdAt = yield* DateTime.now; - const handoff = yield* contextHandoffService.prepareProviderHandoff({ - threadId: projection.thread.id, - targetRunId: run.id, - transferId, - fromProviderThreadIds: [providerThread.id], - toProviderThreadId: providerThread.id, - fromProviderInstanceId: providerThread.providerInstanceId, - toProviderInstanceId: run.providerInstanceId, - coveredRunOrdinals: { from: 1, to: Math.max(1, run.ordinal - 1) }, - strategy: "full_thread_summary", - runs: projection.runs, - items: (yield* projectionStore.getTurnStartHistory(input.threadId)).filter( - (item) => - item.runId === null || - projection.runs.some( - (source) => source.id === item.runId && source.ordinal < run.ordinal, - ), - ), - createdAt, - }); - // Delivery ranks later handoffs as newer history. Pending handoffs from - // earlier runs predate this summary; handoffs for this run (provider-switch - // delta, merge-back) are newer and must stay last. - effectiveHandoffs = [ - ...effectiveHandoffs.filter((candidate) => candidate.targetRunId !== run.id), - handoff, - ...effectiveHandoffs.filter((candidate) => candidate.targetRunId === run.id), - ]; + const { handoff, transferId, createdAt } = yield* recoveryHandoff; + effectiveHandoffs = withRecoveryHandoff(handoff); resumeFellBack = true; yield* eventSink.write({ events: [ @@ -934,39 +1234,17 @@ export const layer: Layer.Layer< return; } const now = yield* DateTime.now; - const measuredContext = latestNativeContextUsage(projection, providerThread); - const previousSelection = - measuredContext?.modelSelection ?? - projection.runs.findLast( - (source) => - source.ordinal < run.ordinal && - source.startedAt !== null && - source.providerThreadId === providerThread.id, - )?.modelSelection; - const sameSelection = - previousSelection === undefined || - modelSelectionsEqual(previousSelection, run.modelSelection); - const sameNativeThread = - loadedProviderThread.nativeThreadRef?.nativeId === providerThread.nativeThreadRef?.nativeId; - const threadUsage = loadedProviderThread.contextUsage ?? providerThread.contextUsage; - const previousUsage = measuredContext - ? { ...threadUsage, ...measuredContext.usage } - : threadUsage; - const reuseTelemetry = - sameSelection || - (previousSelection !== undefined && - session.canReuseContextUsage?.(previousSelection, run.modelSelection) === true); - const knownModelWindow = session.getModelContextWindow?.( - run.modelSelection, - resolvedRuntimePolicy.cwd, - ); - const handoffUsage = contextUsageForHandoff({ + const handoffContext = resumedHandoffContext ?? handoffContextForThread(loadedProviderThread); + const { sameNativeThread, - sameSelection, - reuseTelemetry, - previousUsage, - knownModelWindow, - }); + handoffUsage, + deliveredItemIds, + acceptedAttempts, + nativeInputRunIds, + legacyInputRunIds, + budget, + } = handoffContext; + const retryHandoff = resumedRetryHandoff ?? (yield* prepareRetryHandoff(handoffContext)); const runningProviderThread: OrchestrationV2ProviderThread = { ...loadedProviderThread, contextUsage: handoffUsage, @@ -1094,191 +1372,13 @@ export const layer: Layer.Layer< const routableSubagents = projection.subagents.filter((subagent) => RunExecutionService.canRouteRelatedSubagent(subagent.status), ); - const userText = projectComposerContextForProvider({ - 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, - projection.runs, - projection.providerTurns, - projection.attempts, - ); - const restartCancelledWork = pendingRestartCancelledBackgroundWork({ - runs: projection.runs, - providerTurns: projection.providerTurns, - compactionMessageIds: new Set( - projection.messages - .filter( - (candidate) => - isCompactionCommandMessage(candidate) || - (goalDriver && parseGoalCommand(candidate) !== null), - ) - .map((candidate) => candidate.id), - ), - run, - attempts: projection.attempts, - }); - const restartNote = - restartCancelledWork.length === 0 - ? "" - : restartCancelledBackgroundWorkNote(restartCancelledWork); - const tokenCap = yield* handoffTokenCapConfig.pipe( - Effect.orElseSucceed(() => DEFAULT_HANDOFF_TOKEN_CAP), - ); - const settledHandoffs = projection.contextHandoffs.filter( - (handoff) => - handoff.toProviderThreadId === providerThread.id && - handoff.delivery?.nativeThreadId === runningProviderThread.nativeThreadRef?.nativeId && - handoff.delivery?.status !== "pending", - ); - const deliveredItemIds = new Set( - settledHandoffs.flatMap((handoff) => handoff.delivery?.itemIds ?? []), - ); - const coveredItemIds = new Set([ - ...deliveredItemIds, - ...settledHandoffs.flatMap((handoff) => handoff.delivery?.omittedItemIds ?? []), - ]); - const deliveredAttemptIds = new Set( - projection.providerTurns.map((turn) => turn.runAttemptId), - ); - const acceptedAttempts = projection.attempts.filter( - (source) => - source.providerThreadId === providerThread.id && deliveredAttemptIds.has(source.id), - ); - const nativeInputRunIds = new Set( - acceptedAttempts - .filter( - (source) => - source.nativeThreadId !== undefined && - source.nativeThreadId === runningProviderThread.nativeThreadRef?.nativeId, - ) - .map((source) => source.runId), - ); - const legacyInputRunIds = new Set( - acceptedAttempts - .filter((source) => source.nativeThreadId === undefined) - .map((source) => source.runId), - ); - const legacyRecoveredRunIds = new Set( - projection.runs - .filter( - (source) => - source.providerThreadId === providerThread.id && - settledHandoffs.some( - (handoff) => - handoff.strategy === "full_thread_summary" && - handoff.fromProviderThreadIds.includes(providerThread.id) && - source.ordinal >= handoff.coveredRunOrdinals.from && - source.ordinal <= handoff.coveredRunOrdinals.to, - ), - ) - .map((source) => source.id), - ); - // Legacy attempts lack native ids; exclude recovered attachments absent from the replacement thread. - const nativeContextEstimate = Effect.gen(function* () { - return sameNativeThread - ? (yield* projectionStore.getTurnStartHistory(input.threadId)).reduce((sum, item) => { - if ( - item.runId === run.id || - (item.runId !== null && - missedRunIds.has(item.runId) && - !deliveredItemIds.has(item.id)) || - (item.providerThreadId !== providerThread.id && !deliveredItemIds.has(item.id)) - ) - return sum; - const historical = historicalMessage(item); - const nativeAttachments = - item.type === "user_message" && - item.providerThreadId === providerThread.id && - item.runId !== null && - (nativeInputRunIds.has(item.runId) || - (legacyInputRunIds.has(item.runId) && - !coveredItemIds.has(item.id) && - !legacyRecoveredRunIds.has(item.runId))) - ? attachmentTokenAllowance(item.attachments) - : 0; - return ( - sum + - (historical === null ? 0 : Buffer.byteLength(historical.text)) + - nativeAttachments - ); - }, 0) - : 0; - }); - const modelContextWindow = - knownModelWindow ?? - (handoffUsage !== null || reuseTelemetry ? previousUsage?.maxTokens : undefined); - const budgetProviderThread = { - ...runningProviderThread, - contextUsage: handoffUsage, - }; - // The resume fallback summary already covers every earlier run in order. A - // retry copy after it would rank older missed requests as the newest history. - const missedRuns = projection.runs.filter( - (source) => - !resumeFellBack && - source.ordinal < run.ordinal && - source.providerThreadId === providerThread.id && - (source.status === "failed" || source.status === "interrupted") && - !deliveredAttemptIds.has(source.activeAttemptId), - ); - const missedRunIds = new Set(missedRuns.map((source) => source.id)); - const missedItems = - missedRunIds.size === 0 - ? [] - : (yield* projectionStore.getTurnStartHistory(input.threadId, [...missedRunIds])).filter( - (item) => - item.runId !== null && - missedRunIds.has(item.runId) && - !coveredItemIds.has(item.id) && - historicalMessage(item) !== null, - ); const startWithHandoffs = Effect.fn("ProviderTurnStartService.startWithHandoffs")( function* (turnInput: Parameters[0], compact = false) { - const retryHandoff = - missedItems.length === 0 - ? [] - : [ - yield* contextHandoffService.prepareProviderHandoff({ - threadId: projection.thread.id, - targetRunId: run.id, - transferId: null, - fromProviderThreadIds: [providerThread.id], - toProviderThreadId: providerThread.id, - fromProviderInstanceId: run.providerInstanceId, - toProviderInstanceId: run.providerInstanceId, - coveredRunOrdinals: { - from: missedRuns[0]!.ordinal, - to: missedRuns.at(-1)!.ordinal, - }, - strategy: "delta_since_target_last_seen", - items: missedItems, - runs: projection.runs, - createdAt: yield* DateTime.now, - }), - ]; const delivery = yield* deliverContextHandoffs({ handoffs: [...effectiveHandoffs, ...retryHandoff], deferInline: compact || goalCommand, providerThread: runningProviderThread, - budget: Effect.gen(function* () { - return handoffBudget({ - tokenCap, - modelContextWindow, - userText: restartNote === "" ? userText : `${restartNote}\n\n${userText}`, - attachments: message.attachments, - providerThread: budgetProviderThread, - nativeContextEstimate: - budgetProviderThread.contextUsage?.usedTokens === undefined - ? yield* nativeContextEstimate - : 0, - }); - }), + budget, alreadyDeliveredItemIds: deliveredItemIds, ...(session.injectHistory === undefined ? {} @@ -1361,7 +1461,7 @@ export const layer: Layer.Layer< ); const deliverySession = effectiveHandoffs.length === 0 && - missedItems.length === 0 && + retryHandoff.length === 0 && restartNote === "" && !noteContinuation ? session diff --git a/apps/server/src/orchestration-v2/testkit/ProviderSwitch.integration.test.ts b/apps/server/src/orchestration-v2/testkit/ProviderSwitch.integration.test.ts index 0fd201750464..bb21d8ca9929 100644 --- a/apps/server/src/orchestration-v2/testkit/ProviderSwitch.integration.test.ts +++ b/apps/server/src/orchestration-v2/testkit/ProviderSwitch.integration.test.ts @@ -115,6 +115,8 @@ function makeTestAdapter(input: { readonly failedRunOrdinals?: ReadonlySet; readonly interruptedRunOrdinals?: ReadonlySet; readonly holdRunOrdinal?: number; + readonly pendingBackgroundWork?: Ref.Ref; + readonly subagentOnlyBackgroundWork?: boolean; readonly holdFirstTurn?: Deferred.Deferred; readonly releaseFirstTurn?: Deferred.Deferred; }): ProviderAdapter.ProviderAdapterV2["Service"] { @@ -146,6 +148,14 @@ function makeTestAdapter(input: { providerSessionId: sessionInput.providerSessionId, providerSession, events: Stream.fromPubSub(events), + ...(input.pendingBackgroundWork === undefined + ? {} + : { + hasPendingBackgroundWorkForThread: (_thread, options) => + input.subagentOnlyBackgroundWork && options?.includeSubagents !== true + ? Effect.succeed(false) + : Ref.get(input.pendingBackgroundWork!), + }), ...(input.getModelContextWindow === undefined ? {} : { getModelContextWindow: input.getModelContextWindow }), @@ -419,6 +429,13 @@ describe("orchestration v2 provider switching", () => { "screenshot-fallback", "screenshot-reported-capacity-native", "screenshot-option-change-retry-native", + "screenshot-exhausted-target-native", + "screenshot-exhausted-target-fallback", + "screenshot-exhausted-target-estimate-native", + "screenshot-exhausted-target-oversized-native", + "screenshot-exhausted-target-tight-native", + "screenshot-exhausted-target-background-native", + "screenshot-exhausted-target-subagent-native", "prior-images-reasoning-change-replacement-small-native", "imported-prior-images-fallback", "prior-images-telemetry-native", @@ -438,11 +455,16 @@ describe("orchestration v2 provider switching", () => { const failStartOnce = yield* Ref.make(false); const failResumeOnce = yield* Ref.make(false); const generation = yield* Ref.make(0); + const pendingBackgroundWork = yield* Ref.make(false); const priorImages = scenario.includes("prior-images"); + const exhaustedTarget = scenario.includes("exhausted-target"); const replaceNative = scenario.includes("replacement"); const capacityScenario = modelScenario || scenario.includes("reported-capacity"); const returning = - scenario === "large-current-input" || scenario.includes("-change-") || priorImages; + scenario === "large-current-input" || + scenario.includes("-change-") || + priorImages || + exhaustedTarget; const targetSelection: ModelSelection = !modelScenario ? CLAUDE_MODEL_SELECTION : { @@ -467,10 +489,19 @@ describe("orchestration v2 provider switching", () => { driver: CLAUDE_DRIVER, capabilities: ClaudeProviderCapabilitiesV2, modelSelection: CLAUDE_MODEL_SELECTION, - responseByRunOrdinal: {}, + responseByRunOrdinal: scenario.includes("target-estimate") + ? { 2: "x".repeat(10_000) } + : {}, capturedTurns, ...(scenario.includes("retry") || scenario.includes("unsent") ? { failStartOnce } : {}), ...(replaceNative ? { failResumeOnce, nativeThreadGeneration: generation } : {}), + ...(exhaustedTarget ? { nativeThreadGeneration: generation } : {}), + ...(scenario.includes("target-background") || scenario.includes("target-subagent") + ? { + pendingBackgroundWork, + subagentOnlyBackgroundWork: scenario.includes("target-subagent"), + } + : {}), ...(scenario.includes("reported-capacity") ? { initialContextUsage: { usedTokens: 999_999, maxTokens: 1_000_000 } } : {}), @@ -498,8 +529,11 @@ describe("orchestration v2 provider switching", () => { ? 1_000_000 : 32_000, } - : scenario === "large-current-input" || priorImages - ? { getModelContextWindow: () => 32_000 } + : scenario === "large-current-input" || priorImages || exhaustedTarget + ? { + getModelContextWindow: () => + scenario.includes("target-tight") ? 25_121 : 32_000, + } : {}), ...(scenario.endsWith("-native") || scenario === "large-current-input" ? { injectedHistory } @@ -546,7 +580,9 @@ describe("orchestration v2 provider switching", () => { })) : turnUsageScenario && ordinal >= targetOrdinal ? [screenshot, { ...screenshot, id: "current-2" }] - : scenario.startsWith("screenshot") && ordinal >= targetOrdinal + : scenario.startsWith("screenshot") && + ordinal >= targetOrdinal && + (!exhaustedTarget || ordinal === targetOrdinal) ? screenshots : priorImages && ordinal === (scenario.startsWith("imported") ? 1 : 2) ? [screenshot] @@ -602,19 +638,49 @@ describe("orchestration v2 provider switching", () => { yield* Effect.addFinalizer(() => Effect.sync(() => spy?.mockRestore())); const current = scenario.startsWith("compact") ? "/compact" - : capacityScenario && !turnUsageScenario - ? "x".repeat(70_000) - : scenario === "large-current-input" - ? "x".repeat(9_000) - : "Continue work"; + : scenario.includes("target-oversized") + ? "x".repeat(40_000) + : capacityScenario && !turnUsageScenario + ? "x".repeat(70_000) + : scenario === "large-current-input" + ? "x".repeat(9_000) + : "Continue work"; if (returning) { if (scenario.includes("unsent")) yield* Ref.set(failStartOnce, true); yield* dispatch( 2, - turnUsageScenario ? "x".repeat(27_460) : "Establish target", + turnUsageScenario + ? "x".repeat(27_460) + : scenario.includes("target-estimate") + ? "x".repeat(12_000) + : "Establish target", CLAUDE_MODEL_SELECTION, ); yield* wait(2); + if (exhaustedTarget) { + assert.equal( + (yield* orchestrator.getThreadProjection(threadId)).runs.at(-1)?.status, + "completed", + ); + } + if (exhaustedTarget && !scenario.includes("target-estimate")) { + const target = (yield* orchestrator.getThreadProjection( + threadId, + )).providerThreads.find( + (thread) => thread.providerInstanceId === CLAUDE_MODEL_SELECTION.instanceId, + )!; + yield* eventSink.write({ + events: [ + { + id: EventId.make("exhausted-target-usage"), + type: "provider-thread.updated", + threadId, + occurredAt: yield* DateTime.now, + payload: { ...target, contextUsage: { usedTokens: 30_000, maxTokens: 32_000 } }, + }, + ], + }); + } if (modelScenario && !turnUsageScenario) { const target = (yield* orchestrator.getThreadProjection( threadId, @@ -695,6 +761,8 @@ describe("orchestration v2 provider switching", () => { yield* Ref.set(failResumeOnce, true); } if (scenario.includes("retry")) yield* Ref.set(failStartOnce, true); + if (scenario.includes("target-background") || scenario.includes("target-subagent")) + yield* Ref.set(pendingBackgroundWork, true); yield* dispatch(targetOrdinal, current, targetSelection); yield* wait(targetOrdinal); if (scenario.includes("retry")) { @@ -780,7 +848,73 @@ describe("orchestration v2 provider switching", () => { yield* wait(targetOrdinal + 3); } const projection = yield* orchestrator.getThreadProjection(threadId); + if ( + scenario.includes("target-oversized") || + scenario.includes("target-tight") || + scenario.includes("target-background") || + scenario.includes("target-subagent") + ) { + assert.equal(projection.runs.at(-1)?.status, "failed"); + assert.equal(yield* Ref.get(generation), 1); + assert.equal(projection.messages.at(-1)!.text, current); + assert.deepEqual(projection.messages.at(-1)!.attachments, screenshots); + if (scenario.includes("target-background") || scenario.includes("target-subagent")) { + assert.isTrue(yield* Ref.get(pendingBackgroundWork)); + const failure = projection.turnItems.findLast((item) => item.type === "error"); + assert.equal( + failure?.type === "error" ? failure.failure.message : null, + "The target provider still has background work running. Wait for it to finish or stop it before retrying the handoff.", + ); + } + return; + } assert.equal(projection.runs.at(-1)?.status, "completed"); + if (exhaustedTarget) { + assert.equal(yield* Ref.get(generation), 2); + const turns = (yield* Ref.get(capturedTurns)).filter( + (turn) => turn.driver === CLAUDE_DRIVER, + ); + assert.notEqual(turns[0]!.nativeThreadId, turns[1]!.nativeThreadId); + const latestRun = projection.runs.at(-1)!; + assert.equal( + projection.attempts.find((attempt) => attempt.id === latestRun.activeAttemptId)! + .nativeThreadId, + turns[1]!.nativeThreadId, + ); + const context = scenario.endsWith("-native") + ? yield* encodeJson(yield* Ref.get(injectedHistory)) + : turns[1]!.text; + assert.include(context, "Original request with constraints"); + assert.include(context, "Original partial work"); + assert.include(context, "New source constraint"); + assert.include(turns[1]!.text, current); + assert.deepEqual(turns[1]!.attachments, screenshots); + const historyBeforeFollowup = yield* Ref.get(injectedHistory); + yield* dispatch(5, "Continue after automatic recovery", CLAUDE_MODEL_SELECTION); + yield* wait(5); + assert.equal( + (yield* orchestrator.getThreadProjection(threadId)).runs.at(-1)?.status, + "completed", + ); + assert.equal(yield* Ref.get(generation), 2); + assert.equal( + (yield* Ref.get(capturedTurns)).at(-1)!.text, + "Continue after automatic recovery", + ); + assert.deepEqual(yield* Ref.get(injectedHistory), historyBeforeFollowup); + if (scenario.includes("target-estimate")) { + yield* dispatch(6, "Another source turn", CODEX_MODEL_SELECTION); + yield* wait(6); + yield* dispatch(7, "Return after recovery", CLAUDE_MODEL_SELECTION); + yield* wait(7); + assert.equal( + (yield* orchestrator.getThreadProjection(threadId)).runs.at(-1)?.status, + "completed", + ); + assert.equal(yield* Ref.get(generation), 2); + } + return; + } if (priorImages) { const handoff = projection.contextHandoffs.at(-1)!; const context = scenario.endsWith("-native") @@ -901,7 +1035,12 @@ describe("orchestration v2 provider switching", () => { }), ), ); - it.live.each(["turn-start", "injection", "large-missed-request"] as const)( + it.live.each([ + "turn-start", + "injection", + "large-missed-request", + "exhausted-target-retry", + ] as const)( "recovers %s failure without duplicating history in the same native thread", (failure) => Effect.scoped( @@ -999,6 +1138,22 @@ describe("orchestration v2 provider switching", () => { failure !== "injection" ? "injected" : "pending", ); const historyBeforeRetry = (yield* Ref.get(injectedHistory)).length; + if (failure === "exhausted-target-retry") { + const target = failed.providerThreads.find( + (thread) => thread.providerInstanceId === CLAUDE_MODEL_SELECTION.instanceId, + )!; + yield* (yield* EventSink.EventSinkV2).write({ + events: [ + { + id: EventId.make("exhausted-retry-usage"), + type: "provider-thread.updated", + threadId, + occurredAt: yield* DateTime.now, + payload: { ...target, contextUsage: { usedTokens: 30_000, maxTokens: 32_000 } }, + }, + ], + }); + } const retryText = failure === "large-missed-request" ? "y".repeat(10_000) : "Retry target request"; yield* dispatch(3, retryText, CLAUDE_MODEL_SELECTION); @@ -1006,7 +1161,7 @@ describe("orchestration v2 provider switching", () => { const retried = yield* orchestrator.getThreadProjection(threadId); const lastTurn = (yield* Ref.get(capturedTurns)).at(-1)!; assert.equal(lastTurn.text, retryText); - if (failure !== "injection") { + if (failure !== "injection" && failure !== "exhausted-target-retry") { const delta = yield* encodeJson( (yield* Ref.get(injectedHistory)).slice(historyBeforeRetry), ); diff --git a/docs/user/portable-handoffs.md b/docs/user/portable-handoffs.md index e2d84458ed93..04b2a8ce1498 100644 --- a/docs/user/portable-handoffs.md +++ b/docs/user/portable-handoffs.md @@ -20,9 +20,13 @@ agent-written summary. ## Context limits A handoff must leave room for existing provider context, your request and attachments, instructions, -tools, and subsequent work. If even its retrieval references cannot fit, T3 Code reports an error -instead of shortening your request. Compact the target conversation or select a larger-context -model before trying again. +tools, and subsequent work. If the target conversation has no room for the handoff, T3 Code starts +a fresh provider conversation with a bounded selection of saved history and sends your request +automatically. You do not need to compact manually. If the request and attachments leave too little +room even in a fresh conversation, shorten the request or select a larger-context model. T3 Code +never shortens your request to make it fit. + +Recovery leaves running background work intact. Let it finish or stop it before retrying the handoff. Server operators can set `T3CODE_CONTEXT_HANDOFF_TOKEN_CAP` to change the initial history allowance (default 16,000; clamped to 1,024–64,000). This is an upper bound, not a provider context-window diff --git a/packages/provider-acp/src/server/adapter.ts b/packages/provider-acp/src/server/adapter.ts index 6fa99a058cae..65263a67cb7a 100644 --- a/packages/provider-acp/src/server/adapter.ts +++ b/packages/provider-acp/src/server/adapter.ts @@ -6550,6 +6550,46 @@ export const makeAcpAdapterV2 = Effect.fn("makeAcpAdapterV2")(function* ( return replacementExit.value.started; }); + const commitReplacementSession = Effect.fn("AcpAdapterV2.commitReplacementSession")( + function* ( + threadId: ThreadId | null, + candidate: AcpSessionRuntime.AcpSessionRuntimeStartResult, + ) { + rememberTerminalEnvironment(candidate.sessionId, yield* readMcpContext(threadId)); + yield* Ref.set(runtimeRestartRequired, false); + yield* Ref.set(activeSessionId, candidate.sessionId); + yield* setActiveSessionSetup(candidate); + yield* Ref.set(activeSelection, null); + yield* Ref.set(activeInteractionMode, null); + yield* Ref.set(promptInstructionStates, new Map()); + yield* Ref.set(compactionMarkers, new Map()); + yield* Ref.set(providerTurns, new Map()); + yield* Ref.set(snapshot, { + order: [], + messages: new Map(), + loadingRole: null, + loadingMessageId: null, + loadingIndex: 0, + }); + yield* continuationPermit.withPermit( + Effect.gen(function* () { + yield* Ref.update(continuationGeneration, (value) => value + 1); + yield* Ref.set(stoppedRunQuarantine, false); + yield* Ref.set(wakeBuffer, []); + yield* Ref.set(wakeReports, noWakeReports); + yield* Ref.set(continuationRequested, false); + yield* Ref.set(runningBackgroundTaskIds, new Set()); + yield* Ref.set(endedBackgroundTaskIds, new Set()); + yield* Ref.set(midTurnUnreportedCompletedTaskIds, new Set()); + yield* Ref.set(handledBackgroundTaskIdsInActiveTurn, new Set()); + yield* Ref.set(carryoverSubagents, null); + yield* Ref.set(suppressPostSettleMonitorPrompt, false); + yield* Ref.set(lastTurnRoute, null); + }), + ); + }, + ); + const initialStart = yield* Effect.result( startAcpRuntime(input.threadId, input.initialNativeThreadId), ); @@ -7784,28 +7824,61 @@ export const makeAcpAdapterV2 = Effect.fn("makeAcpAdapterV2")(function* ( : {}), ensureThread: Effect.fn("AcpAdapterV2.ensureThread")( function* (threadInput: ProviderAdapter.ProviderAdapterV2EnsureThreadInput) { - const now = yield* DateTime.now; - const sessionId = yield* Ref.get(activeSessionId); - if (sessionId === null) { - return yield* new ProviderAdapter.ProviderAdapterProtocolError({ - driver, - detail: "ACP runtime did not produce a session id", - }); - } - const providerThread = makeProviderThread({ - driver, - providerInstanceId: options.instanceId, - idAllocator, - appThreadId: threadInput.threadId, - providerSessionId: input.providerSessionId, - nativeThreadId: sessionId, - ...(itemIdentityVersion === undefined ? {} : { itemIdentityVersion }), - now, - }); - yield* Ref.update(providerThreadByNativeSessionId, (current) => - new Map(current).set(sessionId, providerThread), + return yield* runtimeTransitionPermit.withPermit( + Effect.gen(function* () { + if (threadInput.forceNewNativeThread) { + const currentTurn = yield* Ref.get(activeTurn); + const pendingWork = yield* ( + sessionRuntime.hasPendingBackgroundWork ?? Effect.succeed(false) + ); + if (currentTurn !== null || pendingWork) + return yield* new ProviderAdapter.ProviderAdapterProtocolError({ + driver, + detail: + "Cannot replace ACP session while a turn or background work is active", + }); + yield* awaitRuntimeTeardown(); + const previousItemIdentityVersion = itemIdentityVersion; + itemIdentityVersion = 2; + prepareTerminalEnvironment(yield* readMcpContext(threadInput.threadId)); + yield* startReplacementAcpRuntime(threadInput.threadId, (candidate) => + commitReplacementSession(threadInput.threadId, candidate), + ).pipe( + Effect.tapError(() => + Effect.sync(() => { + itemIdentityVersion = previousItemIdentityVersion; + }), + ), + ); + } + const now = yield* DateTime.now; + const sessionId = yield* Ref.get(activeSessionId); + if (sessionId === null) { + return yield* new ProviderAdapter.ProviderAdapterProtocolError({ + driver, + detail: "ACP runtime did not produce a session id", + }); + } + const created = makeProviderThread({ + driver, + providerInstanceId: options.instanceId, + idAllocator, + appThreadId: threadInput.threadId, + providerSessionId: input.providerSessionId, + nativeThreadId: sessionId, + ...(itemIdentityVersion === undefined ? {} : { itemIdentityVersion }), + now, + }); + const providerThread = + threadInput.forceNewNativeThread && threadInput.existingProviderThread + ? { ...created, id: threadInput.existingProviderThread.id } + : created; + yield* Ref.update(providerThreadByNativeSessionId, (current) => + new Map(current).set(sessionId, providerThread), + ); + return providerThread; + }), ); - return providerThread; }, (effect, threadInput) => effect.pipe( @@ -8348,43 +8421,7 @@ export const makeAcpAdapterV2 = Effect.fn("makeAcpAdapterV2")(function* ( const replacement = yield* startReplacementAcpRuntime( rollbackInput.providerThread.appThreadId, (candidate) => - Effect.gen(function* () { - rememberTerminalEnvironment( - candidate.sessionId, - yield* readMcpContext(rollbackInput.providerThread.appThreadId), - ); - yield* Ref.set(runtimeRestartRequired, false); - yield* Ref.set(activeSessionId, candidate.sessionId); - yield* setActiveSessionSetup(candidate); - yield* Ref.set(activeSelection, null); - yield* Ref.set(activeInteractionMode, null); - yield* Ref.set(promptInstructionStates, new Map()); - yield* Ref.set(compactionMarkers, new Map()); - yield* Ref.set(providerTurns, new Map()); - yield* Ref.set(snapshot, { - order: [], - messages: new Map(), - loadingRole: null, - loadingMessageId: null, - loadingIndex: 0, - }); - yield* continuationPermit.withPermit( - Effect.gen(function* () { - yield* Ref.update(continuationGeneration, (value) => value + 1); - yield* Ref.set(stoppedRunQuarantine, false); - yield* Ref.set(wakeBuffer, []); - yield* Ref.set(wakeReports, noWakeReports); - yield* Ref.set(continuationRequested, false); - yield* Ref.set(runningBackgroundTaskIds, new Set()); - yield* Ref.set(endedBackgroundTaskIds, new Set()); - yield* Ref.set(midTurnUnreportedCompletedTaskIds, new Set()); - yield* Ref.set(handledBackgroundTaskIdsInActiveTurn, new Set()); - yield* Ref.set(carryoverSubagents, null); - yield* Ref.set(suppressPostSettleMonitorPrompt, false); - yield* Ref.set(lastTurnRoute, null); - }), - ); - }), + commitReplacementSession(rollbackInput.providerThread.appThreadId, candidate), ).pipe(Effect.retry({ times: 1 })); const now = yield* DateTime.now; return { diff --git a/packages/provider-core/src/server/ProviderAdapter.ts b/packages/provider-core/src/server/ProviderAdapter.ts index 7c77b7bb0ac6..3396117c31b0 100644 --- a/packages/provider-core/src/server/ProviderAdapter.ts +++ b/packages/provider-core/src/server/ProviderAdapter.ts @@ -411,6 +411,7 @@ export interface ProviderAdapterV2EnsureThreadInput { readonly runtimePolicy: ProviderAdapterV2RuntimePolicy; readonly providerSessionId?: ProviderSessionId; readonly existingProviderThread?: OrchestrationV2ProviderThread; + readonly forceNewNativeThread?: boolean; } export interface ProviderAdapterV2TurnInput { @@ -549,9 +550,10 @@ export interface ProviderAdapterV2SessionRuntime { /** Manager runtimes broadcast independent subscriptions; adapters may expose only a single-consumer stream. */ readonly subscribeEvents?: Effect.Effect; readonly hasPendingBackgroundWork?: Effect.Effect; - /** Overrides the session-wide probe so sibling threads cannot keep this root subscription open. */ + /** Root-subscription probe; replacement checks opt in to subagents owned by the live process. */ readonly hasPendingBackgroundWorkForThread?: ( providerThread: OrchestrationV2ProviderThread, + options?: { readonly includeSubagents?: boolean }, ) => Effect.Effect; /** Set only if all session background tasks end with its process; otherwise threads must keep Waiting. */ readonly backgroundTasksEndWithProcess?: true; diff --git a/packages/provider-core/src/server/failure.test.ts b/packages/provider-core/src/server/failure.test.ts index 16f0f871ccc2..8fb2e153aa10 100644 --- a/packages/provider-core/src/server/failure.test.ts +++ b/packages/provider-core/src/server/failure.test.ts @@ -16,6 +16,7 @@ import { makeProviderFailureTurnItem, MAX_PROVIDER_FAILURE_MESSAGE_LENGTH, ContextHandoffBudgetError, + ContextHandoffBackgroundWorkError, } from "./failure.ts"; import * as IdAllocator from "./IdAllocator.ts"; import * as ProviderAdapter from "@t3tools/provider-core/server/ProviderAdapter"; @@ -69,17 +70,16 @@ it("does not expose arbitrary cause messages and preserves explicit messages", ( }); it("preserves actionable handoff errors wrapped by turn startup", () => { - const cause = new ProviderAdapter.ProviderAdapterTurnStartError({ - driver: ProviderDriverKind.make("codex"), - threadId: ThreadId.make("thread:handoff-error"), - providerThreadId: ProviderThreadId.make("provider-thread:handoff-error"), - runId: RunId.make("run:handoff-error"), - cause: new ContextHandoffBudgetError(), - }); - assert.equal( - makeProviderFailure({ cause: Cause.fail(cause) }).message, - new ContextHandoffBudgetError().message, - ); + for (const error of [new ContextHandoffBudgetError(), new ContextHandoffBackgroundWorkError()]) { + const cause = new ProviderAdapter.ProviderAdapterTurnStartError({ + driver: ProviderDriverKind.make("codex"), + threadId: ThreadId.make("thread:handoff-error"), + providerThreadId: ProviderThreadId.make("provider-thread:handoff-error"), + runId: RunId.make("run:handoff-error"), + cause: error, + }); + assert.equal(makeProviderFailure({ cause: Cause.fail(cause) }).message, error.message); + } }); it("keeps unrecognized reasons and provider text out of category messages", () => { diff --git a/packages/provider-core/src/server/failure.ts b/packages/provider-core/src/server/failure.ts index 48331a10eac2..4e24fcc945c7 100644 --- a/packages/provider-core/src/server/failure.ts +++ b/packages/provider-core/src/server/failure.ts @@ -22,7 +22,16 @@ export class ContextHandoffBudgetError extends Schema.TaggedError()( + "ContextHandoffBackgroundWorkError", + {}, +) { + override get message() { + return "The target provider still has background work running. Wait for it to finish or stop it before retrying the handoff."; } } @@ -87,6 +96,11 @@ function curatedCauseMessages(cause: unknown): { switch (tag) { case "ContextHandoffBudgetError": return { known: new ContextHandoffBudgetError().message, category: categoryMessage() }; + case "ContextHandoffBackgroundWorkError": + return { + known: new ContextHandoffBackgroundWorkError().message, + category: categoryMessage(), + }; case "AcpModelSelectionError": case "ClaudeBackgroundWorkBlocksQueryReplacementError": case "ClaudeAppendSystemPromptFileError": From 4914cdd98f3d15ab2ddef34f0af70d3d68a2d8e5 Mon Sep 17 00:00:00 2001 From: oliver <97427849+flamboh@users.noreply.github.com> Date: Tue, 27 Oct 2026 13:01:41 -0700 Subject: [PATCH 04/19] fix(web): chips drop their pill and stay readable on every theme (#25066) Co-authored-by: Claude Opus 5.5 Co-authored-by: maria-rcks --- .../components/ComposerPromptEditorTiptap.tsx | 11 +- .../components/ContextChip.browser.test.tsx | 42 +++++ apps/web/src/components/ContextChip.tsx | 79 ++++----- apps/web/src/components/ThreadContextChip.tsx | 2 +- .../components/chat/AssistantCitationChip.tsx | 2 +- .../src/components/chat/MarkdownFileLink.tsx | 13 +- .../chat/TimelineUserMessageRow.tsx | 5 + .../chat/UserMessageContextChips.tsx | 8 +- apps/web/src/components/contextChipParts.tsx | 62 ++----- .../web/src/components/files/AudioPreview.tsx | 6 +- apps/web/src/lib/contextChipAccents.ts | 29 ++++ apps/web/src/pierre-icons.ts | 24 ++- apps/web/src/themePalette.browser.test.ts | 56 +++++++ apps/web/src/themePalette.test.ts | 151 ++++++++++++++++++ apps/web/src/themePalette.ts | 110 +++++++++++-- vite.config.ts | 6 +- 16 files changed, 455 insertions(+), 151 deletions(-) create mode 100644 apps/web/src/components/ContextChip.browser.test.tsx create mode 100644 apps/web/src/lib/contextChipAccents.ts create mode 100644 apps/web/src/themePalette.browser.test.ts diff --git a/apps/web/src/components/ComposerPromptEditorTiptap.tsx b/apps/web/src/components/ComposerPromptEditorTiptap.tsx index ba5b46252590..af5a200e264a 100644 --- a/apps/web/src/components/ComposerPromptEditorTiptap.tsx +++ b/apps/web/src/components/ComposerPromptEditorTiptap.tsx @@ -1,4 +1,4 @@ -import { documentIconColorForFileName } from "~/pierre-icons"; +import { documentChipKindForFileName } from "~/pierre-icons"; import { isProviderSkillUserInvocable } from "@t3tools/client-runtime/providerSkills"; import { formatProviderSkillDisplayName } from "@t3tools/shared/inlineSkills"; import { @@ -46,7 +46,6 @@ import { useLayoutEffect, useRef, useState, - type CSSProperties, type KeyboardEvent as ReactKeyboardEvent, type RefObject, } from "react"; @@ -232,7 +231,7 @@ function resolvedThemeFromDocument(): "light" | "dark" { } const CHIP_NODE_SELECTION_CLASS_NAME = - "relative inline-flex select-none items-center align-middle leading-none data-[composer-chip-selected]:after:pointer-events-none data-[composer-chip-selected]:after:absolute data-[composer-chip-selected]:after:inset-0 data-[composer-chip-selected]:after:rounded-sm data-[composer-chip-selected]:after:bg-[Highlight] data-[composer-chip-selected]:after:opacity-30 data-[composer-chip-selected]:after:content-['']"; + "relative inline-flex min-w-0 max-w-full select-none items-baseline data-[composer-chip-selected]:after:pointer-events-none data-[composer-chip-selected]:after:absolute data-[composer-chip-selected]:after:inset-0 data-[composer-chip-selected]:after:rounded-sm data-[composer-chip-selected]:after:bg-[Highlight] data-[composer-chip-selected]:after:opacity-30 data-[composer-chip-selected]:after:content-['']"; const ComposerMentionExtension = Node.create({ name: "composer-mention", @@ -260,11 +259,9 @@ const ComposerMentionExtension = Node.create({ function ComposerMentionNodeView({ node }: NodeViewProps) { const actions = use(ComposerContextActionsContext); const path = (node.attrs.path as string) ?? ""; - const accent = documentIconColorForFileName(path); const chip = ( } onClick={() => actions.openMention(path)} @@ -442,7 +439,7 @@ function ComposerCitationNodeView({ node, editor, getPos }: NodeViewProps) { return ( ; +} + +describe("ContextChip", () => { + it("sets its label on the sentence baseline at the sentence size without growing the line", async () => { + await using screen = await render( +

+ Open + the{" "} + }> + + + src/ + + Typography.tsx + + {" "} + file +

, + ); + const paragraph = screen.container.querySelector("p")!; + const label = screen.container.querySelector('[data-slot="context-chip-label"]')!; + const baseline = (id: string) => + screen.container.querySelector(`[data-baseline="${id}"]`)!.getBoundingClientRect().bottom; + + expect(getComputedStyle(label).fontSize).toBe(getComputedStyle(paragraph).fontSize); + expect(Math.abs(baseline("chip") - baseline("text"))).toBeLessThanOrEqual(0.5); + expect(paragraph.getBoundingClientRect().height).toBeCloseTo( + Number.parseFloat(getComputedStyle(paragraph).lineHeight), + 0, + ); + }); +}); diff --git a/apps/web/src/components/ContextChip.tsx b/apps/web/src/components/ContextChip.tsx index 72633d478e11..b6abe578770c 100644 --- a/apps/web/src/components/ContextChip.tsx +++ b/apps/web/src/components/ContextChip.tsx @@ -6,60 +6,40 @@ import type * as React from "react"; import { cn } from "~/lib/utils"; const contextChipVariants = cva( - "inline-flex h-[1.41em] max-w-full items-center gap-[0.33em] rounded-[0.5em] border px-[0.5em] align-middle font-medium text-[0.86em] leading-none [&_svg]:block [&_svg]:size-[1.17em] [&_svg]:shrink-0 [&_svg]:self-center [button&,a&,[data-popup-open]&]:cursor-pointer [button&,a&]:transition-colors [button&,a&]:motion-reduce:transition-none focus-visible:outline-2 focus-visible:-outline-offset-2 focus-visible:outline-foreground disabled:cursor-default", + "inline-flex max-w-full items-center gap-[0.2em] font-medium leading-[normal] text-(--context-chip-ink) [--context-chip-ink:var(--context-chip-accent)] [--document-icon-stroke:var(--context-chip-ink)] [&_svg]:block [&_svg]:size-[1em] [&_svg]:shrink-0 [&_svg]:self-center [button&,a&,[data-popup-open]&]:cursor-pointer focus-visible:outline-2 focus-visible:-outline-offset-2 focus-visible:outline-current rounded-sm disabled:cursor-default", { defaultVariants: { kind: "neutral" }, variants: { kind: { - neutral: "border-border/70 bg-accent/40 text-foreground", - image: "[--context-chip-accent:oklch(0.62_0.16_16)]", - video: "[--context-chip-accent:oklch(0.62_0.16_48)]", - file: "[--context-chip-accent:oklch(0.62_0.136_237)]", - mention: "[--context-chip-accent:oklch(0.62_0.11_215)]", - terminal: "[--context-chip-accent:oklch(0.62_0.134_163)]", - element: "[--context-chip-accent:oklch(0.62_0.134_70)]", - "preview-annotation": "[--context-chip-accent:oklch(0.62_0.134_70)]", - "review-comment": "[--context-chip-accent:oklch(0.62_0.16_292)]", - "pull-request": "[--context-chip-accent:oklch(0.62_0.16_277)]", - "pr-open": "[--context-chip-accent:oklch(0.62_0.134_163)]", - "pr-draft": "[--context-chip-accent:oklch(0.62_0.02_259)]", - "pr-merged": "[--context-chip-accent:oklch(0.62_0.16_292)]", - "pr-closed": "[--context-chip-accent:oklch(0.62_0.16_16)]", - skill: "[--context-chip-accent:oklch(0.62_0.16_322)]", - thread: "[--context-chip-accent:oklch(0.62_0.12_190)]", - citation: "[--context-chip-accent:oklch(0.62_0.16_259)]", + neutral: "[--context-chip-accent:currentColor]", + image: "[--context-chip-accent:var(--context-chip-image)]", + video: "[--context-chip-accent:var(--context-chip-video)]", + file: "[--context-chip-accent:var(--context-chip-file)]", + mention: "[--context-chip-accent:var(--context-chip-mention)]", + terminal: "[--context-chip-accent:var(--context-chip-terminal)]", + element: "[--context-chip-accent:var(--context-chip-element)]", + "preview-annotation": "[--context-chip-accent:var(--context-chip-preview-annotation)]", + "review-comment": "[--context-chip-accent:var(--context-chip-review-comment)]", + "pull-request": "[--context-chip-accent:var(--context-chip-pull-request)]", + "pr-open": "[--context-chip-accent:var(--context-chip-pr-open)]", + "pr-draft": "[--context-chip-accent:var(--context-chip-pr-draft)]", + "pr-merged": "[--context-chip-accent:var(--context-chip-pr-merged)]", + "pr-closed": "[--context-chip-accent:var(--context-chip-pr-closed)]", + skill: "[--context-chip-accent:var(--context-chip-skill)]", + thread: "[--context-chip-accent:var(--context-chip-thread)]", + citation: "[--context-chip-accent:var(--context-chip-citation)]", + "doc-pdf": "[--context-chip-accent:var(--context-chip-doc-pdf)]", + "doc-word": "[--context-chip-accent:var(--context-chip-doc-word)]", + "doc-spreadsheet": "[--context-chip-accent:var(--context-chip-doc-spreadsheet)]", + "doc-slides": "[--context-chip-accent:var(--context-chip-doc-slides)]", + "doc-epub": "[--context-chip-accent:var(--context-chip-doc-epub)]", }, state: { - unresolved: "border-dashed", - invalid: "", + unresolved: "data-[state=unresolved]:[--context-chip-accent:currentColor]", + invalid: + "data-[state=invalid]:[--context-chip-accent:var(--context-chip-invalid,var(--destructive-foreground))]", }, }, - compoundVariants: [ - { - kind: [ - "image", - "video", - "file", - "mention", - "terminal", - "element", - "preview-annotation", - "review-comment", - "pull-request", - "pr-open", - "pr-draft", - "pr-merged", - "pr-closed", - "skill", - "thread", - "citation", - ], - className: - "[--context-chip-border:color-mix(in_oklab,var(--context-chip-accent)_34%,var(--contrast-border))] [--context-chip-border-hover:color-mix(in_oklab,var(--context-chip-accent)_48%,var(--contrast-border))] [--context-chip-foreground:color-mix(in_oklab,var(--context-chip-accent)_22%,var(--contrast-foreground))] border-(--context-chip-border) bg-(--context-chip-accent)/11 text-(--context-chip-foreground) [button:enabled&,a&]:hover:border-(--context-chip-border-hover) [button:enabled&,a&]:hover:bg-(--context-chip-accent)/17", - }, - { state: "unresolved", className: "text-foreground" }, - { state: "invalid", className: "border-destructive/35 bg-destructive/8 text-destructive" }, - ], }, ); @@ -86,7 +66,10 @@ function ContextChip({ className, kind, state, render, ...props }: ContextChipPr function ContextChipLabel({ className, ...props }: React.ComponentProps<"span">) { return ( &]:decoration-current", + className, + )} data-slot="context-chip-label" {...props} /> @@ -96,7 +79,7 @@ function ContextChipLabel({ className, ...props }: React.ComponentProps<"span">) function ContextChipAction({ className, render, ...props }: useRender.ComponentProps<"button">) { const defaultProps = { className: cn( - "ml-[0.17em] inline-flex size-[1.17em] shrink-0 cursor-pointer items-center justify-center rounded-sm text-current transition-colors hover:bg-(--context-chip-accent,var(--color-foreground))/17 focus-visible:outline-none focus-visible:ring-1 focus-visible:ring-inset focus-visible:ring-ring motion-reduce:transition-none [&_svg]:size-[0.85em]", + "ml-[0.17em] inline-flex size-[1em] shrink-0 cursor-pointer items-center justify-center rounded-sm text-current transition-colors hover:bg-(--context-chip-accent,var(--color-foreground))/17 focus-visible:outline-none focus-visible:ring-1 focus-visible:ring-inset focus-visible:ring-ring motion-reduce:transition-none [&_svg]:size-[0.85em]", className, ), "data-slot": "context-chip-action", diff --git a/apps/web/src/components/ThreadContextChip.tsx b/apps/web/src/components/ThreadContextChip.tsx index 5f1a4202f1d3..715a210f4e91 100644 --- a/apps/web/src/components/ThreadContextChip.tsx +++ b/apps/web/src/components/ThreadContextChip.tsx @@ -30,7 +30,7 @@ export function ThreadContextChip(props: { className="no-underline" > {project ? ( - + ) : ( )} diff --git a/apps/web/src/components/chat/AssistantCitationChip.tsx b/apps/web/src/components/chat/AssistantCitationChip.tsx index dd21f7c46684..1e97f00319c2 100644 --- a/apps/web/src/components/chat/AssistantCitationChip.tsx +++ b/apps/web/src/components/chat/AssistantCitationChip.tsx @@ -124,7 +124,7 @@ export function AssistantCitationChip({ const composerSourceLink = (