diff --git a/knip.jsonc b/knip.jsonc index babc22932ca0d..31a3b58e5abfb 100644 --- a/knip.jsonc +++ b/knip.jsonc @@ -23,7 +23,7 @@ // electron-builder loads the macOS signer by path. "entry": [ // These fixture scripts are injected into a browser page by path. - "fixtures/issue-16534/*.js", + "fixtures/*/*.js", "knip-schemas.ts", "mobile-native-client.ts", "setup-worktree.ts", diff --git a/packages/client-runtime/src/state/threadCommands.test.ts b/packages/client-runtime/src/state/threadCommands.test.ts index cdc7bcb57a947..7b7ad59b751d2 100644 --- a/packages/client-runtime/src/state/threadCommands.test.ts +++ b/packages/client-runtime/src/state/threadCommands.test.ts @@ -3,6 +3,7 @@ import { CommandId, RuntimeRequestId, EnvironmentId, + RunId, ORCHESTRATION_V2_WS_METHODS, ProjectId, ProviderInstanceId, @@ -14,9 +15,12 @@ import * as DateTime from "effect/DateTime"; import * as Crypto from "effect/Crypto"; import * as Deferred from "effect/Deferred"; import * as Effect from "effect/Effect"; +import * as Exit from "effect/Exit"; +import * as Fiber from "effect/Fiber"; import * as Layer from "effect/Layer"; import * as Option from "effect/Option"; import * as Queue from "effect/Queue"; +import * as Scope from "effect/Scope"; import * as SubscriptionRef from "effect/SubscriptionRef"; import { Atom, AtomRegistry } from "effect/reactivity"; @@ -24,6 +28,7 @@ import * as EnvironmentRegistry from "../connection/registry.ts"; import * as EnvironmentSupervisor from "../connection/supervisor.ts"; import type { RpcSession } from "../rpc/session.ts"; import { createThreadEnvironmentAtoms } from "./threadCommands.ts"; +import * as ThreadHistoryController from "./threadHistoryController.ts"; const ENVIRONMENT_ID = EnvironmentId.make("remote"); const THREAD_ID = ThreadId.make("thread"); @@ -80,11 +85,14 @@ const SNAPSHOT: OrchestrationV2ShellSnapshot = { ], }; -const makeHarness = Effect.fn("TestThreadCommands.makeHarness")(function* () { +const makeHarness = Effect.fn("TestThreadCommands.makeHarness")(function* (options?: { + readonly reconcilePendingWork?: (threadId: ThreadId) => Effect.Effect; +}) { const requests = yield* Queue.unbounded<{ command: OrchestrationV2Command; reply: Deferred.Deferred<{ sequence: number }, Error>; }>(); + const reconciled = yield* Queue.unbounded(); const supervisor = EnvironmentSupervisor.EnvironmentSupervisor.of({ target: { environmentId: ENVIRONMENT_ID }, session: yield* SubscriptionRef.make( @@ -113,6 +121,16 @@ const makeHarness = Effect.fn("TestThreadCommands.makeHarness")(function* () { digest: (_algorithm, data) => Effect.succeed(data), }), ), + Layer.succeed( + ThreadHistoryController.ThreadHistoryController, + ThreadHistoryController.ThreadHistoryController.of({ + register: () => Effect.die("unused"), + unregister: () => Effect.void, + loadEarlier: () => Effect.die("unused"), + reconcilePendingWork: (_environmentId, threadId) => + options?.reconcilePendingWork?.(threadId) ?? Queue.offer(reconciled, threadId), + }), + ), ), ); const snapshotAtom = Atom.family((_environmentId: EnvironmentId) => Atom.make(SNAPSHOT)); @@ -121,7 +139,63 @@ const makeHarness = Effect.fn("TestThreadCommands.makeHarness")(function* () { yield* Effect.addFinalizer(() => Effect.sync(() => registry.dispose())); const visibleAtom = commands.snapshotAtom(ENVIRONMENT_ID); registry.mount(visibleAtom); - return { registry, commands, snapshotAtom, visibleAtom, requests }; + return { registry, commands, snapshotAtom, visibleAtom, requests, reconciled }; +}); + +describe("interrupt turn", () => { + // A stale cached item has nothing left to interrupt, so the server can reject Stop (#18188). + it.effect("reconciles pending work when the server rejects the interrupt", () => + Effect.gen(function* () { + const h = yield* makeHarness(); + const result = h.commands.interruptTurn.run(h.registry, { + environmentId: ENVIRONMENT_ID, + input: { threadId: THREAD_ID, runId: RunId.make("finished-run") }, + }); + const request = yield* Queue.take(h.requests); + expect(request.command.type).toBe("run.interrupt"); + yield* Deferred.fail(request.reply, new Error("Run is not interruptible")); + expect((yield* Effect.promise(() => result))._tag).toBe("Failure"); + expect(yield* Queue.take(h.reconciled)).toBe(THREAD_ID); + }), + ); + + // Mirrors the thread state's handler: the lookup runs in the thread's scope and is joined. + const stopWithBlockedLookup = Effect.fn("stopWithBlockedLookup")(function* () { + const threadScope = yield* Scope.make(); + const asked = yield* Deferred.make(); + const h = yield* makeHarness({ + reconcilePendingWork: () => + Deferred.succeed(asked, undefined).pipe( + Effect.andThen(Effect.never), + Effect.forkIn(threadScope), + Effect.flatMap(Fiber.join), + ), + }); + const result = h.commands.interruptTurn.run(h.registry, { + environmentId: ENVIRONMENT_ID, + input: { threadId: THREAD_ID, runId: RunId.make("running-run") }, + }); + const request = yield* Queue.take(h.requests); + yield* Deferred.succeed(request.reply, { sequence: 1 }); + yield* Deferred.await(asked); + return { threadScope, result }; + }); + + it.effect("resolves Stop without waiting for the pending work lookup", () => + Effect.gen(function* () { + const { threadScope, result } = yield* stopWithBlockedLookup(); + expect((yield* Effect.promise(() => result))._tag).toBe("Success"); + yield* Scope.close(threadScope, Exit.void); + }), + ); + + it.effect("keeps an accepted Stop when the thread closes during the lookup", () => + Effect.gen(function* () { + const { threadScope, result } = yield* stopWithBlockedLookup(); + yield* Scope.close(threadScope, Exit.void); + expect((yield* Effect.promise(() => result))._tag).toBe("Success"); + }), + ); }); describe("remote thread lifecycle commands", () => { diff --git a/packages/client-runtime/src/state/threadCommands.ts b/packages/client-runtime/src/state/threadCommands.ts index 0ef8d2d6e1798..c9e333734959a 100644 --- a/packages/client-runtime/src/state/threadCommands.ts +++ b/packages/client-runtime/src/state/threadCommands.ts @@ -166,6 +166,28 @@ export function createThreadEnvironmentAtoms( ); }); + // Runs whether or not the server accepted the interrupt, so a stale cached item still + // settles when Stop had nothing to interrupt; the interrupt's own failure still surfaces. + // The repair is detached: its lookups must not hold the thread's serial command key, and + // closing the thread cancels the repair, not the Stop that triggered it. + const interruptTurn = Effect.fn("ThreadCommands.interruptTurn")(function* ( + input: InterruptThreadTurnInput, + ) { + const supervisor = yield* EnvironmentSupervisor.EnvironmentSupervisor; + const controller = yield* Effect.serviceOption(ThreadHistoryController.ThreadHistoryController); + return yield* interruptThreadTurn(input).pipe( + Effect.ensuring( + Option.match(controller, { + onNone: () => Effect.void, + onSome: (value) => + value + .reconcilePendingWork(supervisor.target.environmentId, input.threadId) + .pipe(Effect.forkDetach, Effect.asVoid), + }), + ), + ); + }); + const commands = { create: createEnvironmentCommand(runtime, { label: "environment-data:commands:thread:create", @@ -301,7 +323,7 @@ export function createThreadEnvironmentAtoms( }), interruptTurn: createEnvironmentCommand(runtime, { label: "environment-data:commands:thread:interrupt-turn", - execute: (input: InterruptThreadTurnInput) => interruptThreadTurn(input), + execute: interruptTurn, scheduler, concurrency, }), diff --git a/packages/client-runtime/src/state/threadHistoryController.test.ts b/packages/client-runtime/src/state/threadHistoryController.test.ts index f1a1f6b0ad04f..3d6505b02dd37 100644 --- a/packages/client-runtime/src/state/threadHistoryController.test.ts +++ b/packages/client-runtime/src/state/threadHistoryController.test.ts @@ -24,6 +24,7 @@ function handler( _tag: "loaded", } satisfies ThreadHistoryController.ThreadHistoryLoadEarlierResult); }, + reconcilePendingWork: () => Effect.void, }; } diff --git a/packages/client-runtime/src/state/threadHistoryController.ts b/packages/client-runtime/src/state/threadHistoryController.ts index 4c7a3dcc76d22..2c33feeab5aa9 100644 --- a/packages/client-runtime/src/state/threadHistoryController.ts +++ b/packages/client-runtime/src/state/threadHistoryController.ts @@ -14,6 +14,7 @@ export type ThreadHistoryLoadEarlierResult = export type ThreadHistoryHandler = { readonly loadEarlier: (throughEntryId?: string) => Effect.Effect; + readonly reconcilePendingWork: () => Effect.Effect; }; export type ThreadHistoryRegistration = { @@ -36,6 +37,10 @@ export class ThreadHistoryController extends Context.Service< threadId: ThreadId, throughEntryId?: string, ) => Effect.Effect; + readonly reconcilePendingWork: ( + environmentId: EnvironmentId, + threadId: ThreadId, + ) => Effect.Effect; } >()("@t3tools/client-runtime/state/threadHistoryController") {} @@ -81,6 +86,14 @@ export const layer: Layer.Layer = Layer.effect( return handler.loadEarlier(throughEntryId); }), ), + reconcilePendingWork: (environmentId, threadId) => + Ref.get(handlers).pipe( + Effect.flatMap( + (current) => + current.get(threadKey({ environmentId, threadId }))?.reconcilePendingWork() ?? + Effect.void, + ), + ), }); }), ); diff --git a/packages/client-runtime/src/state/threads-sync.test.ts b/packages/client-runtime/src/state/threads-sync.test.ts index b87ab18a50ae7..a772eaaf76c26 100644 --- a/packages/client-runtime/src/state/threads-sync.test.ts +++ b/packages/client-runtime/src/state/threads-sync.test.ts @@ -24,6 +24,7 @@ import * as Stream from "effect/Stream"; import * as SubscriptionRef from "effect/SubscriptionRef"; import * as TestClock from "effect/testing/TestClock"; import { HttpClient, HttpClientResponse } from "effect/http"; +import { pendingBackgroundTurnItems } from "@t3tools/shared/orchestrationV2PendingBackgroundWork"; import type { WsRpcProtocolClient } from "../rpc/protocol.ts"; import { @@ -37,6 +38,7 @@ import * as EnvironmentSupervisor from "../connection/supervisor.ts"; import * as Persistence from "../platform/persistence.ts"; import * as RpcSession from "../rpc/session.ts"; import { v2Projection, v2ThreadId } from "./orchestrationV2TestFixtures.ts"; +import { EMPTY_THREAD_HISTORY_META } from "./threadHistoryMerge.ts"; import * as ThreadHistoryController from "./threadHistoryController.ts"; import { EMPTY_ENVIRONMENT_THREAD_STATE, @@ -112,6 +114,9 @@ const makeHarness = Effect.fn("TestEnvironmentThreads.makeHarness")(function* (o readonly saveThread?: Persistence.EnvironmentCacheStore["Service"]["saveThread"]; readonly historyPaging?: "enabled" | "no-http" | "no-controller"; readonly historyHttpClient?: HttpClient.HttpClient; + readonly getTurnItem?: (input: { + readonly itemId: TurnItemId; + }) => Effect.Effect<{ readonly item: OrchestrationV2TurnItem | null }>; }) { const inputs = yield* Queue.unbounded(); const observed = yield* Queue.unbounded(); @@ -153,6 +158,8 @@ const makeHarness = Effect.fn("TestEnvironmentThreads.makeHarness")(function* (o Effect.as(streamFrom(inputs)), ), ), + [ORCHESTRATION_V2_WS_METHODS.getTurnItem]: (input: { readonly itemId: TurnItemId }) => + options?.getTurnItem?.(input) ?? Effect.die("Unexpected getTurnItem request"), } as unknown as WsRpcProtocolClient; const supervisorSession = yield* SubscriptionRef.make>( Option.some(testSession(client, options)), @@ -256,6 +263,7 @@ const makeHarness = Effect.fn("TestEnvironmentThreads.makeHarness")(function* (o threadState, loadEarlier: (throughEntryId?: string) => historyController.loadEarlier(TARGET.environmentId, THREAD_ID, throughEntryId), + reconcilePendingWork: historyController.reconcilePendingWork(TARGET.environmentId, THREAD_ID), inputs, observed, latest, @@ -578,6 +586,450 @@ describe("EnvironmentThreads", () => { }), ); + describe("Stop on cached work the server already finished (#18188)", () => { + const at = DateTime.makeUnsafe("2026-06-20T00:00:00.000Z"); + const parentThreadId = ThreadId.make("thread:parent"); + const command = ( + id: string, + status: "running" | "completed", + exitCode?: number, + ordinal = id === "older-command" ? 1 : 2, + ): OrchestrationV2TurnItem => ({ + id: TurnItemId.make(id), + threadId: THREAD_ID, + runId: null, + nodeId: null, + providerThreadId: null, + providerTurnId: null, + nativeItemRef: null, + parentItemId: null, + ordinal, + status, + title: null, + startedAt: at, + completedAt: status === "completed" ? at : null, + updatedAt: at, + type: "command_execution", + input: `run ${id}`, + output: "", + ...(exitCode === undefined ? {} : { exitCode }), + }); + const localRow = (item: OrchestrationV2TurnItem, position: number) => ({ + position, + visibility: "local" as const, + sourceThreadId: THREAD_ID, + sourceItemId: item.id, + item, + }); + // A fork's parent row keeps its own source; the child's Stop must not touch it. + const inheritedRow = { + position: 0, + visibility: "inherited" as const, + sourceThreadId: parentThreadId, + sourceItemId: TurnItemId.make("cached-command"), + item: { ...command("cached-command", "running"), threadId: parentThreadId }, + }; + const cachedProjection = (items: ReadonlyArray) => ({ + ...BASE_PROJECTION, + turnItems: items, + visibleTurnItems: [inheritedRow, ...items.map((item, index) => localRow(item, index + 1))], + }); + // Loaded older history the bounded window would drop, next to the stale item. + const cached: OrchestrationV2ThreadProjection = cachedProjection([ + command("older-command", "completed", 0), + command("cached-command", "running"), + ]); + const cachedHistory = { historyCursor: "older", hasMoreHistory: true }; + const projectionStatuses = (projection: OrchestrationV2ThreadProjection) => ({ + items: projection.turnItems.map((item) => [String(item.id), item.status]), + rows: projection.visibleTurnItems.map((row) => [ + row.visibility, + String(row.sourceItemId), + row.item.status, + ]), + pending: pendingBackgroundTurnItems(projection).map((item) => String(item.id)), + }); + const statuses = (state: EnvironmentThreadState) => + projectionStatuses(Option.getOrThrow(state.data)); + const settledStatuses = { + items: [ + ["older-command", "completed"], + ["cached-command", "completed"], + ], + rows: [ + ["inherited", "cached-command", "running"], + ["local", "older-command", "completed"], + ["local", "cached-command", "completed"], + ], + pending: [], + }; + const serverItems = (items: Record) => ({ + getTurnItem: ({ itemId }: { readonly itemId: TurnItemId }) => + Effect.succeed({ item: items[itemId] ?? null }), + }); + + it.effect("settles an item the server finished without any new event, keeping history", () => + Effect.gen(function* () { + const written = yield* Deferred.make(); + const harness = yield* makeHarness({ + cached, + cachedHistory, + saveThread: (_environmentId, thread) => Deferred.succeed(written, thread), + ...serverItems({ "cached-command": command("cached-command", "completed", 0) }), + }); + const before = yield* SubscriptionRef.get(harness.threadState); + + yield* harness.reconcilePendingWork; + + const after = yield* SubscriptionRef.get(harness.threadState); + expect(statuses(after)).toEqual(settledStatuses); + const settled = Option.getOrThrow(after.data); + expect(settled.visibleTurnItems[0]).toBe(inheritedRow); + expect(settled.visibleTurnItems[2]!.item).toBe(settled.turnItems[1]); + expect(after.history).toEqual(before.history); + yield* TestClock.adjust("500 millis"); + const saved = yield* Deferred.await(written); + expect(projectionStatuses(saved.projection)).toEqual(settledStatuses); + expect(saved).toMatchObject({ historyCursor: "older", hasMoreHistory: true }); + }), + ); + + it.effect("saves the repair after older history was loaded, without the expansion", () => + Effect.gen(function* () { + const evenOlder = command("even-older-command", "completed", 0, 0); + const scope = yield* Scope.make(); + const harness = yield* makeHarness({ + cached, + cachedHistory, + historyHttpClient: HttpClient.make((request) => + Effect.succeed( + HttpClientResponse.fromWeb( + request, + Response.json({ + snapshotSequence: CACHED_SNAPSHOT_SEQUENCE, + items: [localRow(evenOlder, 0)], + nextCursor: null, + hasMoreHistory: false, + }), + ), + ), + ), + ...serverItems({ "cached-command": command("cached-command", "completed", 0) }), + }).pipe(Scope.provide(scope)); + expect(yield* harness.loadEarlier()).toEqual({ _tag: "loaded" }); + expect((yield* SubscriptionRef.get(harness.threadState)).history.expanded).toBe(true); + + yield* harness.reconcilePendingWork; + const after = yield* SubscriptionRef.get(harness.threadState); + expect(after.history).toMatchObject({ expanded: true, historyCursor: null }); + expect(statuses(after).pending).toEqual([]); + yield* TestClock.adjust("500 millis"); + yield* Scope.close(scope, Exit.void); + + const saved = (yield* Ref.get(harness.savedThreads)).at(-1); + expect(saved).toMatchObject({ + snapshotSequence: CACHED_SNAPSHOT_SEQUENCE, + historyCursor: "older", + hasMoreHistory: true, + }); + expect(projectionStatuses(saved!.projection)).toEqual(settledStatuses); + + const reloaded = yield* makeHarness({ + cached: saved!.projection, + cachedHistory: { historyCursor: saved!.historyCursor!, hasMoreHistory: true }, + }); + expect(statuses(yield* SubscriptionRef.get(reloaded.threadState))).toEqual(settledStatuses); + }), + ); + + const expandedHarness = Effect.fn("expandedHarness")(function* ( + scope: Scope.Closeable, + options?: Pick[0]>, "getTurnItem">, + ) { + const harness = yield* makeHarness({ + cached, + cachedHistory, + historyHttpClient: HttpClient.make((request) => + Effect.succeed( + HttpClientResponse.fromWeb( + request, + Response.json({ + snapshotSequence: CACHED_SNAPSHOT_SEQUENCE, + items: [localRow(command("even-older-command", "completed", 0, 0), 0)], + nextCursor: null, + hasMoreHistory: false, + }), + ), + ), + ), + ...serverItems({ "cached-command": command("cached-command", "completed", 0) }), + ...options, + }).pipe(Scope.provide(scope)); + return harness; + }); + const expand = (harness: Effect.Success>) => + Effect.gen(function* () { + expect(yield* harness.loadEarlier()).toEqual({ _tag: "loaded" }); + expect((yield* SubscriptionRef.get(harness.threadState)).history.expanded).toBe(true); + }); + const lastSaved = (harness: Effect.Success>) => + Ref.get(harness.savedThreads).pipe(Effect.map((saved) => saved.at(-1))); + + it.effect("saves an expanded thread's repair when its scope closes before the save runs", () => + Effect.gen(function* () { + const scope = yield* Scope.make(); + const harness = yield* expandedHarness(scope); + yield* expand(harness); + + yield* harness.reconcilePendingWork; + yield* Scope.close(scope, Exit.void); + + const saved = yield* lastSaved(harness); + expect(saved).toMatchObject({ historyCursor: "older", hasMoreHistory: true }); + expect(projectionStatuses(saved!.projection)).toEqual(settledStatuses); + }), + ); + + it.effect( + "saves an expanded thread's repair when its scope closes during the save cooldown", + () => + Effect.gen(function* () { + const scope = yield* Scope.make(); + const harness = yield* expandedHarness(scope); + yield* Queue.offer(harness.inputs, titleUpdated("Renamed", CACHED_SNAPSHOT_SEQUENCE + 1)); + yield* awaitThreadState( + harness.observed, + (value) => Option.getOrThrow(value.data).thread.title === "Renamed", + ); + yield* TestClock.adjust("500 millis"); + expect((yield* lastSaved(harness))?.projection.thread.title).toBe("Renamed"); + yield* expand(harness); + + yield* harness.reconcilePendingWork; + yield* Scope.close(scope, Exit.void); + + const saved = yield* lastSaved(harness); + expect(saved).toMatchObject({ + snapshotSequence: CACHED_SNAPSHOT_SEQUENCE + 1, + historyCursor: "older", + hasMoreHistory: true, + }); + expect(saved!.projection.thread.title).toBe("Renamed"); + expect(projectionStatuses(saved!.projection)).toEqual(settledStatuses); + }), + ); + + it.effect("saves a completion event that lands while the server is asked", () => + Effect.gen(function* () { + const asked = yield* Deferred.make(); + const answer = yield* Deferred.make(); + const scope = yield* Scope.make(); + const harness = yield* expandedHarness(scope, { + getTurnItem: () => + Deferred.succeed(asked, undefined).pipe( + Effect.andThen(Deferred.await(answer)), + Effect.map((item) => ({ item })), + ), + }); + yield* expand(harness); + const reconciling = yield* harness.reconcilePendingWork.pipe(Effect.forkChild); + yield* Deferred.await(asked); + const fromEvent = command("cached-command", "completed", 1); + yield* Queue.offer(harness.inputs, { + kind: "event", + sequence: CACHED_SNAPSHOT_SEQUENCE + 1, + event: { + id: EventId.make("event-cached-command-completed"), + type: "turn-item.updated", + threadId: THREAD_ID, + occurredAt: at, + payload: fromEvent, + }, + }); + yield* awaitThreadState(harness.observed, (value) => + Option.getOrThrow(value.data).turnItems.some( + (item) => item.id === fromEvent.id && item.status === "completed", + ), + ); + + yield* Deferred.succeed(answer, command("cached-command", "completed", 0)); + yield* Fiber.join(reconciling); + yield* Scope.close(scope, Exit.void); + + const live = Option.getOrThrow((yield* SubscriptionRef.get(harness.threadState)).data); + expect(live.turnItems.find((item) => item.id === fromEvent.id)).toEqual(fromEvent); + const saved = yield* lastSaved(harness); + expect(saved).toMatchObject({ historyCursor: "older", hasMoreHistory: true }); + expect(projectionStatuses(saved!.projection)).toEqual(settledStatuses); + expect(saved!.projection.turnItems.find((item) => item.id === fromEvent.id)).toEqual( + fromEvent, + ); + }), + ); + + // The event is newer than any reply, so the saved copy follows it however the lookup ends. + it.effect.each(["running", "missing", "timed out"] as const)( + "saves a completion event that overtakes a %s lookup reply", + (reply) => + Effect.gen(function* () { + const asked = yield* Deferred.make(); + const answer = yield* Deferred.make(); + const scope = yield* Scope.make(); + const harness = yield* expandedHarness(scope, { + getTurnItem: () => + Deferred.succeed(asked, undefined).pipe( + Effect.andThen(reply === "timed out" ? Effect.never : Deferred.await(answer)), + Effect.as({ + item: reply === "running" ? command("cached-command", "running") : null, + }), + ), + }); + yield* expand(harness); + const reconciling = yield* harness.reconcilePendingWork.pipe(Effect.forkChild); + yield* Deferred.await(asked); + const fromEvent = command("cached-command", "completed", 1); + yield* Queue.offer(harness.inputs, { + kind: "event", + sequence: CACHED_SNAPSHOT_SEQUENCE + 1, + event: { + id: EventId.make("event-cached-command-completed"), + type: "turn-item.updated", + threadId: THREAD_ID, + occurredAt: at, + payload: fromEvent, + }, + }); + yield* awaitThreadState(harness.observed, (value) => + Option.getOrThrow(value.data).turnItems.some( + (item) => item.id === fromEvent.id && item.status === "completed", + ), + ); + + if (reply === "timed out") yield* TestClock.adjust("10 seconds"); + else yield* Deferred.succeed(answer, undefined); + yield* Fiber.join(reconciling); + yield* Scope.close(scope, Exit.void); + + const saved = yield* lastSaved(harness); + expect(saved).toMatchObject({ historyCursor: "older", hasMoreHistory: true }); + expect(projectionStatuses(saved!.projection)).toEqual(settledStatuses); + expect(saved!.projection.turnItems.find((item) => item.id === fromEvent.id)).toEqual( + fromEvent, + ); + }), + ); + + it.effect("saves the repair for an expanded thread resumed in the same session", () => + Effect.gen(function* () { + const resumeCache: NonNullable[1]> = { + snapshot: { + state: { + data: Option.some(cached), + status: "cached", + error: Option.none(), + history: { ...EMPTY_THREAD_HISTORY_META, historyCursor: null, expanded: true }, + }, + sequence: CACHED_SNAPSHOT_SEQUENCE, + persisted: true, + }, + owner: undefined, + }; + const scope = yield* Scope.make(); + const harness = yield* makeHarness({ + cached, + cachedHistory, + resumeCache, + ...serverItems({ "cached-command": command("cached-command", "completed", 0) }), + }).pipe(Scope.provide(scope)); + + yield* harness.reconcilePendingWork; + yield* TestClock.adjust("500 millis"); + yield* Scope.close(scope, Exit.void); + + const saved = (yield* Ref.get(harness.savedThreads)).at(-1); + expect(saved).toMatchObject({ historyCursor: "older", hasMoreHistory: true }); + expect(projectionStatuses(saved!.projection)).toEqual(settledStatuses); + }), + ); + + it.effect("gives up on a lookup the server never answers, settling the rest", () => + Effect.gen(function* () { + const askedCached = yield* Deferred.make(); + const askedSilent = yield* Deferred.make(); + const harness = yield* makeHarness({ + cached: cachedProjection([...cached.turnItems, command("silent-command", "running")]), + cachedHistory, + getTurnItem: ({ itemId }) => + itemId === "silent-command" + ? Deferred.succeed(askedSilent, undefined).pipe(Effect.andThen(Effect.never)) + : Deferred.succeed(askedCached, undefined).pipe( + Effect.as({ item: command("cached-command", "completed", 0) }), + ), + }); + const reconciling = yield* harness.reconcilePendingWork.pipe(Effect.forkChild); + yield* Deferred.await(askedCached); + yield* Deferred.await(askedSilent); + + yield* TestClock.adjust("10 seconds"); + + expect(reconciling.pollUnsafe()).toEqual(Exit.void); + expect(statuses(yield* SubscriptionRef.get(harness.threadState)).items).toEqual([ + ["older-command", "completed"], + ["cached-command", "completed"], + ["silent-command", "running"], + ]); + }), + ); + + it.effect("keeps items the server still runs or does not return", () => + Effect.gen(function* () { + const harness = yield* makeHarness({ + cached: cachedProjection([...cached.turnItems, command("missing-command", "running")]), + cachedHistory, + ...serverItems({ "cached-command": command("cached-command", "running") }), + }); + + yield* harness.reconcilePendingWork; + + expect(statuses(yield* SubscriptionRef.get(harness.threadState)).items).toEqual([ + ["older-command", "completed"], + ["cached-command", "running"], + ["missing-command", "running"], + ]); + }), + ); + + it.effect("keeps events that land while the server is asked", () => + Effect.gen(function* () { + const asked = yield* Deferred.make(); + const answer = yield* Deferred.make(); + const harness = yield* makeHarness({ + cached, + cachedHistory, + getTurnItem: () => + Deferred.succeed(asked, undefined).pipe( + Effect.andThen(Deferred.await(answer)), + Effect.map((item) => ({ item })), + ), + }); + const reconciling = yield* harness.reconcilePendingWork.pipe(Effect.forkChild); + yield* Deferred.await(asked); + yield* Queue.offer(harness.inputs, titleUpdated("Renamed", CACHED_SNAPSHOT_SEQUENCE + 1)); + yield* awaitThreadState( + harness.observed, + (value) => Option.getOrThrow(value.data).thread.title === "Renamed", + ); + + yield* Deferred.succeed(answer, command("cached-command", "completed", 0)); + yield* Fiber.join(reconciling); + + const after = yield* SubscriptionRef.get(harness.threadState); + expect(Option.getOrThrow(after.data).thread.title).toBe("Renamed"); + expect(statuses(after)).toEqual(settledStatuses); + }), + ); + }); + it.effect("keeps progressive history metadata from a bounded socket fallback", () => Effect.gen(function* () { const olderItem = { diff --git a/packages/client-runtime/src/state/threads.ts b/packages/client-runtime/src/state/threads.ts index fc8b8e5008133..31985507aac1a 100644 --- a/packages/client-runtime/src/state/threads.ts +++ b/packages/client-runtime/src/state/threads.ts @@ -1,12 +1,15 @@ import { + isOrchestrationV2WorkActive, ORCHESTRATION_V2_WS_METHODS, type EnvironmentId as EnvironmentIdType, type OrchestrationV2ThreadDetailSnapshot, type OrchestrationV2ThreadProjection, type OrchestrationV2ThreadStreamItem, + type OrchestrationV2TurnItem, type ThreadId as ThreadIdType, } from "@t3tools/contracts"; import { boundedSnapshotProjection } from "@t3tools/shared/orchestrationV2BoundedSnapshot"; +import { pendingBackgroundTurnItems } from "@t3tools/shared/orchestrationV2PendingBackgroundWork"; import * as Cause from "effect/Cause"; import * as Effect from "effect/Effect"; import * as Fiber from "effect/Fiber"; @@ -29,7 +32,7 @@ import * as ConnectionWakeups from "../connection/wakeups.ts"; import * as Persistence from "../platform/persistence.ts"; import { runCachePersistence } from "./cachePersistence.ts"; import * as ManagedRelay from "../relay/managedRelay.ts"; -import { subscribeDynamic } from "../rpc/client.ts"; +import { request, subscribeDynamic } from "../rpc/client.ts"; import { parseThreadKey, threadKey } from "./entities.ts"; import { applyOrchestrationV2ProjectionEvent } from "./orchestrationV2Projection.ts"; import { THREAD_SNAPSHOT_IDLE_TTL_MS } from "./threadRetention.ts"; @@ -120,6 +123,29 @@ function shouldPersistThread( ); } +// Only still-active local copies take the server's item: events applied meanwhile win, +// and inherited rows belong to another thread. +function settleTurnItems( + threadId: ThreadIdType, + projection: OrchestrationV2ThreadProjection, + settled: ReadonlyMap, +): OrchestrationV2ThreadProjection { + let changed = false; + const settle = (item: OrchestrationV2TurnItem, id: string) => { + const server = settled.get(id); + if (server === undefined || !isOrchestrationV2WorkActive(item.status)) return item; + changed = true; + return server; + }; + const turnItems = projection.turnItems.map((item) => settle(item, item.id)); + const visibleTurnItems = projection.visibleTurnItems.map((row) => { + if (row.sourceThreadId !== threadId) return row; + const item = settle(row.item, row.sourceItemId); + return item === row.item ? row : { ...row, item }; + }); + return changed ? { ...projection, turnItems, visibleTurnItems } : projection; +} + interface ThreadResumeSnapshot { readonly state: EnvironmentThreadState; readonly sequence: number; @@ -242,6 +268,13 @@ export const makeEnvironmentThreadState = Effect.fn("EnvironmentThreadState.make if (resumeCache?.owner === owner) resumeCache.snapshot = committed; }); const persistence = yield* Queue.sliding(1); + let queuedSnapshot = Option.getOrUndefined(cached); + let savedSnapshot = queuedSnapshot; + const queuePersist = (snapshot: OrchestrationV2ThreadDetailSnapshot) => + Effect.suspend(() => { + queuedSnapshot = snapshot; + return Queue.offer(persistence, snapshot); + }); const persist = Effect.fn("EnvironmentThreadState.persist")(function* ( snapshot: OrchestrationV2ThreadDetailSnapshot, @@ -261,6 +294,7 @@ export const makeEnvironmentThreadState = Effect.fn("EnvironmentThreadState.make yield* cache.saveThread(environmentId, snapshot).pipe( Effect.tap(() => Effect.sync(() => { + savedSnapshot = snapshot; if ( !matchesThreadSnapshot( committed, @@ -301,7 +335,11 @@ export const makeEnvironmentThreadState = Effect.fn("EnvironmentThreadState.make committed.acceptsBoundedSnapshots === true, ), ) - : Effect.void, + : // The consumer may still be waiting out its delay; an expanded thread queues + // only cache repairs, which must survive leaving the thread. + queuedSnapshot !== undefined && queuedSnapshot !== savedSnapshot + ? persist(queuedSnapshot) + : Effect.void, }); }), ); @@ -380,8 +418,7 @@ export const makeEnvironmentThreadState = Effect.fn("EnvironmentThreadState.make // streaming path. Progressive meta rides along when the window is incomplete. if (shouldPersistThread(thread, next.history)) { const snapshotSequence = yield* SubscriptionRef.get(lastSequence); - yield* Queue.offer( - persistence, + yield* queuePersist( snapshotToPersist( snapshotSequence, thread, @@ -535,8 +572,7 @@ export const makeEnvironmentThreadState = Effect.fn("EnvironmentThreadState.make } if (result._tag === "applied" && shouldPersistThread(result.projection, result.history)) { const snapshotSequence = yield* SubscriptionRef.get(lastSequence); - yield* Queue.offer( - persistence, + yield* queuePersist( snapshotToPersist( snapshotSequence, result.projection, @@ -721,11 +757,80 @@ export const makeEnvironmentThreadState = Effect.fn("EnvironmentThreadState.make ); }); + // A cached item can stay running after the server finished it, and no event will ever + // name it again: it can sit outside every bounded window, and a repeated Stop records + // nothing new (#18188). Stop asks the server about each pending item instead of + // inferring status from what a snapshot omits, and settles only those items. + const reconcilePendingWork = Effect.fn("EnvironmentThreadState.reconcilePendingWork")( + function* () { + const current = yield* SubscriptionRef.get(state); + if (current.status === "deleted" || Option.isNone(current.data)) return; + const pending = pendingBackgroundTurnItems(current.data.value); + if (pending.length === 0) return; + const queried = new Set(pending.map((item) => item.id)); + const fetched = yield* Effect.forEach( + pending, + (item) => + request(ORCHESTRATION_V2_WS_METHODS.getTurnItem, { threadId, itemId: item.id }).pipe( + Effect.timeout("10 seconds"), + Effect.map((result) => result.item), + Effect.orElseSucceed(() => null), + ), + { concurrency: 4 }, + ); + const settled = new Map( + fetched.flatMap((item) => + item === null || isOrchestrationV2WorkActive(item.status) ? [] : [[item.id, item]], + ), + ); + yield* applyLock.withPermits(1)( + Effect.gen(function* () { + const latest = yield* SubscriptionRef.get(state); + if (latest.status === "deleted" || Option.isNone(latest.data)) return; + const thread = settleTurnItems(threadId, latest.data.value, settled); + if (thread !== latest.data.value) { + yield* setThread(thread); + yield* remember; + } + // An expanded timeline is never saved, so reload would bring the stale row back. + // Repair the snapshot last handed to the cache instead, keeping its window and cursor. + // A resumed state skipped the cache load, so read the saved snapshot then. + if (shouldPersistThread(thread, latest.history)) return; + // A terminal event that landed during the lookup is newer than any reply, and the + // expanded live state never saves it, whatever the lookup returned. + const latestItems = new Map(settled); + for (const item of thread.turnItems) { + if (queried.has(item.id) && !isOrchestrationV2WorkActive(item.status)) { + latestItems.set(item.id, item); + } + } + if (latestItems.size === 0) return; + const saved = + queuedSnapshot ?? + Option.getOrUndefined( + yield* cache + .loadThread(environmentId, threadId) + .pipe(Effect.orElseSucceed(() => Option.none())), + ); + if (saved === undefined) return; + const projection = settleTurnItems(threadId, saved.projection, latestItems); + if (projection !== saved.projection) yield* queuePersist({ ...saved, projection }); + }), + ); + }, + ); + if (Option.isSome(historyController)) { const scope = yield* Scope.Scope; const registration = yield* historyController.value.register(environmentId, threadId, { loadEarlier: (throughEntryId) => loadEarlier(throughEntryId).pipe(Effect.forkIn(scope), Effect.flatMap(Fiber.join)), + reconcilePendingWork: () => + reconcilePendingWork().pipe( + Effect.provideService(EnvironmentSupervisor.EnvironmentSupervisor, supervisor), + Effect.forkIn(scope), + Effect.flatMap(Fiber.join), + ), }); yield* Effect.addFinalizer(() => historyController.value.unregister(registration)); } diff --git a/scripts/fixtures/issue-18188/seed-stale-thread-command.js b/scripts/fixtures/issue-18188/seed-stale-thread-command.js new file mode 100644 index 0000000000000..a9a3facb63ead --- /dev/null +++ b/scripts/fixtures/issue-18188/seed-stale-thread-command.js @@ -0,0 +1,68 @@ +/** + * Run in the isolated dev client's browser with preview_evaluate. + * Seed only IndexedDB; authoritative server state remains unchanged. + * Await completion, then reload before exercising Stop. + */ +globalThis.seedStaleThreadCommand = async function ({ environmentId, threadId, itemId }) { + const response = await fetch(`/api/orchestration/threads/${threadId}`, { + headers: { "x-t3-orchestration-protocol": "2" }, + }); + if (!response.ok) throw new Error(`Thread fetch failed: ${response.status}`); + const full = await response.json(); + const authoritative = full.projection.turnItems.find((item) => item.id === itemId); + const visible = full.projection.visibleTurnItems.find((row) => row.item.id === itemId); + if ( + authoritative?.type !== "command_execution" || + authoritative.status !== "completed" || + !visible + ) { + throw new Error("Choose an authoritative completed command with a visible row"); + } + return new Promise((resolve, reject) => { + const request = indexedDB.open("t3code:connection-runtime", 4); + request.addEventListener("error", () => reject(request.error)); + request.addEventListener("success", () => { + const db = request.result; + const transaction = db.transaction("thread", "readwrite"); + const store = transaction.objectStore("thread"); + const key = `${environmentId}:${threadId}`; + const read = store.get(key); + let assertion; + transaction.addEventListener("error", () => reject(transaction.error)); + transaction.addEventListener("abort", () => + reject(transaction.error ?? new Error("Seed aborted")), + ); + transaction.addEventListener("complete", () => { + db.close(); + resolve(assertion); + }); + read.addEventListener("success", () => { + if (!read.result) { + transaction.abort(); + return; + } + const saved = JSON.parse(read.result); + const snapshot = saved.snapshot; + const projection = snapshot.projection; + const stale = { ...authoritative, status: "running", completedAt: null }; + assertion = { + authoritativeStatus: authoritative.status, + omittedBeforeSeed: !projection.turnItems.some((item) => item.id === itemId), + originalCount: projection.turnItems.length, + snapshotSequence: snapshot.snapshotSequence, + historyCursor: snapshot.historyCursor, + hasMoreHistory: snapshot.hasMoreHistory, + }; + projection.turnItems = [ + stale, + ...projection.turnItems.filter((item) => item.id !== itemId), + ]; + projection.visibleTurnItems = [ + { ...visible, item: stale }, + ...projection.visibleTurnItems.filter((row) => row.item.id !== itemId), + ]; + store.put(JSON.stringify(saved), key); + }); + }); + }); +};