diff --git a/apps/server/src/orchestration-v2/ProjectionStore.test.ts b/apps/server/src/orchestration-v2/ProjectionStore.test.ts index e8a3c1bcfc08..c58add1181fc 100644 --- a/apps/server/src/orchestration-v2/ProjectionStore.test.ts +++ b/apps/server/src/orchestration-v2/ProjectionStore.test.ts @@ -2461,6 +2461,7 @@ it.layer(layerTest)("ProjectionStoreV2", (it) => { const assertSummary = Effect.fnUntraced(function* ( lastError: string | null, lastErrorClass: string | null, + lastErrorAt: string | null = lastError === null ? null : DateTime.formatIso(now), ) { const projection = yield* store.getThreadProjection(threadId); const memoryShell = ProjectionStore.threadShellFromProjection(projection); @@ -2468,6 +2469,7 @@ it.layer(layerTest)("ProjectionStoreV2", (it) => { const sqlShell = shells.threads.find((row) => row.id === threadId)!; for (const shell of [memoryShell, sqlShell]) { assert.equal(shell.lastError, lastError); + assert.equal(shell.lastErrorAt, lastErrorAt); assert.equal(shell.lastErrorClass, lastErrorClass); assert.equal( shell.usageLimitResetAt, @@ -2733,9 +2735,22 @@ it.layer(layerTest)("ProjectionStoreV2", (it) => { type: "provider-session.updated", threadId, occurredAt: now, + payload: { + ...session, + status: "error", + lastError: "Provider process exited.", + lastErrorAt: DateTime.makeUnsafe("2026-09-20T02:00:00.000Z"), + }, + }); + yield* assertSummary("Provider process exited.", null, "2026-09-20T02:00:00.000Z"); + yield* store.apply({ + id: EventId.make("event:limit-shell:legacy-error"), + type: "provider-session.updated", + threadId, + occurredAt: now, payload: { ...session, status: "error", lastError: "Provider process exited." }, }); - yield* assertSummary("Provider process exited.", null); + yield* assertSummary("Provider process exited.", null, null); yield* store.apply({ id: EventId.make("event:limit-shell:session-recovered"), type: "provider-session.updated", diff --git a/apps/server/src/orchestration-v2/ProjectionStore.ts b/apps/server/src/orchestration-v2/ProjectionStore.ts index 8c33faef808b..6606026540e2 100644 --- a/apps/server/src/orchestration-v2/ProjectionStore.ts +++ b/apps/server/src/orchestration-v2/ProjectionStore.ts @@ -7,8 +7,9 @@ import type { OrchestrationV2ThreadHistoryPage, } from "@t3tools/contracts"; import { - latestRootProviderFailure, + latestRootProviderFailureItem, latestUnheldRun, + providerFailureOccurredAt, threadErrorSummary, usageLimitRunPresentedAsLatest, } from "@t3tools/shared/orchestrationV2ThreadError"; @@ -955,7 +956,7 @@ type ShellThreadRow = { readonly active_run_id: string | null; readonly activity_run_status: string | null; readonly activity_run_started_at: string | null; - readonly last_error: string | null; + readonly error_state_json: string | null; readonly terminal_failure_payload_json: string | null; readonly blocking_run_id: string | null; readonly blocking_run_requested_at: string | null; @@ -1433,6 +1434,7 @@ export function threadShellFromProjection( runs: projection.runs, pullRequests: projection.thread.pullRequests, }); + const failureItem = latestRootProviderFailureItem(latestRun, projection.turnItems); return { createdBy: projection.thread.createdBy, creationSource: projection.thread.creationSource, @@ -1470,10 +1472,13 @@ export function threadShellFromProjection( activityRunStartedAt: activityRun === null ? null : orchestrationV2RunWorkStartedAt(activityRun), status: latestRun?.status ?? "idle", - ...threadErrorSummary( - latestRootProviderFailure(latestRun, projection.turnItems), - providerSession?.lastError ?? null, - ), + ...threadErrorSummary(failureItem?.failure ?? null, providerSession?.lastError ?? null, { + sessionErrorAt: + providerSession?.lastErrorAt == null + ? null + : DateTime.formatIso(providerSession.lastErrorAt), + failureAt: providerFailureOccurredAt(failureItem), + }), pendingRuntimeRequest: pendingRuntimeRequest === null ? null @@ -1569,6 +1574,15 @@ function isActivityRunForShell( return isInterruptibleRunForShell(run) || run.status === "waiting"; } +const decodeShellErrorState = Schema.decodeUnknownEffect( + Schema.fromJsonString( + Schema.Struct({ + lastError: Schema.NullOr(Schema.String), + lastErrorAt: Schema.NullOr(Schema.String), + }), + ), +); + type ShellThreadState = { readonly thread: OrchestrationV2ThreadProjection["thread"]; readonly latestRunId: RunId | null; @@ -1580,6 +1594,7 @@ type ShellThreadState = { readonly activityRunStatus: ShellActivityRunStatus | null; readonly activityRunStartedAt: DateTime.Utc | null; readonly lastError: string | null; + readonly lastErrorAt: string | null; readonly lastErrorClass: OrchestrationV2ThreadShell["lastErrorClass"]; readonly usageLimitResetAt: OrchestrationV2ThreadShell["usageLimitResetAt"]; readonly pendingRuntimeRequest: OrchestrationV2ThreadProjection["runtimeRequests"][number] | null; @@ -1738,6 +1753,7 @@ function shellFromState(input: { activityRunStartedAt: input.state.activityRunStartedAt, status: input.state.latestRunStatus, lastError: input.state.lastError, + lastErrorAt: input.state.lastErrorAt, lastErrorClass: input.state.lastErrorClass, usageLimitResetAt: input.state.usageLimitResetAt, pendingRuntimeRequest: @@ -5160,7 +5176,10 @@ export const layer: Layer.Layer = LIMIT 1 ) AS activity_run_started_at, ( - SELECT json_extract(session.payload_json, '$.lastError') + SELECT json_object( + 'lastError', json_extract(session.payload_json, '$.lastError'), + 'lastErrorAt', json_extract(session.payload_json, '$.lastErrorAt') + ) FROM orchestration_v2_projection_provider_sessions session INNER JOIN orchestration_v2_projection_provider_session_bindings binding ON binding.provider_session_id = session.provider_session_id @@ -5168,7 +5187,7 @@ export const layer: Layer.Layer = AND session.provider_instance_id = t.provider_instance_id ORDER BY session.updated_at DESC, session.provider_session_id DESC LIMIT 1 - ) AS last_error, + ) AS error_state_json, ( SELECT item.payload_json FROM orchestration_v2_projection_turn_items item @@ -5596,6 +5615,8 @@ export const layer: Layer.Layer = yield* decodeTurnItemPayload(row.pending_secret_request_payload_json), ) : null; + const sessionError = + row.error_state_json === null ? null : yield* decodeShellErrorState(row.error_state_json); let terminalFailureItem = row.terminal_failure_payload_json === null ? null @@ -5621,7 +5642,7 @@ export const layer: Layer.Layer = const blocksQueue = threadErrorSummary( blockingFailure?.type === "error" ? blockingFailure.failure : null, - row.last_error, + sessionError?.lastError ?? null, ).lastErrorClass === "usage_limit"; if (blocksQueue && blockingFailure !== null) { terminalFailureItem = blockingFailure; @@ -5679,7 +5700,13 @@ export const layer: Layer.Layer = : null, ...threadErrorSummary( terminalFailureItem?.type === "error" ? terminalFailureItem.failure : null, - row.last_error, + sessionError?.lastError ?? null, + { + sessionErrorAt: sessionError?.lastErrorAt ?? null, + failureAt: providerFailureOccurredAt( + terminalFailureItem?.type === "error" ? terminalFailureItem : null, + ), + }, ), pendingRuntimeRequest, latestUserMessageAt: diff --git a/apps/server/src/orchestration-v2/ProviderSessionManager.test.ts b/apps/server/src/orchestration-v2/ProviderSessionManager.test.ts index 139cc705a2bf..c307f14c49df 100644 --- a/apps/server/src/orchestration-v2/ProviderSessionManager.test.ts +++ b/apps/server/src/orchestration-v2/ProviderSessionManager.test.ts @@ -13,7 +13,7 @@ import { ProjectId, ProviderDriverKind, ProviderInstanceId, - type ProviderSessionId, + ProviderSessionId, ProviderTurnId, ThreadId, } from "@t3tools/contracts"; @@ -26,6 +26,7 @@ import * as Fiber from "effect/Fiber"; import * as FileSystem from "effect/FileSystem"; import * as Layer from "effect/Layer"; import * as Metric from "effect/Metric"; +import * as Logger from "effect/Logger"; import * as Option from "effect/Option"; import * as Queue from "effect/Queue"; import * as Ref from "effect/Ref"; @@ -470,6 +471,7 @@ function layerTest(input: { readonly armed: Ref.Ref; readonly paused: Deferred.Deferred; }; + readonly failProjectionReads?: Ref.Ref; readonly hasPendingBackgroundWork?: Effect.Effect; readonly hasPendingBackgroundWorkForThread?: Effect.Effect; readonly hasBufferedOutputForThread?: ( @@ -521,6 +523,32 @@ function layerTest(input: { input.pauseResolve === undefined ? layerTestMcpRegistry : layerPausingMcpRegistry(input.pauseResolve); + const layerReleaseStores = + input.failProjectionReads === undefined + ? layerTestStores + : Layer.effect( + ProjectionStore.ProjectionStoreV2, + Effect.gen(function* () { + const delegate = yield* ProjectionStore.ProjectionStoreV2; + const failReads = input.failProjectionReads!; + return ProjectionStore.ProjectionStoreV2.of({ + ...delegate, + getThreadProviderContext: (threadId, targetInstanceId) => + Ref.get(failReads).pipe( + Effect.flatMap((fail) => + fail + ? Effect.fail( + new ProjectionStore.ProjectionStoreReadError({ + threadId, + cause: "private error detail", + }), + ) + : delegate.getThreadProviderContext(threadId, targetInstanceId), + ), + ), + }); + }), + ).pipe(Layer.provide(layerTestStores)); const layerProviderEventIngestorTest = ProviderEventIngestor.layer.pipe( Layer.provide( Layer.mergeAll( @@ -548,7 +576,7 @@ function layerTest(input: { IdAllocator.layer, layerProviderEventIngestorTest, layerConfiguredMcpRegistry, - layerTestStores, + layerReleaseStores, ...(input.serverSettingsLayer === undefined ? [] : [input.serverSettingsLayer]), ...(input.projectServiceLayer === undefined ? [] : [input.projectServiceLayer]), ), @@ -3395,6 +3423,169 @@ it.effect( }), ); +it("preserves error occurrences on teardown and stamps a new runtime failure", () => { + const now = DateTime.makeUnsafe("2026-09-20T01:00:00Z"); + const later = DateTime.makeUnsafe("2026-09-20T02:00:00Z"); + const session = { + ...makeProviderSession({ + providerSessionId: ProviderSessionId.make("session:occurrence"), + now, + }), + lastError: "Failed", + lastErrorAt: now, + }; + for (const reason of ["idle_timeout", "manual_shutdown", "server_shutdown"] as const) { + const released = ProviderSessionManager.releasedProviderSession({ + session, + now: later, + reason, + }); + assert.equal(released.lastError, "Failed"); + assert.deepStrictEqual(released.lastErrorAt, now); + } + const failed = ProviderSessionManager.releasedProviderSession({ + session, + now: later, + reason: "runtime_error", + detail: "Failed", + }); + assert.deepStrictEqual(failed.lastErrorAt, later); +}); + +it("only preserves a Pi terminal error on clean stream end", () => { + const now = DateTime.makeUnsafe("2026-09-20T01:00:00Z"); + const later = DateTime.makeUnsafe("2026-09-20T02:00:00Z"); + for (const driver of ["opencode", "pi"] as const) { + for (const status of ["error", "ready"] as const) { + const session = { + ...makeProviderSession({ + providerSessionId: ProviderSessionId.make("session:stream-end"), + now, + }), + driver: ProviderDriverKind.make(driver), + status, + lastError: "Earlier turn failed", + lastErrorAt: null, + }; + const released = ProviderSessionManager.releasedProviderSession({ + session, + now: later, + reason: "runtime_error", + runtimeErrorCause: "stream_end", + detail: "Provider event stream ended.", + }); + const preserve = driver === "pi" && status === "error"; + assert.equal( + released.lastError, + preserve ? "Earlier turn failed" : "Provider event stream ended.", + ); + assert.deepStrictEqual(released.lastErrorAt, preserve ? null : later); + } + } +}); + +it.effect("ProviderSessionManagerV2 warns on projection read failure and still releases", () => + Effect.gen(function* () { + const state = yield* Ref.make(emptyState); + const failProjectionReads = yield* Ref.make(false); + const messages: Array<{ level: string; message: unknown }> = []; + const logger = Logger.make(({ logLevel, message }) => { + messages.push({ level: logLevel, message }); + }); + yield* Effect.gen(function* () { + const eventSink = yield* EventSink.EventSinkV2; + const idAllocator = yield* IdAllocator.IdAllocatorV2; + const manager = yield* ProviderSessionManager.ProviderSessionManagerV2; + const now = yield* DateTime.now; + const threadId = ThreadId.make("release-read-failure"); + const providerSessionId = ProviderSessionId.make("release-read-failure-session"); + yield* eventSink.write({ + events: [yield* makeThreadCreatedEvent({ idAllocator, threadId, now })], + }); + yield* manager.open({ threadId, providerSessionId, modelSelection, runtimePolicy }); + yield* Ref.set(failProjectionReads, true); + yield* manager.release({ providerSessionId, reason: "idle_timeout" }); + assert.isTrue(Option.isNone(yield* manager.get(providerSessionId))); + assert.equal((yield* Ref.get(state)).closeCount, 1); + assert.deepStrictEqual( + messages.filter((entry) => entry.level === "Warn"), + [ + { + level: "Warn", + message: [ + "orchestration-v2.driver-session.release-projection-read-failed", + { + threadId, + providerSessionId, + reason: "idle_timeout", + }, + ], + }, + ], + ); + }).pipe( + Effect.provide( + layerTest({ state, idleTimeoutMs: 60_000, failProjectionReads }).pipe( + Layer.provideMerge(Logger.layer([logger], { mergeWithExisting: false })), + ), + ), + ); + }), +); + +it.effect("ProviderSessionManagerV2 preserves the latest projected error during idle release", () => + Effect.gen(function* () { + const state = yield* Ref.make(emptyState); + yield* Effect.gen(function* () { + const eventSink = yield* EventSink.EventSinkV2; + const idAllocator = yield* IdAllocator.IdAllocatorV2; + const manager = yield* ProviderSessionManager.ProviderSessionManagerV2; + const projectionStore = yield* ProjectionStore.ProjectionStoreV2; + const now = yield* DateTime.now; + const projectId = yield* idAllocator.allocate.project({ fixtureName: "error-preservation" }); + const threadId = yield* idAllocator.allocate.thread({ + fixtureName: "error-preservation", + projectId, + }); + const providerSessionId = yield* idAllocator.allocate.providerSession({ + providerInstanceId: modelSelection.instanceId, + threadId, + }); + yield* eventSink.write({ + events: [yield* makeThreadCreatedEvent({ idAllocator, threadId, now })], + }); + const runtime = yield* manager.open({ + threadId, + providerSessionId, + modelSelection, + runtimePolicy, + }); + yield* eventSink.write({ + events: [ + { + id: yield* idAllocator.allocate.event({ threadId, providerSessionId }), + type: "provider-session.updated", + threadId, + driver: CODEX_DRIVER, + providerInstanceId: modelSelection.instanceId, + occurredAt: now, + payload: { + ...runtime.providerSession, + status: "error", + lastError: "Projected error", + lastErrorAt: now, + }, + }, + ], + }); + yield* manager.release({ providerSessionId, reason: "idle_timeout" }); + const after = yield* projectionStore.getThreadProviderContext(threadId); + assert.equal(after.providerSessions.at(-1)?.lastError, "Projected error"); + assert.deepStrictEqual(after.providerSessions.at(-1)?.lastErrorAt, now); + }).pipe(Effect.provide(layerTest({ state, idleTimeoutMs: 60_000 }))); + }), +); + it.effect("ProviderSessionManagerV2 uses the same release path for runtime failures", () => Effect.gen(function* () { const state = yield* Ref.make(emptyState); diff --git a/apps/server/src/orchestration-v2/ProviderSessionManager.ts b/apps/server/src/orchestration-v2/ProviderSessionManager.ts index 999e75cb6f52..7c21964bab10 100644 --- a/apps/server/src/orchestration-v2/ProviderSessionManager.ts +++ b/apps/server/src/orchestration-v2/ProviderSessionManager.ts @@ -285,6 +285,33 @@ function releaseStatusFor( return reason === "runtime_error" ? "error" : "stopped"; } +export function releasedProviderSession(input: { + readonly reason: ProviderSessionReleaseReason; + readonly session: OrchestrationV2ProviderSession; + readonly now: DateTime.Utc; + readonly detail?: string | undefined; + readonly runtimeErrorCause?: "stream_end" | undefined; +}): OrchestrationV2ProviderSession { + if (input.reason === "runtime_error") { + if ( + input.runtimeErrorCause === "stream_end" && + input.session.driver === "pi" && + input.session.status === "error" && + input.session.lastError !== null + ) { + return { ...input.session, updatedAt: input.now }; + } + return { + ...input.session, + status: "error", + updatedAt: input.now, + lastError: input.detail ?? "Provider runtime failed.", + lastErrorAt: input.now, + }; + } + return { ...input.session, status: releaseStatusFor(input.reason), updatedAt: input.now }; +} + function releasedRuntimeRequestStatusFor( reason: ProviderSessionReleaseReason, ): OrchestrationV2RuntimeRequest["status"] { @@ -668,18 +695,39 @@ export const layerWithOptions = ( readonly entry: LiveSessionEntry; readonly reason: ProviderSessionReleaseReason; readonly detail?: string; + readonly runtimeErrorCause?: "stream_end"; }) => Effect.gen(function* () { const now = yield* DateTime.now; - const payload: OrchestrationV2ProviderSession = { - ...input.entry.runtime.providerSession, - status: releaseStatusFor(input.reason), - updatedAt: now, - lastError: - input.reason === "runtime_error" - ? (input.detail ?? "Provider runtime failed.") - : null, - }; + let session = input.entry.runtime.providerSession; + for (const threadId of input.entry.attachedThreadIds) { + const context = yield* projectionStore.getThreadProviderContext(threadId).pipe( + Effect.catch(() => + Effect.logWarning( + "orchestration-v2.driver-session.release-projection-read-failed", + { + threadId, + providerSessionId: session.id, + reason: input.reason, + }, + ).pipe(Effect.as(null)), + ), + ); + const projected = context?.providerSessions.findLast( + (candidate) => candidate.id === session.id, + ); + if (projected !== undefined) { + session = projected; + break; + } + } + const payload = releasedProviderSession({ + session, + reason: input.reason, + now, + detail: input.detail, + runtimeErrorCause: input.runtimeErrorCause, + }); yield* writeProviderSessionEvents({ runtime: input.entry.runtime, threadIds: input.entry.attachedThreadIds, @@ -957,6 +1005,7 @@ export const layerWithOptions = ( readonly providerSessionId: ProviderSessionId; readonly reason: ProviderSessionReleaseReason; readonly detail?: string; + readonly runtimeErrorCause?: "stream_end"; readonly cancelIdleFiber?: boolean; readonly onlyIfIdleGeneration?: number; readonly gracefulSubscribers?: boolean; @@ -988,6 +1037,9 @@ export const layerWithOptions = ( const records = { entry, reason: input.reason, + ...(input.runtimeErrorCause === undefined + ? {} + : { runtimeErrorCause: input.runtimeErrorCause }), ...(input.detail === undefined ? {} : { detail: input.detail }), releasedAt, }; @@ -2399,6 +2451,7 @@ export const layerWithOptions = ( yield* releaseEntry({ providerSessionId: entry.runtime.providerSessionId, reason: "runtime_error", + ...(Exit.isSuccess(exit) ? { runtimeErrorCause: "stream_end" as const } : {}), detail: Cause.pretty(cause), }).pipe(logReleaseFailure(entry.runtime.providerSessionId)); }), diff --git a/apps/server/src/orchestration-v2/testkit/ErrorOccurrence.integration.test.ts b/apps/server/src/orchestration-v2/testkit/ErrorOccurrence.integration.test.ts new file mode 100644 index 000000000000..6df2a6459f27 --- /dev/null +++ b/apps/server/src/orchestration-v2/testkit/ErrorOccurrence.integration.test.ts @@ -0,0 +1,88 @@ +import * as NodeServices from "@effect/platform-node/NodeServices"; +import { assert, it } from "@effect/vitest"; +import * as DateTime from "effect/DateTime"; +import * as Effect from "effect/Effect"; +import * as TestClock from "effect/testing/TestClock"; + +import { ClaudeOrchestratorReplayHarness } from "../Adapters/ClaudeAdapterV2.testkit.ts"; +import * as IdAllocator from "@t3tools/provider-core/server/IdAllocator"; +import { ProjectionStoreV2, threadShellFromProjection } from "../ProjectionStore.ts"; +import { ProviderSessionManagerV2 } from "../ProviderSessionManager.ts"; +import { provideDeterministicTestRuntime } from "./DeterministicRuntime.ts"; +import { ORCHESTRATOR_REPLAY_FIXTURES } from "./fixtures/index.ts"; +import { CLAUDE_RESULT_IS_ERROR_PROMPT, materializeFixtureInput } from "./fixtures/shared.ts"; +import { runOrchestratorV2Scenario } from "./OrchestratorScenario.ts"; +import * as ProviderReplayHarness from "./ProviderReplayHarness.ts"; +import { checkpointWorkspace } from "@t3tools/provider-testing/replayWorkspace"; +import { readProviderReplayTranscript } from "@t3tools/provider-testing/replayTranscript"; + +it.effect("preserves a replayed root failure occurrence through session release", () => + Effect.gen(function* () { + const fixture = ORCHESTRATOR_REPLAY_FIXTURES.find( + (entry) => entry.name === "claude_result_is_error", + )!; + const provider = fixture.providers.find((entry) => entry.driver === "claudeAgent")!; + const raw = yield* readProviderReplayTranscript(provider.transcriptFile).pipe( + Effect.provide(NodeServices.layer), + ); + const end = raw.entries.findIndex( + (entry) => entry.type === "emit_inbound" && entry.label === "result:1_is_error", + ); + assert.isAtLeast(end, 0); + const transcript = yield* ClaudeOrchestratorReplayHarness.decodeTranscript({ + ...raw, + entries: raw.entries.slice(0, end + 1), + }); + const materialized = yield* materializeFixtureInput({ + scenario: "error-occurrence", + fixtureInput: { steps: [{ type: "message", text: CLAUDE_RESULT_IS_ERROR_PROMPT }] }, + driver: provider.driver, + modelSelection: provider.modelSelection, + }).pipe(Effect.provide(IdAllocator.layer), provideDeterministicTestRuntime); + const scenario = { + name: "error-occurrence", + transcript, + ...materialized, + runtimePolicyOverride: { + ...provider.runtimePolicyOverride, + cwd: yield* checkpointWorkspace("error-occurrence"), + }, + }; + yield* Effect.gen(function* () { + const result = yield* runOrchestratorV2Scenario(scenario); + const threadId = materialized.projectionThreadIds[0]!; + const projection = result.projections.get(threadId)!; + assert.equal(projection.runs.at(-1)?.status, "failed"); + const item = projection.turnItems.find((item) => item.type === "error"); + assert.isDefined(item); + if (item?.type !== "error") throw new Error("Missing replayed error"); + const occurrenceAt = item.completedAt ?? item.startedAt; + assert.isNotNull(occurrenceAt); + if (occurrenceAt === null) throw new Error("Missing failure occurrence"); + const expectedAt = DateTime.formatIso(occurrenceAt); + const before = threadShellFromProjection(projection); + assert.equal(before.lastError, item.failure.message); + assert.equal(before.lastErrorAt, expectedAt); + const manager = yield* ProviderSessionManagerV2; + const store = yield* ProjectionStoreV2; + const session = projection.providerSessions.at(-1)!; + assert.isDefined(session); + yield* TestClock.adjust("1 second"); + yield* manager.release({ providerSessionId: session.id, reason: "idle_timeout" }); + const after = yield* store.getThreadProjection(threadId); + assert.equal(after.providerSessions.at(-1)?.status, "stopped"); + const sqlShell = (yield* store.getShellSnapshot()).threads.find( + (thread) => thread.id === threadId, + )!; + for (const shell of [threadShellFromProjection(after), sqlShell]) { + assert.equal(shell.lastError, before.lastError); + assert.equal(shell.lastErrorAt, expectedAt); + } + }).pipe( + Effect.provide( + ProviderReplayHarness.layerProviderReplay(scenario, ClaudeOrchestratorReplayHarness), + ), + provideDeterministicTestRuntime, + ); + }).pipe(Effect.scoped), +); diff --git a/apps/server/src/orchestration-v2/testkit/ProviderReplayHarness.ts b/apps/server/src/orchestration-v2/testkit/ProviderReplayHarness.ts index 27c1cc28ae5a..b6d850645d19 100644 --- a/apps/server/src/orchestration-v2/testkit/ProviderReplayHarness.ts +++ b/apps/server/src/orchestration-v2/testkit/ProviderReplayHarness.ts @@ -242,7 +242,11 @@ export function layerProviderReplay { const layerRegistry = harness.makeProviderAdapterRegistryLayer( @@ -277,6 +281,7 @@ export function layerWithRegistry( | Orchestrator.OrchestratorV2 | EffectWorker.OrchestrationEffectWorkerV2 | EventSink.EventSinkV2 + | ProjectionStore.ProjectionStoreV2 | ProviderSessionManager.ProviderSessionManagerV2, Error | MigrationError | PlatformError.PlatformError | SqlError > { @@ -499,6 +504,7 @@ export function layerWithRegistry( layerEffectWorkerProvided, layerEventSinkProvided, layerContinuationWorkerProvided, + layerStores, ).pipe(Layer.provide(ProviderTurnStartServiceTestkit.layer), Layer.provide(NodeServices.layer)); // Build the daemon from the exact worker instance exposed alongside the diff --git a/apps/web/src/components/ChatView.tsx b/apps/web/src/components/ChatView.tsx index a4f56c75e5e3..c0ea1642bd74 100644 --- a/apps/web/src/components/ChatView.tsx +++ b/apps/web/src/components/ChatView.tsx @@ -473,13 +473,8 @@ import { ProviderStatusBanner, shouldShowProviderStatusBanner, } from "./chat/ProviderStatusBanner"; -import { - dismissThreadErrorBannerForSession, - getThreadErrorBannerKey, - isThreadErrorBannerDismissedForSession, - shouldShowThreadErrorBanner, - ThreadErrorBanner, -} from "./chat/ThreadErrorBanner"; +import { ThreadErrorBanner } from "./chat/ThreadErrorBanner"; +import { useThreadErrorState } from "./chat/useThreadErrorState"; import { QueuedRunsControl, type QueuedRunsControlHandle, @@ -561,7 +556,6 @@ import { releaseDraftAttachments, startAttachmentUpload, } from "../lib/attachmentUploadQueue"; -import { sanitizeThreadErrorMessage } from "~/rpc/transportError"; import { RightPanelSheet } from "./RightPanelSheet"; import { previewEnvironment } from "../state/preview"; import { clampFileAttachmentUploadBytes } from "@t3tools/client-runtime/state/attachments"; @@ -1533,14 +1527,6 @@ const PersistentThreadTerminalPanel = memo(function PersistentThreadTerminalPane ); }); -// Errors surface through two maps (draft-keyed and thread-keyed) whose entries -// can race around promotion, so each write carries its time to let the latest -// one win when they collide. -type LocalThreadErrorEntry = { - readonly message: string | null; - readonly at: number; -}; - function chatActionErrorMessage(error: unknown): string { return error instanceof Error ? error.message : "An error occurred."; } @@ -1902,12 +1888,6 @@ export default function ChatView(props: ChatViewProps) { ); const optimisticUserMessagesRef = useRef(optimisticUserMessages); optimisticUserMessagesRef.current = optimisticUserMessages; - const [localDraftErrorsByDraftId, setLocalDraftErrorsByDraftId] = useState< - Record - >({}); - const [localServerErrorsByThreadKey, setLocalServerErrorsByThreadKey] = useState< - Record - >({}); const [isConnecting, _setIsConnecting] = useState(false); const isRevertingCheckpoint = useComposerDraftStore((store) => store.rewindingThreadKeys.has(routeThreadKey), @@ -2055,45 +2035,6 @@ export default function ChatView(props: ChatViewProps) { ? scopeProjectRef(draftThread.environmentId, draftThread.projectId) : null; const fallbackDraftProject = useProject(fallbackDraftProjectRef); - const localDraftError = serverThread - ? null - : ((draftId ? localDraftErrorsByDraftId[draftId]?.message : null) ?? null); - const localServerError = localServerErrorsByThreadKey[routeThreadKey]?.message ?? null; - // Draft errors are keyed by draftId while server errors are keyed by thread - // key, so a pending draft entry must migrate when the server thread loads or - // a failed send would silently disappear on promotion. When both keys hold - // an entry, the most recent write wins. - useEffect(() => { - if (!serverThread || !draftId) { - return; - } - const pendingDraftEntry = localDraftErrorsByDraftId[draftId]; - if (pendingDraftEntry === undefined) { - return; - } - setLocalDraftErrorsByDraftId((existing) => { - if (existing[draftId] === undefined) { - return existing; - } - const next = { ...existing }; - delete next[draftId]; - return next; - }); - setLocalServerErrorsByThreadKey((existing) => { - const currentEntry = existing[routeThreadKey]; - if ( - currentEntry !== undefined && - (currentEntry.at > pendingDraftEntry.at || - currentEntry.message === pendingDraftEntry.message) - ) { - return existing; - } - return { - ...existing, - [routeThreadKey]: pendingDraftEntry, - }; - }); - }, [draftId, localDraftErrorsByDraftId, routeThreadKey, serverThread]); const localDraftThread = useMemo( () => draftThread @@ -2176,27 +2117,18 @@ export default function ChatView(props: ChatViewProps) { }, [parentSubagentThread?.title, parentSubagentThreadRef], ); - const threadError = isServerThread - ? (localServerError ?? serverRuntime?.lastError ?? null) - : localDraftError; - // Dismissals can only mask the shown error, never clear it: a server thread - // keeps its error in session.lastError, so clearing the local shadow would - // just fall through to the persisted one. Mask the current error until a - // different error arrives, mirroring the provider status banner. - const threadErrorBannerKey = getThreadErrorBannerKey(routeThreadKey, threadError); - const visibleThreadError = shouldShowThreadErrorBanner( - routeThreadKey, - threadError, - isThreadErrorBannerDismissedForSession(threadErrorBannerKey), - ) - ? threadError - : null; - // Dismissing only mutates the session-scoped mask set, which does not - // trigger a render on its own; setThreadError(null) can also bail when the - // local shadow is already empty and the banner is driven purely by - // session.lastError. Bump a tick so the banner hides immediately. Mirrors - // the branch mismatch banner. - const [, setThreadErrorBannerDismissTick] = useState(0); + const { + error: visibleThreadError, + currentError: threadError, + errorClass: threadErrorClass, + dismiss: dismissThreadError, + setError: setLocalThreadError, + } = useThreadErrorState({ + threadKey: routeThreadKey, + draftId, + isServerThread, + runtime: isServerThread ? (serverRuntime ?? null) : null, + }); const defaultRuntimeMode = resolveProjectSettings(settings, activeThread?.projectId ?? null) .settings.defaultRuntimeMode; // Implicit drafts follow their current project/environment, including retargets. @@ -4538,37 +4470,19 @@ export default function ChatView(props: ChatViewProps) { const setThreadError = useCallback( (targetThreadId: ThreadId | null, error: string | null) => { if (!targetThreadId) return; - const nextError = sanitizeThreadErrorMessage(error); - const nextEntry: LocalThreadErrorEntry = { message: nextError, at: Date.now() }; if ( serverThread && targetThreadId === routeThreadRef.threadId && serverThread.environmentId === routeThreadRef.environmentId && serverThread.id === targetThreadId ) { - setLocalServerErrorsByThreadKey((existing) => { - if ((existing[routeThreadKey]?.message ?? null) === nextError) { - return existing; - } - return { - ...existing, - [routeThreadKey]: nextEntry, - }; - }); + setLocalThreadError(routeThreadKey, true, error); return; } const localDraftErrorKey = draftId ?? targetThreadId; - setLocalDraftErrorsByDraftId((existing) => { - if ((existing[localDraftErrorKey]?.message ?? null) === nextError) { - return existing; - } - return { - ...existing, - [localDraftErrorKey]: nextEntry, - }; - }); + setLocalThreadError(localDraftErrorKey, false, error); }, - [draftId, routeThreadKey, routeThreadRef, serverThread], + [draftId, routeThreadKey, routeThreadRef, serverThread, setLocalThreadError], ); const focusComposer = useCallback(() => { @@ -11370,16 +11284,8 @@ export default function ChatView(props: ChatViewProps) { /> { - setThreadError(activeThread.id, null); - dismissThreadErrorBannerForSession(threadErrorBannerKey); - setThreadErrorBannerDismissTick((tick) => tick + 1); - }} + errorClass={threadErrorClass} + onDismiss={dismissThreadError} /> {/* Messages Wrapper */} diff --git a/apps/web/src/components/chat/ThreadErrorBanner.test.tsx b/apps/web/src/components/chat/ThreadErrorBanner.test.tsx index a3dbb27515ae..3677c9ece156 100644 --- a/apps/web/src/components/chat/ThreadErrorBanner.test.tsx +++ b/apps/web/src/components/chat/ThreadErrorBanner.test.tsx @@ -6,6 +6,7 @@ import { getThreadErrorBannerKey, isThreadErrorBannerDismissedForSession, shouldShowThreadErrorBanner, + resolveThreadErrorBanner, ThreadErrorBanner, } from "./ThreadErrorBanner"; @@ -66,6 +67,65 @@ describe("ThreadErrorBanner", () => { ).toBe(false); }); + it("distinguishes same-text occurrences and local errors from hidden runtime failures", () => { + const thread = "env:occurrence"; + const first = getThreadErrorBannerKey(thread, "Failed", "2026-09-20T01:00:00Z"); + dismissThreadErrorBannerForSession(first); + expect( + isThreadErrorBannerDismissedForSession( + getThreadErrorBannerKey(thread, "Failed", "2026-09-20T01:00:00Z"), + ), + ).toBe(true); + expect( + isThreadErrorBannerDismissedForSession( + getThreadErrorBannerKey(thread, "Failed", "2026-09-20T02:00:00Z"), + ), + ).toBe(false); + const local = getThreadErrorBannerKey(thread, "Failed", 12, "local"); + dismissThreadErrorBannerForSession(local); + expect( + isThreadErrorBannerDismissedForSession( + getThreadErrorBannerKey(thread, "Failed", 13, "local"), + ), + ).toBe(false); + expect( + isThreadErrorBannerDismissedForSession( + getThreadErrorBannerKey(thread, "Failed", 12, "runtime"), + ), + ).toBe(false); + }); + + it("keeps the hidden runtime occurrence masked but shows a later provider occurrence", () => { + const input = { + threadKey: "env:precedence", + localError: "Failed", + localErrorAt: 1, + runtime: { lastError: "Failed", lastErrorAt: "2026-09-20T01:00:00Z" }, + }; + const local = resolveThreadErrorBanner(input); + dismissThreadErrorBannerForSession( + local.key, + getThreadErrorBannerKey(input.threadKey, input.runtime.lastError, input.runtime.lastErrorAt), + ); + expect(resolveThreadErrorBanner(input)).toEqual(local); + const laterRuntime = resolveThreadErrorBanner({ + ...input, + runtime: { lastError: "Provider failed", lastErrorAt: "2026-09-20T02:00:00Z" }, + }); + expect(laterRuntime.error).toBe("Provider failed"); + expect(isThreadErrorBannerDismissedForSession(laterRuntime.key)).toBe(false); + const laterLocal = resolveThreadErrorBanner({ ...input, localErrorAt: 2 }); + expect(isThreadErrorBannerDismissedForSession(laterLocal.key)).toBe(false); + const runtime = resolveThreadErrorBanner({ ...input, localError: null }); + expect(isThreadErrorBannerDismissedForSession(runtime.key)).toBe(false); + const unknownAt = resolveThreadErrorBanner({ + ...input, + localError: null, + runtime: { lastError: "Failed", lastErrorAt: null }, + }); + expect(unknownAt.key).toBe(getThreadErrorBannerKey(input.threadKey, "Failed")); + }); + it("never shows a null error", () => { expect(shouldShowThreadErrorBanner("env:thread-e", null, false)).toBe(false); }); diff --git a/apps/web/src/components/chat/ThreadErrorBanner.tsx b/apps/web/src/components/chat/ThreadErrorBanner.tsx index 555a76972db4..1f98c86ecd9b 100644 --- a/apps/web/src/components/chat/ThreadErrorBanner.tsx +++ b/apps/web/src/components/chat/ThreadErrorBanner.tsx @@ -7,8 +7,51 @@ import { Tooltip, TooltipPopup, TooltipTrigger } from "../ui/tooltip"; import { OpenAI } from "../Icons"; import { ChatGptUsageButton } from "../settings/ChatGptUsageButton"; -export function getThreadErrorBannerKey(threadKey: string, error: string | null): string | null { - return error === null ? null : `${threadKey}\u0000${error}`; +export function getThreadErrorBannerKey( + threadKey: string, + error: string | null, + occurrence: string | number | null = null, + source: "local" | "runtime" = "runtime", +): string | null { + return error === null + ? null + : `${threadKey}\u0000${source}\u0000${occurrence ?? ""}\u0000${error}`; +} + +export function resolveThreadErrorBanner(input: { + threadKey: string; + localError: string | null; + localErrorAt: number | null; + runtime: { readonly lastError: string | null; readonly lastErrorAt?: string | null } | null; +}) { + const localKey = getThreadErrorBannerKey( + input.threadKey, + input.localError, + input.localErrorAt, + "local", + ); + const runtimeKey = getThreadErrorBannerKey( + input.threadKey, + input.runtime?.lastError ?? null, + input.runtime?.lastErrorAt ?? null, + ); + const isLocal = + input.localError !== null && + !( + localKey !== null && + sessionDismissedThreadErrorBannerKeys.has(localKey) && + sessionDismissedThreadErrorBannerKeys.get(localKey) !== runtimeKey + ); + const error = isLocal ? input.localError : (input.runtime?.lastError ?? null); + return { + error, + key: getThreadErrorBannerKey( + input.threadKey, + error, + isLocal ? input.localErrorAt : (input.runtime?.lastErrorAt ?? null), + isLocal ? "local" : "runtime", + ), + }; } export function shouldShowThreadErrorBanner( @@ -21,14 +64,16 @@ export function shouldShowThreadErrorBanner( // Session-scoped (module-level so it survives ChatView remounts, e.g. route // changes between threads). Mirrors the branch-mismatch banner: a dismissal -// is remembered per thread key plus message, so navigating away to a thread -// with no error cannot resurrect the banner, while a different error message -// on the same thread still appears. -const sessionDismissedThreadErrorBannerKeys = new Set(); +// is remembered per thread and visible error occurrence. Navigation cannot +// resurrect that occurrence, but a later failure with the same message appears. +const sessionDismissedThreadErrorBannerKeys = new Map(); -export function dismissThreadErrorBannerForSession(bannerKey: string | null): void { +export function dismissThreadErrorBannerForSession( + bannerKey: string | null, + runtimeKey: string | null = null, +): void { if (bannerKey !== null) { - sessionDismissedThreadErrorBannerKeys.add(bannerKey); + sessionDismissedThreadErrorBannerKeys.set(bannerKey, runtimeKey); } } diff --git a/apps/web/src/components/chat/useThreadErrorState.test.tsx b/apps/web/src/components/chat/useThreadErrorState.test.tsx new file mode 100644 index 000000000000..19157334e30e --- /dev/null +++ b/apps/web/src/components/chat/useThreadErrorState.test.tsx @@ -0,0 +1,96 @@ +import { act, useLayoutEffect } from "react"; +import { create, type ReactTestRenderer } from "react-test-renderer"; +import { afterEach, beforeEach, expect, it, vi } from "vite-plus/test"; +import { useThreadErrorState } from "./useThreadErrorState"; + +let renderer: ReactTestRenderer; +let value: ReturnType; +let input: Parameters[0]; +let serial = 0; +function Probe() { + const state = useThreadErrorState(input); + useLayoutEffect(() => { + value = state; + }); + return
{state.error}
; +} +function render() { + act(() => renderer.update()); +} +beforeEach(() => { + vi.stubGlobal("IS_REACT_ACT_ENVIRONMENT", true); + vi.spyOn(Date, "now").mockReturnValue(100); + input = { + threadKey: `env:hook-${++serial}`, + draftId: "draft", + isServerThread: true, + runtime: { + lastError: "Durable failure", + lastErrorAt: "2026-09-25T00:00:00Z", + lastErrorClass: "usage_limit", + }, + }; + act(() => { + renderer = create(); + }); +}); +afterEach(() => { + act(() => renderer.unmount()); + vi.restoreAllMocks(); + vi.unstubAllGlobals(); +}); +it("dismisses a local failure without revealing the unchanged durable failure", () => { + act(() => value.setError(input.threadKey, true, "Save failed")); + expect(value.error).toBe("Save failed"); + act(() => value.dismiss()); + expect(value.error).toBeNull(); + render(); + expect(value.error).toBeNull(); + input.runtime = { ...input.runtime!, lastErrorAt: "2026-09-25T00:01:00Z" }; + render(); + expect(value.error).toBe("Durable failure"); + expect(value.errorClass).toBe("usage_limit"); +}); +it("reappears for an identical local write even in the same clock tick", () => { + act(() => value.setError(input.threadKey, true, "Save failed")); + act(() => value.dismiss()); + expect(value.error).toBeNull(); + act(() => value.setError(input.threadKey, true, "Save failed")); + expect(value.error).toBe("Save failed"); + expect(value.errorClass).toBeNull(); +}); +it("keeps a newer undismissed local failure visible", () => { + act(() => value.setError(input.threadKey, true, "Save failed")); + act(() => value.dismiss()); + input.runtime = { lastError: "Provider failure", lastErrorAt: "2026-09-25T00:01:00Z" }; + render(); + expect(value.error).toBe("Provider failure"); + act(() => value.setError(input.threadKey, true, "Save failed")); + expect(value.error).toBe("Save failed"); +}); +it("dismisses runtime errors immediately and remembers them across remounts", () => { + act(() => value.dismiss()); + expect(value.error).toBeNull(); + act(() => { + renderer.unmount(); + renderer = create(); + }); + expect(value.error).toBeNull(); + input.runtime = { lastError: "Durable failure", lastErrorAt: "2026-09-25T00:01:00Z" }; + render(); + expect(value.error).toBe("Durable failure"); +}); +it("preserves repeated draft occurrences and migrates the latest write on promotion", () => { + input.isServerThread = false; + input.runtime = null; + render(); + act(() => value.setError("draft", false, "Send failed")); + act(() => value.dismiss()); + act(() => value.setError("draft", false, "Send failed")); + expect(value.error).toBe("Send failed"); + input.isServerThread = true; + render(); + expect(value.error).toBe("Send failed"); + act(() => value.setError(input.threadKey, true, null)); + expect(value.error).toBeNull(); +}); diff --git a/apps/web/src/components/chat/useThreadErrorState.ts b/apps/web/src/components/chat/useThreadErrorState.ts new file mode 100644 index 000000000000..bba77b72a351 --- /dev/null +++ b/apps/web/src/components/chat/useThreadErrorState.ts @@ -0,0 +1,82 @@ +import type { OrchestrationV2ProviderFailureClass } from "@t3tools/contracts"; +import { useCallback, useEffect, useState } from "react"; +import { sanitizeThreadErrorMessage } from "~/rpc/transportError"; +import { + dismissThreadErrorBannerForSession, + getThreadErrorBannerKey, + isThreadErrorBannerDismissedForSession, + resolveThreadErrorBanner, + shouldShowThreadErrorBanner, +} from "./ThreadErrorBanner"; + +type LocalErrorEntry = { message: string | null; at: number }; + +export function useThreadErrorState(input: { + threadKey: string; + draftId: string | null; + isServerThread: boolean; + runtime: { + readonly lastError: string | null; + readonly lastErrorAt?: string | null; + readonly lastErrorClass?: OrchestrationV2ProviderFailureClass | null; + } | null; +}) { + const { threadKey, draftId, isServerThread, runtime } = input; + const [serverErrors, setServerErrors] = useState>({}); + const [draftErrors, setDraftErrors] = useState>({}); + const [, setDismissTick] = useState(0); + useEffect(() => { + if (!isServerThread || !draftId) return; + const pending = draftErrors[draftId]; + if (pending === undefined) return; + setDraftErrors((existing) => { + const next = { ...existing }; + delete next[draftId]; + return next; + }); + setServerErrors((existing) => { + const current = existing[threadKey]; + if (current !== undefined && current.at > pending.at) return existing; + return { ...existing, [threadKey]: pending }; + }); + }, [draftErrors, draftId, isServerThread, threadKey]); + + const setError = useCallback((key: string, isServer: boolean, error: string | null) => { + const message = sanitizeThreadErrorMessage(error); + const at = Date.now(); + const update = isServer ? setServerErrors : setDraftErrors; + update((existing) => { + if (message === null && existing[key]?.message == null) return existing; + return { ...existing, [key]: { message, at: Math.max(at, (existing[key]?.at ?? 0) + 1) } }; + }); + }, []); + const local = isServerThread ? serverErrors[threadKey] : draftErrors[draftId ?? ""]; + const resolved = resolveThreadErrorBanner({ + threadKey, + localError: local?.message ?? null, + localErrorAt: local?.at ?? null, + runtime, + }); + const runtimeKey = getThreadErrorBannerKey( + threadKey, + runtime?.lastError ?? null, + runtime?.lastErrorAt ?? null, + ); + const dismiss = () => { + dismissThreadErrorBannerForSession(resolved.key, runtimeKey); + setDismissTick((tick) => tick + 1); + }; + return { + setError, + dismiss, + currentError: resolved.error, + error: shouldShowThreadErrorBanner( + threadKey, + resolved.error, + isThreadErrorBannerDismissedForSession(resolved.key), + ) + ? resolved.error + : null, + errorClass: resolved.key === runtimeKey ? (runtime?.lastErrorClass ?? null) : null, + }; +} diff --git a/packages/client-runtime/src/state/models.ts b/packages/client-runtime/src/state/models.ts index e9049239e7e7..842b12ec8c6d 100644 --- a/packages/client-runtime/src/state/models.ts +++ b/packages/client-runtime/src/state/models.ts @@ -56,6 +56,7 @@ export interface ThreadRuntimeSummary { readonly providerInstanceId: ProviderInstanceId; readonly providerName: string | null; readonly lastError: string | null; + readonly lastErrorAt?: string | null; readonly lastErrorClass?: OrchestrationV2ProviderFailureClass | null; readonly usageLimitResetAt?: string | null; readonly updatedAt: string; @@ -192,6 +193,7 @@ function shellRuntime(thread: OrchestrationV2ThreadShell): ThreadRuntimeSummary providerInstanceId: thread.providerInstanceId, providerName: null, lastError: thread.lastError ?? null, + lastErrorAt: thread.lastErrorAt ?? null, lastErrorClass: thread.lastErrorClass ?? null, usageLimitResetAt: thread.usageLimitResetAt ?? null, updatedAt: iso(thread.updatedAt), diff --git a/packages/client-runtime/src/state/threadExecution.test.ts b/packages/client-runtime/src/state/threadExecution.test.ts index ba426b8195fe..ee19c5460fbd 100644 --- a/packages/client-runtime/src/state/threadExecution.test.ts +++ b/packages/client-runtime/src/state/threadExecution.test.ts @@ -10,6 +10,7 @@ import { ThreadId, type OrchestrationV2ExecutionNode, type OrchestrationV2RunStatus, + type OrchestrationV2ProviderCapabilities, } from "@t3tools/contracts"; import * as DateTime from "effect/DateTime"; import { usageLimitBlockedRun } from "@t3tools/shared/orchestrationV2ThreadError"; @@ -83,6 +84,55 @@ describe("thread execution presentation", () => { expect(deriveThreadRuntime(projection)).toMatchObject({ lastError: "Plan limit reached", lastErrorClass: "usage_limit", + lastErrorAt: DateTime.formatIso(now), + }); + const later = DateTime.makeUnsafe("2026-09-20T02:00:00Z"); + const session = { + id: ProviderSessionId.make("session:error"), + driver: ProviderDriverKind.make("pi"), + providerInstanceId: projection.thread.providerInstanceId, + status: "error" as const, + cwd: "/workspace", + model: null, + capabilities: {} as OrchestrationV2ProviderCapabilities, + createdAt: now, + updatedAt: later, + lastError: "Separate session failure", + lastErrorAt: later, + }; + expect(deriveThreadRuntime({ ...projection, providerSessions: [session] })).toMatchObject({ + lastError: session.lastError, + lastErrorClass: null, + lastErrorAt: DateTime.formatIso(later), + }); + expect( + deriveThreadRuntime({ ...projection, providerSessions: [{ ...session, lastErrorAt: null }] }), + ).toMatchObject({ + lastError: session.lastError, + lastErrorAt: null, + }); + expect( + deriveThreadRuntime({ + ...projection, + providerSessions: [{ ...session, lastError: item.failure.message }], + }), + ).toMatchObject({ + lastErrorClass: "usage_limit", + lastErrorAt: DateTime.formatIso(later), + }); + expect( + deriveThreadRuntime({ ...projection, providerSessions: [{ ...session, lastError: null }] }), + ).toMatchObject({ + lastErrorClass: "usage_limit", + lastErrorAt: DateTime.formatIso(now), + }); + expect( + deriveThreadRuntime({ + ...projection, + turnItems: [{ ...item, startedAt: null, completedAt: null, updatedAt: later }], + }), + ).toMatchObject({ + lastErrorAt: null, }); expect( deriveThreadRuntime({ diff --git a/packages/client-runtime/src/state/threadExecution.ts b/packages/client-runtime/src/state/threadExecution.ts index b1e2c2731228..89775414f64a 100644 --- a/packages/client-runtime/src/state/threadExecution.ts +++ b/packages/client-runtime/src/state/threadExecution.ts @@ -1,6 +1,7 @@ import { - latestRootProviderFailure, + latestRootProviderFailureItem, latestUnheldRun, + providerFailureOccurredAt, threadErrorSummary, usageLimitRunPresentedAsLatest, } from "@t3tools/shared/orchestrationV2ThreadError"; @@ -256,6 +257,7 @@ export function deriveThreadRuntime( } const activeRunId = latestMatchingRun(projection, (run) => INTERRUPTIBLE_RUN_STATUSES.has(run.status))?.id ?? null; + const failureItem = latestRootProviderFailureItem(latestRunProjection, projection.turnItems); return { status: usageLimitedRun ? "failed" @@ -269,10 +271,13 @@ export function deriveThreadRuntime( : DateTime.formatIso(orchestrationV2RunWorkStartedAt(liveActivityRun)), providerInstanceId: projection.thread.providerInstanceId, providerName: providerSession?.driver ?? null, - ...threadErrorSummary( - latestRootProviderFailure(latestRunProjection, projection.turnItems), - providerSession?.lastError ?? null, - ), + ...threadErrorSummary(failureItem?.failure ?? null, providerSession?.lastError ?? null, { + sessionErrorAt: + providerSession?.lastErrorAt == null + ? null + : DateTime.formatIso(providerSession.lastErrorAt), + failureAt: providerFailureOccurredAt(failureItem), + }), updatedAt: DateTime.formatIso(projection.updatedAt), }; } diff --git a/packages/contracts/src/orchestrationV2.ts b/packages/contracts/src/orchestrationV2.ts index 84843c1f2452..d3833b5d7609 100644 --- a/packages/contracts/src/orchestrationV2.ts +++ b/packages/contracts/src/orchestrationV2.ts @@ -736,6 +736,7 @@ export const OrchestrationV2ProviderSession = Schema.Struct({ createdAt: Schema.DateTimeUtc, updatedAt: Schema.DateTimeUtc, lastError: Schema.NullOr(Schema.String), + lastErrorAt: Schema.optional(Schema.NullOr(Schema.DateTimeUtc)), }); export type OrchestrationV2ProviderSession = typeof OrchestrationV2ProviderSession.Type; @@ -1878,6 +1879,7 @@ export const OrchestrationV2ThreadShell = Schema.Struct({ ), status: OrchestrationV2ShellThreadStatus, lastError: Schema.optional(Schema.NullOr(Schema.String)), + lastErrorAt: Schema.optional(Schema.NullOr(IsoDateTime)), lastErrorClass: Schema.optional(Schema.NullOr(OrchestrationV2ProviderFailureClass)), usageLimitResetAt: Schema.optional(Schema.NullOr(IsoDateTime)), pendingRuntimeRequest: Schema.NullOr(OrchestrationV2PendingRuntimeRequestSummary), @@ -2073,6 +2075,7 @@ export const OrchestrationV2ProviderSessionJson = OrchestrationV2ProviderSession ...fields, createdAt: Schema.DateTimeUtcFromString, updatedAt: Schema.DateTimeUtcFromString, + lastErrorAt: Schema.optional(Schema.NullOr(Schema.DateTimeUtcFromString)), }), ); export type OrchestrationV2ProviderSessionJson = typeof OrchestrationV2ProviderSessionJson.Type; diff --git a/packages/provider-pi/src/server/adapter.test.ts b/packages/provider-pi/src/server/adapter.test.ts index e13f8213077f..c283fa7280a4 100644 --- a/packages/provider-pi/src/server/adapter.test.ts +++ b/packages/provider-pi/src/server/adapter.test.ts @@ -43,9 +43,11 @@ import { makePiAdapterV2, PiAdapterV2Driver, PI_PROVIDER, + piLastErrorAt, type PiAdapterV2Options, } from "./adapter.ts"; import { makePiRpcConnection, type PiRpcRecord } from "./rpc.ts"; +import * as Deferred from "effect/Deferred"; const layerTest = Layer.mergeAll( NodeServices.layer, @@ -366,8 +368,10 @@ const openRuntime = Effect.fnUntraced(function* ( runtimePolicy, }); const emitted = yield* Queue.unbounded(); + const eventsEnded = yield* Deferred.make(); yield* runtime.events.pipe( Stream.runForEach((event) => Queue.offer(emitted, event)), + Effect.ensuring(Deferred.succeed(eventsEnded, undefined)), Effect.forkScoped, ); const takeEvent = (predicate: (event: ProviderAdapter.ProviderAdapterV2Event) => boolean) => @@ -377,7 +381,7 @@ const openRuntime = Effect.fnUntraced(function* ( if (predicate(event)) return event; } }); - return { runtime, takeEvent }; + return { runtime, takeEvent, eventsEnded: Deferred.await(eventsEnded) }; }); const makeAppThread = Effect.fnUntraced(function* (model: string, threadId = THREAD_ID) { @@ -470,7 +474,10 @@ const expectModelFailure = (errorMessage: string) => ); assert.isTrue( sessionError.type === "provider_session.updated" && - sessionError.providerSession.lastError === errorMessage, + sessionError.providerSession.lastError === errorMessage && + sessionError.providerSession.lastErrorAt != null && + DateTime.toEpochMillis(sessionError.providerSession.lastErrorAt) === + DateTime.toEpochMillis(sessionError.providerSession.updatedAt), ); const terminal = yield* takeEvent((event) => event.type === "turn.terminal"); assert.isTrue( @@ -481,6 +488,118 @@ const expectModelFailure = (errorMessage: string) => }).pipe(Effect.scoped, Effect.provide(layerTest)); describe("PiAdapterV2", () => { + it("keeps the occurrence identity on same-text refreshes, including legacy null", () => { + const now = DateTime.makeUnsafe("2026-09-20T12:00:00Z"); + for (const previousErrorAt of [null, DateTime.makeUnsafe("2026-09-19T12:00:00Z")]) { + assert.strictEqual( + piLastErrorAt({ + previousError: "capacity exhausted", + previousErrorAt, + nextError: "capacity exhausted", + now, + }), + previousErrorAt, + ); + } + }); + + it("clears the occurrence and stamps changed or newly set errors", () => { + const now = DateTime.makeUnsafe("2026-09-20T12:00:00Z"); + const previousErrorAt = DateTime.makeUnsafe("2026-09-19T12:00:00Z"); + assert.isNull( + piLastErrorAt({ previousError: "capacity exhausted", previousErrorAt, nextError: null, now }), + ); + for (const previousError of [null, "different failure"]) { + assert.strictEqual( + piLastErrorAt({ + previousError, + previousErrorAt: null, + nextError: "capacity exhausted", + now, + }), + now, + ); + } + }); + + it.effect("emits a new occurrence for the same failure after the next turn clears it", () => + Effect.gen(function* () { + const fake = yield* makeFakePi; + const { runtime, takeEvent } = yield* openRuntime(fake); + const providerThread = yield* runtime.ensureThread({ + threadId: THREAD_ID, + modelSelection: modelSelection("default"), + runtimePolicy, + }); + const occurrences: Array = []; + for (const runOrdinal of [1, 2]) { + yield* startTurn(runtime, providerThread, "default", [], "Hello pi", undefined, runOrdinal); + yield* fake.takeRequest("prompt"); + const running = yield* takeEvent( + (event) => + event.type === "provider_session.updated" && event.providerSession.status === "running", + ); + assert.equal(running.type, "provider_session.updated"); + if (running.type !== "provider_session.updated") return; + assert.isNull(running.providerSession.lastError); + assert.isNull(running.providerSession.lastErrorAt); + + yield* TestClock.adjust("1 second"); + const failedAt = yield* DateTime.now; + yield* fake.emit({ type: "agent_start" }); + yield* fake.emit({ + type: "message_end", + message: { + role: "assistant", + content: [], + stopReason: "error", + errorMessage: "capacity exhausted", + }, + }); + yield* fake.emit({ type: "agent_settled" }); + const failed = yield* takeEvent( + (event) => + event.type === "provider_session.updated" && event.providerSession.status === "error", + ); + assert.equal(failed.type, "provider_session.updated"); + if (failed.type !== "provider_session.updated") return; + assert.equal(failed.providerSession.lastError, "capacity exhausted"); + assert.isNotNull(failed.providerSession.lastErrorAt); + assert.isDefined(failed.providerSession.lastErrorAt); + const occurrence = DateTime.toEpochMillis(failed.providerSession.lastErrorAt!); + assert.equal(occurrence, DateTime.toEpochMillis(failedAt)); + occurrences.push(occurrence); + yield* takeEvent((event) => event.type === "turn.terminal"); + } + assert.isAbove(occurrences[1]!, occurrences[0]!); + }).pipe(Effect.scoped, Effect.provide(layerTest)), + ); + + it.effect("retains the occurrence when transport cleanup repeats an unsolicited-work error", () => + Effect.gen(function* () { + const fake = yield* makeFakePi; + const { runtime, takeEvent, eventsEnded } = yield* openRuntime(fake); + yield* runtime.ensureThread({ + threadId: THREAD_ID, + modelSelection: modelSelection("default"), + runtimePolicy, + }); + yield* fake.emit({ type: "agent_start" }); + const first = yield* takeEvent((event) => event.type === "provider_session.updated"); + assert.equal(first.type, "provider_session.updated"); + if (first.type !== "provider_session.updated") return; + assert.equal(first.providerSession.status, "error"); + assert.isNotNull(first.providerSession.lastErrorAt); + assert.isDefined(first.providerSession.lastErrorAt); + yield* eventsEnded; + const refreshed = yield* takeEvent((event) => event.type === "provider_session.updated"); + assert.equal(refreshed.type, "provider_session.updated"); + if (refreshed.type !== "provider_session.updated") return; + assert.equal(refreshed.providerSession.lastError, first.providerSession.lastError); + assert.deepEqual(refreshed.providerSession.lastErrorAt, first.providerSession.lastErrorAt); + }).pipe(Effect.scoped, Effect.provide(layerTest)), + ); + it.effect.each([false, true])( "serializes rollback fork hooks with native wake ownership, wake=%s", (nativeWake) => diff --git a/packages/provider-pi/src/server/adapter.ts b/packages/provider-pi/src/server/adapter.ts index c650a444daf9..1baf3a80b19c 100644 --- a/packages/provider-pi/src/server/adapter.ts +++ b/packages/provider-pi/src/server/adapter.ts @@ -95,6 +95,18 @@ import { import { PI_FILE_CHANGE_TOOLS } from "./mcpExtensionSource.ts"; export const PI_PROVIDER = ProviderDriverKind.make("pi"); +export function piLastErrorAt(input: { + readonly previousError: string | null; + readonly previousErrorAt: DateTime.Utc | null; + readonly nextError: string | null; + readonly now: DateTime.Utc; +}): DateTime.Utc | null { + if (input.nextError === null) return null; + // A status refresh is not a new occurrence, even for untimestamped legacy errors. + if (input.nextError === input.previousError) return input.previousErrorAt; + return input.now; +} + const PI_DRIVER_KIND = PI_PROVIDER; const PI_DEFAULT_INSTANCE_ID = defaultInstanceIdForDriver(PI_DRIVER_KIND); const DEFAULT_PI_SETTINGS = Schema.decodeSync(PiSettings)({}); @@ -560,7 +572,13 @@ export const makePiAdapterV2 = Effect.fn("makePiAdapterV2")(function* ( ) => Effect.gen(function* () { const updatedAt = yield* DateTime.now; - sessionEntity = { ...sessionEntity, status, lastError, updatedAt }; + const lastErrorAt = piLastErrorAt({ + previousError: sessionEntity.lastError, + previousErrorAt: sessionEntity.lastErrorAt ?? null, + nextError: lastError, + now: updatedAt, + }); + sessionEntity = { ...sessionEntity, status, lastError, lastErrorAt, updatedAt }; yield* emit({ type: "provider_session.updated", driver: PI_PROVIDER, diff --git a/packages/shared/src/orchestrationV2ThreadError.ts b/packages/shared/src/orchestrationV2ThreadError.ts index ca8dcceaa912..ce3802412972 100644 --- a/packages/shared/src/orchestrationV2ThreadError.ts +++ b/packages/shared/src/orchestrationV2ThreadError.ts @@ -10,6 +10,13 @@ export function latestRootProviderFailure( run: OrchestrationV2Run | null, turnItems: ReadonlyArray, ): OrchestrationV2ProviderFailure | null { + return latestRootProviderFailureItem(run, turnItems)?.failure ?? null; +} + +export function latestRootProviderFailureItem( + run: OrchestrationV2Run | null, + turnItems: ReadonlyArray, +): Extract | null { if (run?.status !== "failed") return null; let latest: Extract | null = null; for (const item of turnItems) { @@ -29,13 +36,21 @@ export function latestRootProviderFailure( latest = item; } } - return latest?.failure ?? null; + return latest; +} + +export function providerFailureOccurredAt( + item: Extract | null, +): string | null { + const at = item?.completedAt ?? item?.startedAt; + return at == null ? null : DateTime.formatIso(at); } /** A distinct session failure supersedes the turn's classification. */ export function threadErrorSummary( failure: OrchestrationV2ProviderFailure | null, sessionError: string | null, + occurrence?: { readonly sessionErrorAt?: string | null; readonly failureAt?: string | null }, ) { const currentFailure = sessionError !== null && sessionError !== failure?.message ? null : failure; @@ -43,6 +58,13 @@ export function threadErrorSummary( usageLimitResetAt: currentFailure?.class === "usage_limit" ? (currentFailure.resetAt ?? null) : null, lastError: sessionError ?? failure?.message ?? null, + // Identity follows the source that supplied the text, not an unrelated refresh. + lastErrorAt: + sessionError !== null + ? (occurrence?.sessionErrorAt ?? null) + : failure !== null + ? (occurrence?.failureAt ?? null) + : null, lastErrorClass: sessionError !== null && sessionError !== failure?.message ? null : (failure?.class ?? null), };