diff --git a/apps/server/src/orchestration-v2/Adapters/AcpAdapterV2.test.ts b/apps/server/src/orchestration-v2/Adapters/AcpAdapterV2.test.ts index a58a83bffa349..a949b0d9ed40f 100644 --- a/apps/server/src/orchestration-v2/Adapters/AcpAdapterV2.test.ts +++ b/apps/server/src/orchestration-v2/Adapters/AcpAdapterV2.test.ts @@ -758,7 +758,10 @@ describe("AcpAdapterV2", () => { Effect.sync(() => { watchedSessions.push(sessionId); const failed = watchedSessions.length === 1; - return () => Effect.succeed(failed ? quota : undefined); + return () => + Effect.succeed( + failed ? { failure: quota, threadDisposition: "reusable" as const } : undefined, + ); }), makeRuntime: makeMockRuntime({ childProcessSpawner: yield* ChildProcessSpawner.ChildProcessSpawner, @@ -824,6 +827,97 @@ describe("AcpAdapterV2", () => { assert.equal(watchedSessions[0], watchedSessions[1]); }).pipe(Effect.provide(layerTest), Effect.scoped), ); + it.live("marks the thread broken when the outcome check reports unusable native history", () => + Effect.gen(function* () { + const path = yield* Path.Path; + const instanceId = ProviderInstanceId.make("acp-test-prompt-outcome-broken"); + const threadId = ThreadId.make("thread-acp-prompt-outcome-broken"); + const failure = { + class: "provider_error", + message: "Antigravity stopped this turn because of an error.", + code: null, + retryable: null, + } as const; + const adapter = yield* makeAcpAdapterV2({ + instanceId, + selfInvocation: yield* resolveSelfInvocation(), + flavor: { + driver: ACP_TEST_DRIVER, + capabilities: AcpProviderCapabilitiesV2, + watchPromptOutcome: () => + Effect.succeed(() => Effect.succeed({ failure, threadDisposition: "broken" as const })), + makeRuntime: makeMockRuntime({ + childProcessSpawner: yield* ChildProcessSpawner.ChildProcessSpawner, + mockAgentPath: yield* path.fromFileUrl( + new URL("../../../scripts/acp-mock-agent.ts", import.meta.url), + ), + wrapRuntime: (runtime) => ({ + ...runtime, + prompt: () => Effect.succeed({ stopReason: "end_turn" as const }), + }), + }), + }, + }); + 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-prompt-outcome-broken"), + modelSelection, + runtimePolicy, + }); + const providerThread = yield* runtime.ensureThread({ + threadId, + modelSelection, + runtimePolicy, + }); + yield* runtime.startTurn( + makeTurnInput({ + threadId, + providerThread, + instanceId, + runtimePolicy, + now: yield* DateTime.now, + ordinal: 1, + }), + ); + const events = Array.from( + yield* runtime.events.pipe( + Stream.takeUntil((event) => event.type === "turn.terminal"), + Stream.runCollect, + ), + ); + + const terminal = events.at(-1); + if (terminal?.type !== "turn.terminal" || terminal.status !== "failed") { + return assert.fail("expected a failed terminal"); + } + assert.equal(terminal.threadDisposition, "broken"); + assert.deepEqual(terminal.failure, failure); + const updated = events.findLast((event) => event.type === "provider_thread.updated"); + assert.equal( + updated?.type === "provider_thread.updated" ? updated.providerThread.status : null, + "error", + ); + // The turn-start service recovers in a fresh native session when resume refuses. + const resumed = yield* Effect.result( + runtime.resumeThread({ + threadId, + providerThread: { ...providerThread, status: "error" }, + modelSelection, + runtimePolicy, + }), + ); + assert.equal( + resumed._tag === "Failure" ? resumed.failure._tag : null, + "ProviderAdapterResumeThreadError", + ); + }).pipe(Effect.provide(layerTest), Effect.scoped), + ); it.live("retains returned usage when Stop finalizes before the prompt callback", () => Effect.gen(function* () { const path = yield* Path.Path; diff --git a/apps/server/src/orchestration-v2/Orchestrator.control-reads.test.ts b/apps/server/src/orchestration-v2/Orchestrator.control-reads.test.ts index c806794732726..7104b2fbb552b 100644 --- a/apps/server/src/orchestration-v2/Orchestrator.control-reads.test.ts +++ b/apps/server/src/orchestration-v2/Orchestrator.control-reads.test.ts @@ -1430,6 +1430,92 @@ it.effect.each([false, true])("rejects thread.create for an existing id (deleted }).pipe(Effect.provide(layerTest)), ); +it.effect("keeps a broken native thread in error when a queued message starts", () => + Effect.gen(function* () { + const orchestrator = yield* Orchestrator.OrchestratorV2; + const sink = yield* EventSink.EventSinkV2; + const threadId = ThreadId.make("thread:broken-native-thread"); + const send = (id: string, dispatchMode: "start_immediately" | "queue_after_active") => + orchestrator.dispatch({ + type: "message.dispatch", + commandId: CommandId.make(`send:broken-native-thread:${id}`), + threadId, + messageId: MessageId.make(`message:broken-native-thread:${id}`), + text: id, + attachments: [], + dispatchMode: { type: dispatchMode }, + createdBy: "user", + creationSource: "web", + }); + yield* orchestrator.dispatch({ + type: "thread.create", + commandId: CommandId.make("create:broken-native-thread"), + threadId, + projectId: ProjectId.make("project:broken-native-thread"), + title: "Broken", + modelSelection, + runtimeMode: "full-access", + interactionMode: "default", + branch: null, + worktreePath: null, + createdBy: "user", + creationSource: "web", + }); + yield* send("first", "start_immediately"); + yield* send("queued", "queue_after_active"); + const { runs, providerThreads } = yield* orchestrator.getThreadRecords(threadId, [ + "runs", + "providerThreads", + ]); + const [failed, queued] = runs; + const [providerThread] = providerThreads; + const occurredAt = yield* DateTime.now; + // The first turn ended with its native history unusable. + yield* sink.write({ + events: [ + { + id: EventId.make("event:broken-native-thread:run"), + type: "run.updated", + threadId, + runId: failed!.id, + providerInstanceId: instanceId, + occurredAt, + payload: { ...failed!, status: "failed", completedAt: occurredAt }, + }, + { + id: EventId.make("event:broken-native-thread:provider-thread"), + type: "provider-thread.updated", + threadId, + providerInstanceId: instanceId, + occurredAt, + payload: { + ...providerThread!, + nativeThreadRef: { + driver: providerThread!.driver, + nativeId: "poisoned-native-history", + strength: "strong", + }, + status: "error", + }, + }, + ], + }); + + yield* orchestrator.dispatch({ + type: "queue.resume", + commandId: CommandId.make("resume:broken-native-thread"), + threadId, + }); + const next = yield* orchestrator.getThreadRecords(threadId, ["runs", "providerThreads"]); + const started = next.runs.find((run) => run.id === queued!.id); + assert.equal(started?.status, "starting"); + assert.equal( + next.providerThreads.find((thread) => thread.id === started?.providerThreadId)?.status, + "error", + ); + }).pipe(Effect.provide(layerTest)), +); + it.effect("replays a thread.create receipt without changing the thread", () => Effect.gen(function* () { const orchestrator = yield* Orchestrator.OrchestratorV2; diff --git a/apps/server/src/orchestration-v2/Orchestrator.ts b/apps/server/src/orchestration-v2/Orchestrator.ts index 16256585dc526..6bc2aa2057834 100644 --- a/apps/server/src/orchestration-v2/Orchestrator.ts +++ b/apps/server/src/orchestration-v2/Orchestrator.ts @@ -1838,6 +1838,7 @@ const makeOrchestrator = Effect.fn("orchestrationV2.Orchestrator.layer")(functio nativeThreadRef: activeProviderThread.nativeThreadRef, nativeConversationHeadRef: activeProviderThread.nativeConversationHeadRef, nativeMetadata: activeProviderThread.nativeMetadata, + status: activeProviderThread.status, } : queuedProviderThread; const targetCapabilities = yield* targetAdapter.getCapabilities().pipe( @@ -2021,7 +2022,8 @@ const makeOrchestrator = Effect.fn("orchestrationV2.Orchestrator.layer")(functio const providerThread: OrchestrationV2ProviderThread = { ...deliveryProviderThread, providerSessionId, - status: "not_loaded", + // A broken native history must reach the turn start, which replaces it. + status: deliveryProviderThread.status === "error" ? "error" : "not_loaded", firstRunOrdinal: queuedProviderThread.firstRunOrdinal ?? ordinal, lastRunOrdinal: ordinal, handoffIds: appendContextHandoffId( @@ -5168,7 +5170,7 @@ const makeOrchestrator = Effect.fn("orchestrationV2.Orchestrator.layer")(functio ...providerThread, providerInstanceId: input.modelSelection.instanceId, providerSessionId: nextProviderSessionId, - status: "not_loaded", + status: providerThread.status === "error" ? "error" : "not_loaded", updatedAt: now, }; restartSessionTransition = { @@ -7029,7 +7031,7 @@ const makeOrchestrator = Effect.fn("orchestrationV2.Orchestrator.layer")(functio : null; const providerThread: OrchestrationV2ProviderThread = { ...ensuredProviderThread, - status: "active", + status: ensuredProviderThread.status === "error" ? "error" : "active", firstRunOrdinal: ensuredProviderThread.firstRunOrdinal ?? ordinal, lastRunOrdinal: ordinal, handoffIds: [ diff --git a/apps/server/src/orchestration-v2/ProviderTurnStartService.test.ts b/apps/server/src/orchestration-v2/ProviderTurnStartService.test.ts index f7f9dc25ddc92..80d14ced8cbaa 100644 --- a/apps/server/src/orchestration-v2/ProviderTurnStartService.test.ts +++ b/apps/server/src/orchestration-v2/ProviderTurnStartService.test.ts @@ -1747,11 +1747,32 @@ effectIt.effect("does not mistake a failed state read for a superseded run", () }), ); -effectIt.effect( - "signs out the existing native provider before opening the newly selected provider", - () => +effectIt.effect.each(["not_loaded", "error"] as const)( + "signs out the existing native provider before opening the newly selected provider with status %s", + (providerThreadStatus) => Effect.gen(function* () { const harness = makeLocalCommandHarness({ text: "/logout", previousNativeSession: true }); + const initial = harness.projection(); + const target = initial.providerThreads.find((thread) => thread.id === "new-provider-thread")!; + harness.setProjection({ + ...initial, + providerThreads: initial.providerThreads.map((thread) => + thread.id === target.id + ? { + ...thread, + status: providerThreadStatus, + nativeThreadRef: + providerThreadStatus === "error" + ? { + driver: thread.driver, + nativeId: "poisoned-native-history", + strength: "strong" as const, + } + : thread.nativeThreadRef, + } + : thread, + ), + }); yield* harness.start; yield* harness.start; @@ -1777,6 +1798,9 @@ effectIt.effect( ]); expect(projection.providerTurns).toEqual([]); expect(projection.checkpoints).toEqual([]); + expect(projection.providerThreads.find((thread) => thread.id === target.id)?.status).toBe( + providerThreadStatus, + ); }), ); diff --git a/apps/server/src/orchestration-v2/ProviderTurnStartService.ts b/apps/server/src/orchestration-v2/ProviderTurnStartService.ts index cd121c7115a2a..b6d5afffb85cd 100644 --- a/apps/server/src/orchestration-v2/ProviderTurnStartService.ts +++ b/apps/server/src/orchestration-v2/ProviderTurnStartService.ts @@ -627,7 +627,12 @@ export const layer: Layer.Layer< }, providerThreadUpdate: { ...providerThread, - status: providerThread.nativeThreadRef === null ? "not_loaded" : "idle", + status: + providerThread.status === "error" + ? "error" + : providerThread.nativeThreadRef === null + ? "not_loaded" + : "idle", updatedAt: now, }, }); diff --git a/apps/server/src/provider/acp/AntigravityTurnOutcome.test.ts b/apps/server/src/provider/acp/AntigravityTurnOutcome.test.ts index 09b8730bd6001..244b54ac13fe1 100644 --- a/apps/server/src/provider/acp/AntigravityTurnOutcome.test.ts +++ b/apps/server/src/provider/acp/AntigravityTurnOutcome.test.ts @@ -6,6 +6,7 @@ import * as Effect from "effect/Effect"; import * as FileSystem from "effect/FileSystem"; import * as Path from "effect/Path"; import type * as EffectAcpSchema from "effect-acp/compat"; +import type { OrchestrationV2ProviderFailure } from "@t3tools/contracts"; import { watchAntigravityPromptOutcome } from "./AntigravityTurnOutcome.ts"; @@ -32,10 +33,16 @@ const RATE_LIMIT_ERROR = errorPayload( const INVALID_ARGUMENT_ERROR = errorPayload( "Error 400, Message: Request contains an invalid argument., Status: INVALID_ARGUMENT, Details: []", ); +// Context summarization replays a stored failed CHECKPOINT step on every later prompt. +const FAILED_CHECKPOINT_ERROR = errorPayload( + 'context summarization: checkpoint step 841 failed: error executing cascade step: CORTEX_STEP_TYPE_CHECKPOINT: doRequest: error sending request: Post "https://aiplatform.googleapis.com/v1beta1/projects/test-project/locations/global/publishers/google/models/gemini-4-argon:streamGenerateContent?alt=sse": auth: "invalid_grant" "reauth related error (invalid_rapt)"', +); // The model's reply quoting the agent's quota notice is an ordinary planner response. const QUOTED_NOTICE = Buffer.from( "Usage Limit Reached\n\nYou have reached your current quota for this period. Your limit will reset in 1 hour.", ); +const reusable = (failure: OrchestrationV2ProviderFailure) => + ({ failure, threadDisposition: "reusable" }) as const; describe("watchAntigravityPromptOutcome", () => { it.effect("fails only turns whose own steps end in an upstream error", () => @@ -83,13 +90,13 @@ describe("watchAntigravityPromptOutcome", () => { [USER_INPUT, input], [ERROR_MESSAGE, QUOTA_ERROR], ]), - { + reusable({ class: "usage_limit", message: "You have reached your Antigravity quota for this period.", code: "RESOURCE_EXHAUSTED", retryable: true, resetAt: "1970-01-01T01:00:00.000Z", - }, + }), ); // The earlier error step is outside this turn. assert.isUndefined( @@ -109,25 +116,25 @@ describe("watchAntigravityPromptOutcome", () => { [USER_INPUT, input], [ERROR_MESSAGE, RATE_LIMIT_ERROR], ]), - { + reusable({ class: "usage_limit", message: "Resource has been exhausted (e.g. check quota).", code: "RESOURCE_EXHAUSTED", retryable: true, resetAt: "1970-01-01T00:00:42.000Z", - }, + }), ); assert.deepEqual( yield* turn([ [USER_INPUT, input], [ERROR_MESSAGE, INVALID_ARGUMENT_ERROR], ]), - { + reusable({ class: "provider_error", message: "Request contains an invalid argument.", code: "INVALID_ARGUMENT", retryable: false, - }, + }), ); // A completed tool keeps its usage, and the turn still fails. assert.equal( @@ -142,7 +149,7 @@ describe("watchAntigravityPromptOutcome", () => { stopReason: "end_turn", usage: { inputTokens: 40, outputTokens: 5, totalTokens: 45 }, }, - ))?.class, + ))?.failure.class, "usage_limit", ); assert.isUndefined( @@ -152,12 +159,15 @@ describe("watchAntigravityPromptOutcome", () => { ]), ); // The agent maps the SDK's QUOTA_EXHAUSTED stop reason to `refusal`. - assert.deepEqual(yield* turn([], { stopReason: "refusal" }), { - class: "usage_limit", - message: "You have reached your Antigravity quota for this period.", - code: "QUOTA_EXHAUSTED", - retryable: true, - }); + assert.deepEqual( + yield* turn([], { stopReason: "refusal" }), + reusable({ + class: "usage_limit", + message: "You have reached your Antigravity quota for this period.", + code: "QUOTA_EXHAUSTED", + retryable: true, + }), + ); }).pipe(Effect.provide(NodeServices.layer), Effect.scoped), ); @@ -193,19 +203,72 @@ describe("watchAntigravityPromptOutcome", () => { const linked = yield* watchAntigravityPromptOutcome(profileDirectory, own); append(peer, ERROR_MESSAGE, QUOTA_ERROR); assert.isUndefined(yield* linked({ stopReason: "end_turn" })); - assert.equal((yield* linked({ stopReason: "refusal" }))?.class, "usage_limit"); + assert.equal((yield* linked({ stopReason: "refusal" }))?.failure.class, "usage_limit"); // An unreadable log gives no evidence, but the agent's typed refusal still fails the turn. yield* fs.remove(path.join(conversations, `${own}.db`)); yield* fs.writeFileString(path.join(conversations, `${own}.db`), "not a database"); const corrupt = yield* watchAntigravityPromptOutcome(profileDirectory, own); assert.isUndefined(yield* corrupt({ stopReason: "end_turn" })); - assert.deepEqual(yield* corrupt({ stopReason: "refusal" }), { - class: "usage_limit", - message: "You have reached your Antigravity quota for this period.", - code: "QUOTA_EXHAUSTED", - retryable: true, - }); + assert.deepEqual( + yield* corrupt({ stopReason: "refusal" }), + reusable({ + class: "usage_limit", + message: "You have reached your Antigravity quota for this period.", + code: "QUOTA_EXHAUSTED", + retryable: true, + }), + ); + }).pipe(Effect.provide(NodeServices.layer), Effect.scoped), + ); + + it.effect("marks the native thread broken when a failed checkpoint step is replayed", () => + Effect.gen(function* () { + const fs = yield* FileSystem.FileSystem; + const path = yield* Path.Path; + const profileDirectory = yield* fs.makeTempDirectoryScoped(); + const sessionId = "e6aba774-5472-4277-b6ee-27b2e05570a7"; + const conversations = path.join(profileDirectory, "antigravity-acp", "conversations"); + yield* fs.makeDirectory(conversations, { recursive: true }); + const check = yield* watchAntigravityPromptOutcome(profileDirectory, sessionId); + const database = new NodeSqlite.DatabaseSync(path.join(conversations, `${sessionId}.db`)); + try { + database.exec( + "CREATE TABLE steps (idx INTEGER PRIMARY KEY, step_type INTEGER, step_payload BLOB)", + ); + const insert = database.prepare("INSERT INTO steps VALUES (?, ?, ?)"); + insert.run(843, USER_INPUT, Buffer.from("Continue.")); + insert.run(844, ERROR_MESSAGE, FAILED_CHECKPOINT_ERROR); + } finally { + database.close(); + } + + const outcome = yield* check({ stopReason: "end_turn" }); + assert.equal(outcome?.threadDisposition, "broken"); + assert.equal(outcome?.failure.class, "provider_error"); + + // A checkpoint that failed on quota still reports the quota and its reset time. + const quotaCheck = yield* watchAntigravityPromptOutcome(profileDirectory, sessionId); + const quotaDatabase = new NodeSqlite.DatabaseSync( + path.join(conversations, `${sessionId}.db`), + ); + try { + const insert = quotaDatabase.prepare("INSERT INTO steps VALUES (?, ?, ?)"); + insert.run(845, USER_INPUT, Buffer.from("Continue.")); + insert.run( + 846, + ERROR_MESSAGE, + errorPayload( + "context summarization: checkpoint step 841 failed: Error 429, Message: Quota exhausted for test model., Status: RESOURCE_EXHAUSTED, Details: [map[@type:type.googleapis.com/google.rpc.ErrorInfo domain:googleapis.com metadata:map[quotaResetDelay:3600s] reason:QUOTA_EXHAUSTED]]", + ), + ); + } finally { + quotaDatabase.close(); + } + const quotaOutcome = yield* quotaCheck({ stopReason: "end_turn" }); + assert.equal(quotaOutcome?.threadDisposition, "broken"); + assert.equal(quotaOutcome?.failure.class, "usage_limit"); + assert.notEqual(quotaOutcome?.failure.resetAt ?? null, null); }).pipe(Effect.provide(NodeServices.layer), Effect.scoped), ); }); diff --git a/apps/server/src/provider/acp/AntigravityTurnOutcome.ts b/apps/server/src/provider/acp/AntigravityTurnOutcome.ts index 00de1e836296e..bca75f1a23d1a 100644 --- a/apps/server/src/provider/acp/AntigravityTurnOutcome.ts +++ b/apps/server/src/provider/acp/AntigravityTurnOutcome.ts @@ -19,6 +19,9 @@ const MAX_ERROR_PAYLOAD_BYTES = 16_384; // The Go harness's rendering of a Google API error, as the agent's api_errors.py reads it. const API_ERROR = /Error (\d{3}), Message: (.+?), Status: ([A-Z_]+)/s; const QUOTA_EXHAUSTED = /\breason:QUOTA_EXHAUSTED\b/; +// The runtime stores a failed context-summarization step and replays it on every later +// prompt to that session, whatever first made it fail. +const FAILED_CHECKPOINT = /\bcontext summarization: checkpoint step \d+ failed\b/; const RESET_TIMESTAMP = /\bquotaResetTimeStamp:([^\s\]]+)/; const RESET_DELAY = /\bquotaResetDelay:((?:\d+(?:\.\d+)?(?:ns|us|µs|μs|ms|h|m|s))+)(?=[\s\]])/; // google.rpc.RetryInfo, which plain rate limits carry instead of a quota reset. @@ -126,6 +129,39 @@ function turnFailure( }); } +function turnOutcome( + stopReason: EffectAcpSchema.StopReason, + step: LastStep | undefined, + now: DateTime.Utc, +): + | { + readonly failure: OrchestrationV2ProviderFailure; + readonly threadDisposition: "reusable" | "broken"; + } + | undefined { + const failure = turnFailure(stopReason, step, now); + if (failure === undefined) return undefined; + const text = + step?.type === ERROR_STEP_TYPE + ? new TextDecoder().decode(step.payload ?? new Uint8Array()) + : ""; + if (!FAILED_CHECKPOINT.test(text)) return { failure, threadDisposition: "reusable" }; + return { + // A quota failure keeps its reset time so usage-limit recovery can resume the thread. + failure: + failure.class === "usage_limit" + ? failure + : makeProviderFailure({ + class: "provider_error", + message: + "Antigravity could not summarize this conversation, so its saved session can no longer continue. The next message starts a fresh Antigravity session with this thread's context.", + code: "CHECKPOINT_FAILED", + retryable: true, + }), + threadDisposition: "broken", + }; +} + const conversationSegments = ["antigravity-acp", "conversations"]; /** @@ -139,7 +175,7 @@ export const watchAntigravityPromptOutcome = Effect.fn("AntigravityTurnOutcome.w sessionId: string, ) { const typedOnly = (result: EffectAcpSchema.PromptResponse) => - DateTime.now.pipe(Effect.map((now) => turnFailure(result.stopReason, undefined, now))); + DateTime.now.pipe(Effect.map((now) => turnOutcome(result.stopReason, undefined, now))); if (!NATIVE_SESSION_ID.test(sessionId)) return typedOnly; const segments = [...conversationSegments, `${sessionId}.db`]; const before = yield* Effect.tryPromise(async () => { @@ -192,6 +228,6 @@ export const watchAntigravityPromptOutcome = Effect.fn("AntigravityTurnOutcome.w Effect.orElseSucceed(() => undefined), ) : undefined; - return turnFailure(result.stopReason, step, yield* DateTime.now); + return turnOutcome(result.stopReason, step, yield* DateTime.now); }); }); diff --git a/packages/provider-acp/src/server/adapter.ts b/packages/provider-acp/src/server/adapter.ts index 985f6c8ee9887..18a821fe17d24 100644 --- a/packages/provider-acp/src/server/adapter.ts +++ b/packages/provider-acp/src/server/adapter.ts @@ -271,6 +271,11 @@ export function acpPerTurnPromptUsage( }; } +interface AcpPromptOutcomeFailure { + readonly failure: OrchestrationV2ProviderFailure; + readonly threadDisposition: "reusable" | "broken"; +} + export interface AcpAdapterV2Flavor { readonly buildPrompt?: (input: { readonly text: string; @@ -285,14 +290,13 @@ export interface AcpAdapterV2Flavor { /** * Runs before `session/prompt`. The returned check sees a prompt the agent ended * without an error or cancel, and reports a failure its stop reason hid. A reported - * failure restarts the runtime before the next turn. + * failure restarts the runtime before the next turn. A `broken` thread's native + * history is unusable, so it is never resumed and the next turn starts fresh. */ readonly watchPromptOutcome?: ( sessionId: string, ) => Effect.Effect< - ( - result: EffectAcpSchema.PromptResponse, - ) => Effect.Effect + (result: EffectAcpSchema.PromptResponse) => Effect.Effect >; readonly requiresSignIn?: (cause: unknown) => boolean; /** @@ -7470,10 +7474,15 @@ export const makeAcpAdapterV2 = Effect.fn("makeAcpAdapterV2")(function* ( context: ActiveAcpTurn, status: "completed" | "interrupted" | "failed" | "cancelled", failure?: OrchestrationV2ProviderFailure, - options?: { readonly childTurnsContinue?: boolean }, + options?: { + readonly childTurnsContinue?: boolean; + readonly threadDisposition?: "reusable" | "broken"; + }, ): Effect.fn.Return { if (context.finalized) return; const settledStatus = context.interrupted ? "interrupted" : status; + const threadDisposition = + settledStatus === "failed" ? (options?.threadDisposition ?? "reusable") : "reusable"; context.finalizedStatus = settledStatus; context.finalized = true; const directStopQuarantine = yield* Ref.get(stoppedRunQuarantine); @@ -7552,7 +7561,7 @@ export const makeAcpAdapterV2 = Effect.fn("makeAcpAdapterV2")(function* ( const updatedProviderThread: OrchestrationV2ProviderThread = { ...context.input.providerThread, providerSessionId: input.providerSessionId, - status: "active", + status: threadDisposition === "broken" ? "error" : "active", lastRunOrdinal: context.input.runOrdinal, firstRunOrdinal: context.input.providerThread.firstRunOrdinal ?? context.input.runOrdinal, @@ -7596,7 +7605,7 @@ export const makeAcpAdapterV2 = Effect.fn("makeAcpAdapterV2")(function* ( retry: context.providerRetry.retry, retryStartedAt: context.providerRetry.startedAt, }), - threadDisposition: "reusable", + threadDisposition, } : { type: "turn.terminal", @@ -8190,7 +8199,8 @@ export const makeAcpAdapterV2 = Effect.fn("makeAcpAdapterV2")(function* ( yield* Ref.set(runtimeRestartRequired, true); } const failure = - hiddenFailure ?? (idleError ? idleErrorFailure(result.error) : undefined); + hiddenFailure?.failure ?? + (idleError ? idleErrorFailure(result.error) : undefined); const status = failure !== undefined ? "failed" @@ -8213,6 +8223,9 @@ export const makeAcpAdapterV2 = Effect.fn("makeAcpAdapterV2")(function* ( // background tools and subagents keep reporting to this run. yield* finalizeTurn(context, "failed", failure, { childTurnsContinue: hasSurvivingBackgroundWork(context), + ...(hiddenFailure === undefined + ? {} + : { threadDisposition: hiddenFailure.threadDisposition }), }); return; } @@ -8477,6 +8490,10 @@ export const makeAcpAdapterV2 = Effect.fn("makeAcpAdapterV2")(function* ( readonly modelSelection?: ModelSelection; readonly runtimePolicy?: ProviderAdapter.ProviderAdapterV2RuntimePolicy; }) { + if (threadInput.providerThread.status === "error") { + // A broken turn left this native history unusable; the caller recovers in a fresh session. + return yield* Effect.fail("The native session history is unusable."); + } return yield* runtimeTransitionPermit.withPermit( Effect.gen(function* () { yield* awaitRuntimeTeardown();