diff --git a/apps/mobile/src/state/threads.test.ts b/apps/mobile/src/state/threads.test.ts new file mode 100644 index 000000000000..b980188ac7d6 --- /dev/null +++ b/apps/mobile/src/state/threads.test.ts @@ -0,0 +1,408 @@ +import { describe, expect, it } from "@effect/vitest"; +import { vi } from "vite-plus/test"; +import { + CommandId, + EnvironmentId, + MessageId, + ORCHESTRATION_V2_WS_METHODS, + RunId, + type OrchestrationV2ShellSnapshot, + type OrchestrationV2ThreadProjection, + type OrchestrationV2ThreadStreamItem, +} from "@t3tools/contracts"; +import * as Deferred from "effect/Deferred"; +import * as Effect from "effect/Effect"; +import * as Layer from "effect/Layer"; +import * as Option from "effect/Option"; +import * as Queue from "effect/Queue"; +import * as Stream from "effect/Stream"; +import * as SubscriptionRef from "effect/SubscriptionRef"; +import { Atom, AtomRegistry } from "effect/reactivity"; +import type { ConnectionCatalogEntry } from "../../../../packages/client-runtime/src/connection/catalog"; +import * as EnvironmentRegistry from "../../../../packages/client-runtime/src/connection/registry"; +import * as EnvironmentSupervisor from "../../../../packages/client-runtime/src/connection/supervisor"; +import { + type PreparedConnection, + type NetworkStatus, + AVAILABLE_CONNECTION_STATE, + PrimaryConnectionTarget, +} from "../../../../packages/client-runtime/src/connection/model"; +import * as Persistence from "../../../../packages/client-runtime/src/platform/persistence"; +import type { RpcSession } from "../../../../packages/client-runtime/src/rpc/session"; +import type { WsRpcProtocolClient } from "../../../../packages/client-runtime/src/rpc/protocol"; +import * as ThreadSnapshotLoader from "../../../../packages/client-runtime/src/state/threadSnapshotHttp"; +import { + v2Now, + v2Projection, + v2ShellSnapshot, + v2ThreadShell, +} from "../../../../packages/client-runtime/src/state/orchestrationV2TestFixtures"; +import { + resolvePendingThreadCreation, + type PendingThreadCreation, +} from "./pending-thread-creation"; +import type { QueuedThreadMessage } from "./thread-outbox-model"; + +vi.mock("../connection/runtime", () => ({ + get connectionAtomRuntime() { + return runtime; + }, +})); +vi.mock("../connection/catalog", () => ({ + environmentCatalog: { catalogValueAtom: Atom.make({ isReady: true, entries: new Map() }) }, +})); +vi.mock("./shell", () => ({ environmentSnapshotAtom: () => shell })); +vi.mock("./thread-outbox", () => ({ + threadOutboxManager: { queuedMessagesByThreadKeyAtom: queued }, +})); + +const environmentId = EnvironmentId.make("environment-1"); +const ref = { environmentId, threadId: v2Projection.thread.id }; +const key = `${environmentId}:${ref.threadId}`; +const shell = Atom.make(null); +const queued = Atom.make>>({}); +let runtime: Atom.AtomRuntime< + | EnvironmentRegistry.EnvironmentRegistry + | Persistence.EnvironmentCacheStore + | ThreadSnapshotLoader.ThreadSnapshotLoader, + never +>; +const creation: QueuedThreadMessage = { + ...ref, + commandId: CommandId.make("create"), + messageId: MessageId.make("prompt"), + text: "Inspect this image", + attachments: [ + { + id: "image", + type: "image", + name: "IMG_0395.jpg", + mimeType: "image/jpeg", + sizeBytes: 10, + previewUri: "file:///image.jpg", + }, + ], + modelSelection: v2ThreadShell.modelSelection, + createdAt: "2026-10-06T20:37:12.000Z", + creation: { + projectId: v2ThreadShell.projectId, + workspaceMode: "local", + branch: null, + worktreePath: null, + }, +}; + +const makeHarness = Effect.fn("MobileThreadTest.makeHarness")(function* () { + let projection: OrchestrationV2ThreadProjection | null = null; + let loads = 0; + const subscriptions = yield* Queue.unbounded<{ + events: Queue.Queue; + closed: Deferred.Deferred; + }>(); + const target = new PrimaryConnectionTarget({ + environmentId, + label: "Test", + httpBaseUrl: "https://test.invalid", + wsBaseUrl: "wss://test.invalid", + }); + const client = { + [ORCHESTRATION_V2_WS_METHODS.subscribeThread]: () => + Stream.unwrap( + Effect.gen(function* () { + const events = yield* Queue.unbounded(); + const closed = yield* Deferred.make(); + yield* Effect.addFinalizer(() => Deferred.succeed(closed, undefined)); + yield* Queue.offer(subscriptions, { events, closed }); + return Stream.fromQueue(events); + }), + ), + } as unknown as WsRpcProtocolClient; + const session: RpcSession = { + client, + initialConfig: Effect.succeed({ threadResumeCompletionMarker: true } as never), + subscribeServerConfig: (input) => client.subscribeServerConfig(input), + ready: Effect.void, + probe: Effect.void, + closed: Effect.never, + }; + const supervisor = EnvironmentSupervisor.EnvironmentSupervisor.of({ + target, + state: yield* SubscriptionRef.make(AVAILABLE_CONNECTION_STATE), + session: yield* SubscriptionRef.make(Option.some(session)), + prepared: yield* SubscriptionRef.make>( + Option.some({ + environmentId, + label: target.label, + httpBaseUrl: target.httpBaseUrl, + socketUrl: target.wsBaseUrl, + httpAuthorization: null, + target, + }), + ), + connect: Effect.void, + disconnect: Effect.void, + retryNow: Effect.void, + }); + const registryService = EnvironmentRegistry.EnvironmentRegistry.of({ + entries: yield* SubscriptionRef.make>( + new Map(), + ), + networkStatus: yield* SubscriptionRef.make("online"), + start: Effect.void, + register: () => Effect.die("Unexpected register"), + registerPlatform: () => Effect.die("Unexpected register"), + reconcilePlatform: () => Effect.die("Unexpected reconcile"), + remove: () => Effect.die("Unexpected remove"), + removeRoute: () => Effect.die("Unexpected remove"), + reorderRoutes: () => Effect.die("Unexpected reorder"), + removeRelayEnvironments: () => Effect.die("Unexpected remove"), + retryNow: () => Effect.void, + setEnabled: () => Effect.die("Unexpected toggle"), + setCompatibility: () => Effect.die("Unexpected compatibility"), + state: () => SubscriptionRef.get(supervisor.state), + stateChanges: () => SubscriptionRef.changes(supervisor.state), + run: (_id, effect) => + Effect.provideService(effect, EnvironmentSupervisor.EnvironmentSupervisor, supervisor), + runStream: (_id, stream) => + Stream.provideService(stream, EnvironmentSupervisor.EnvironmentSupervisor, supervisor), + followStream: (_id, stream) => + Stream.provideService(stream, EnvironmentSupervisor.EnvironmentSupervisor, supervisor), + }); + runtime = Atom.runtime( + Layer.mergeAll( + Layer.succeed(EnvironmentRegistry.EnvironmentRegistry, registryService), + Layer.succeed( + Persistence.EnvironmentCacheStore, + Persistence.EnvironmentCacheStore.of({ + loadShell: () => Effect.succeedNone, + saveShell: () => Effect.void, + loadThread: () => Effect.succeedNone, + saveThread: () => Effect.void, + removeThread: () => Effect.void, + loadServerConfig: () => Effect.succeedNone, + saveServerConfig: () => Effect.void, + loadVcsRefs: () => Effect.succeedNone, + saveVcsRefs: () => Effect.void, + removeVcsRefs: () => Effect.void, + clearVcsRefs: () => Effect.void, + clear: () => Effect.void, + }), + ), + Layer.succeed( + ThreadSnapshotLoader.ThreadSnapshotLoader, + ThreadSnapshotLoader.ThreadSnapshotLoader.of({ + load: () => + Effect.sync(() => { + loads += 1; + return projection === null + ? { _tag: "missing" as const } + : { _tag: "present" as const, snapshot: { snapshotSequence: 1, projection } }; + }), + }), + ), + ), + ); + const registry = yield* Effect.acquireRelease( + Effect.sync(() => AtomRegistry.make({ defaultIdleTTL: 60_000, timeoutResolution: 1 })), + (registry) => Effect.sync(() => registry.dispose()), + ); + vi.resetModules(); + const { environmentThreadDetails } = yield* Effect.promise(() => import("./threads")); + const { pendingThreadCreationOutcomesAtom } = yield* Effect.promise( + () => import("./pending-thread-creation"), + ); + return { + registry, + details: environmentThreadDetails, + outcomes: pendingThreadCreationOutcomesAtom, + subscriptions, + loads: () => loads, + setProjection: (value: OrchestrationV2ThreadProjection) => { + projection = value; + }, + }; +}); + +describe("mobile creation detail subscriptions", () => { + it.effect.each(["shell", "delivery"] as const)( + "holds detail readers until %s, then retires setup through run completion and remount", + (confirmation) => + Effect.gen(function* () { + const h = yield* makeHarness(); + h.registry.set(queued, { [key]: [creation] }); + let previous: PendingThreadCreation | null = { message: creation, outcome: null }; + const state = h.details.stateAtom(ref); + const readers = [ + h.details.threadAtom(ref), + h.details.queueWorkflowAtom(ref), + h.details.queuedCountAtom(ref), + h.details.turnSubagentsAtom(ref), + ]; + const unmounts = readers.map((atom) => h.registry.mount(atom as Atom.Atom)); + // The raw state machine returns a definitive missing snapshot if any reader + // bypasses the creation gate. Observe its settled state, not a timed sleep. + yield* AtomRegistry.toStream(h.registry, state).pipe( + Stream.filter((value) => value.status === "empty" || value.status === "deleted"), + Stream.runHead, + ); + expect(h.loads()).toBe(0); + expect(h.registry.get(state).status).toBe("empty"); + previous = resolvePendingThreadCreation({ + threadKey: key, + pending: previous, + previous: null, + detail: Option.getOrNull(h.registry.get(state).data), + }); + expect(previous).not.toBeNull(); + + h.setProjection(v2Projection); + if (confirmation === "shell") { + h.registry.set(shell, v2ShellSnapshot); + } else { + h.registry.set(h.outcomes, { [key]: { kind: "delivered", message: creation } }); + } + const subscription = yield* Queue.take(h.subscriptions); + yield* Queue.offer(subscription.events, { kind: "synchronized" }); + yield* AtomRegistry.toStream(h.registry, state).pipe( + Stream.filter((value) => value.status === "live"), + Stream.runHead, + ); + expect(h.loads()).toBe(1); + const delivered: PendingThreadCreation = { + message: creation, + outcome: { kind: "delivered", message: creation }, + }; + h.registry.set(h.outcomes, { [key]: delivered.outcome! }); + h.registry.set(queued, {}); + h.registry.set(shell, v2ShellSnapshot); + previous = resolvePendingThreadCreation({ + threadKey: key, + pending: delivered, + previous, + detail: Option.getOrNull(h.registry.get(state).data), + }); + expect(previous).not.toBeNull(); + + const runId = RunId.make("run-1"); + for (const status of ["running", "completed"] as const) { + const projection: OrchestrationV2ThreadProjection = { + ...v2Projection, + messages: [ + { + id: creation.messageId, + threadId: ref.threadId, + role: "user", + text: creation.text, + attachments: [], + createdBy: "user", + creationSource: "mobile", + runId, + nodeId: null, + streaming: false, + createdAt: v2Now, + updatedAt: v2Now, + }, + ], + runs: [ + { + id: runId, + threadId: ref.threadId, + status, + ordinal: 1, + providerInstanceId: v2ThreadShell.providerInstanceId, + modelSelection: v2ThreadShell.modelSelection, + providerThreadId: null, + userMessageId: creation.messageId, + rootNodeId: null, + activeAttemptId: null, + checkpointId: null, + contextHandoffId: null, + requestedAt: v2Now, + startedAt: v2Now, + completedAt: status === "completed" ? v2Now : null, + }, + ], + }; + h.setProjection(projection); + yield* Queue.offer(subscription.events, { + kind: "snapshot", + snapshotSequence: status === "running" ? 2 : 3, + projection, + }); + yield* AtomRegistry.toStream(h.registry, state).pipe( + Stream.filter((value) => Option.getOrNull(value.data)?.runs[0]?.status === status), + Stream.runHead, + ); + h.registry.set(h.outcomes, {}); + previous = resolvePendingThreadCreation({ + threadKey: key, + pending: null, + previous, + detail: Option.getOrNull(h.registry.get(state).data), + }); + expect(previous).toBeNull(); + } + for (const unmount of unmounts) unmount(); + yield* Deferred.await(subscription.closed); + const unmount = h.registry.mount(h.details.threadAtom(ref)); + const resumed = yield* Queue.take(h.subscriptions); + yield* Queue.offer(resumed.events, { kind: "synchronized" }); + yield* AtomRegistry.toStream(h.registry, state).pipe( + Stream.filter((value) => value.status === "live"), + Stream.runHead, + ); + expect(Option.getOrThrow(h.registry.get(state).data).runs[0]?.status).toBe("completed"); + expect( + resolvePendingThreadCreation({ + threadKey: key, + pending: null, + previous, + detail: Option.getOrNull(h.registry.get(state).data), + }), + ).toBeNull(); + unmount(); + }), + ); + + it.effect("keeps failed creations recoverable without requesting a missing thread", () => + Effect.gen(function* () { + const h = yield* makeHarness(); + const failed: PendingThreadCreation = { + message: creation, + outcome: { kind: "failed", message: creation, reason: "Creation rejected" }, + }; + h.registry.set(h.outcomes, { [key]: failed.outcome! }); + const state = h.details.stateAtom(ref); + const unmount = h.registry.mount(h.details.threadAtom(ref)); + expect(h.registry.get(state).status).toBe("empty"); + expect(h.loads()).toBe(0); + expect( + resolvePendingThreadCreation({ + threadKey: key, + pending: failed, + previous: null, + detail: Option.getOrNull(h.registry.get(state).data), + }), + ).toBe(failed); + unmount(); + }), + ); + + it.effect("still loads existing routes without an outbox creation or shell", () => + Effect.gen(function* () { + const h = yield* makeHarness(); + h.setProjection(v2Projection); + const state = h.details.stateAtom(ref); + const unmount = h.registry.mount(h.details.threadAtom(ref)); + const subscription = yield* Queue.take(h.subscriptions); + yield* Queue.offer(subscription.events, { kind: "synchronized" }); + yield* AtomRegistry.toStream(h.registry, state).pipe( + Stream.filter((value) => value.status === "live"), + Stream.runHead, + ); + expect(h.loads()).toBe(1); + expect(Option.getOrNull(h.registry.get(state).data)).toEqual(v2Projection); + unmount(); + }), + ); +}); diff --git a/apps/mobile/src/state/threads.ts b/apps/mobile/src/state/threads.ts index c056ce417f92..367b2e97d027 100644 --- a/apps/mobile/src/state/threads.ts +++ b/apps/mobile/src/state/threads.ts @@ -1,4 +1,5 @@ import { useAtomValue } from "@effect/atom-react"; +import { parseThreadKey, threadKey } from "@t3tools/client-runtime/state/entities"; import { createEnvironmentThreadDetailAtoms, createEnvironmentThreadShellAtoms, @@ -13,21 +14,51 @@ import { AsyncResult, Atom } from "effect/reactivity"; import { environmentCatalog } from "../connection/catalog"; import { connectionAtomRuntime } from "../connection/runtime"; +import { scopedThreadKey } from "../lib/scopedEntities"; import { environmentSnapshotAtom } from "./shell"; +import { pendingThreadCreationOutcomesAtom } from "./pending-thread-creation"; +import { threadOutboxManager } from "./thread-outbox"; export const threadEnvironment = createThreadEnvironmentAtoms( connectionAtomRuntime, environmentSnapshotAtom, ); -export const environmentThreads = createEnvironmentThreadStateAtoms(connectionAtomRuntime); -export const environmentThreadDetails = createEnvironmentThreadDetailAtoms( - environmentThreads.stateAtom, -); +const remoteThreads = createEnvironmentThreadStateAtoms(connectionAtomRuntime); export const environmentThreadShells = createEnvironmentThreadShellAtoms({ catalogValueAtom: environmentCatalog.catalogValueAtom, snapshotAtom: threadEnvironment.snapshotAtom, }); +// Every detail reader must wait for creation, including queue/model/subagent +// atoms that read the stand-in shell without going through thread selection. +// An early HTTP miss parks the shared remote state as deleted until it is reopened. +const creationReadyAtom = Atom.family((key: string) => { + const ref = parseThreadKey(key); + const creationKey = scopedThreadKey(ref.environmentId, ref.threadId); + return Atom.make((get) => { + if (get(environmentThreadShells.threadShellAtom(ref)) !== null) return true; + const outcome = get(pendingThreadCreationOutcomesAtom)[creationKey]; + if (outcome?.kind === "delivered") return true; + const queued = get(threadOutboxManager.queuedMessagesByThreadKeyAtom)[creationKey]; + return outcome === undefined && !queued?.some((message) => message.creation !== undefined); + }).pipe(Atom.setIdleTTL(0)); +}); +const creationAwareStateAtom = Atom.family((key: string) => { + const ref = parseThreadKey(key); + return Atom.make((get) => + get(creationReadyAtom(key)) + ? get(remoteThreads.stateAtom(ref.environmentId, ref.threadId)) + : AsyncResult.success(EMPTY_ENVIRONMENT_THREAD_STATE), + ).pipe(Atom.setIdleTTL(0)); +}); +export const environmentThreads = { + stateAtom: (environmentId: EnvironmentId, threadId: ThreadId) => + creationAwareStateAtom(threadKey({ environmentId, threadId })), +}; +export const environmentThreadDetails = createEnvironmentThreadDetailAtoms( + environmentThreads.stateAtom, +); + const EMPTY_THREAD_STATE_ATOM = Atom.make(AsyncResult.success(EMPTY_ENVIRONMENT_THREAD_STATE)).pipe( Atom.withLabel("mobile-environment-thread:empty"), ); diff --git a/packages/client-runtime/src/state/threads-sync.test.ts b/packages/client-runtime/src/state/threads-sync.test.ts index a832d46b6732..fbe93d88d7fd 100644 --- a/packages/client-runtime/src/state/threads-sync.test.ts +++ b/packages/client-runtime/src/state/threads-sync.test.ts @@ -20,6 +20,7 @@ import * as Option from "effect/Option"; import * as Queue from "effect/Queue"; import * as Ref from "effect/Ref"; import * as Scope from "effect/Scope"; +import * as Scheduler from "effect/Scheduler"; import * as Stream from "effect/Stream"; import * as SubscriptionRef from "effect/SubscriptionRef"; import * as TestClock from "effect/testing/TestClock"; @@ -110,6 +111,7 @@ const makeHarness = Effect.fn("TestEnvironmentThreads.makeHarness")(function* (o readonly resumeCache?: NonNullable[1]>; readonly loadCached?: ReturnType; readonly saveThread?: Persistence.EnvironmentCacheStore["Service"]["saveThread"]; + readonly removeThread?: Persistence.EnvironmentCacheStore["Service"]["removeThread"]; readonly historyPaging?: "enabled" | "no-http" | "no-controller"; readonly historyHttpClient?: HttpClient.HttpClient; }) { @@ -205,8 +207,10 @@ const makeHarness = Effect.fn("TestEnvironmentThreads.makeHarness")(function* (o Ref.update(savedThreads, (current) => [...current, thread]).pipe( Effect.andThen(options?.saveThread?.(environmentId, thread) ?? Effect.void), ), - removeThread: (_environmentId, threadId) => - Ref.update(removedThreads, (current) => [...current, threadId]), + removeThread: (environmentId, threadId) => + Ref.update(removedThreads, (current) => [...current, threadId]).pipe( + Effect.andThen(options?.removeThread?.(environmentId, threadId) ?? Effect.void), + ), loadServerConfig: () => Effect.succeedNone, saveServerConfig: () => Effect.void, loadVcsRefs: () => Effect.succeedNone, @@ -1567,6 +1571,54 @@ describe("EnvironmentThreads", () => { }), ); + it.effect.each(["http", "socket"] as const)( + "rechecks a pre-creation HTTP miss on reopen and recovers through %s", + (recovery) => + Effect.gen(function* () { + const resumeCache: NonNullable[1]> = { + snapshot: undefined, + owner: undefined, + }; + yield* Effect.scoped( + Effect.gen(function* () { + const first = yield* makeHarness({ + httpSnapshot: { _tag: "missing" }, + resumeCache, + // Leaving can interrupt disk cleanup after the UI sees the miss. + removeThread: () => Effect.never, + }); + const missing = yield* awaitThreadState( + first.observed, + (value) => value.status === "deleted", + ); + expect(Option.isNone(missing.data)).toBe(true); + expect(yield* Ref.get(first.subscriptionCount)).toBe(0); + }), + ); + const resumed = yield* makeHarness({ + resumeCache, + httpSnapshot: + recovery === "http" + ? { _tag: "present", snapshot: { snapshotSequence: 7, projection: BASE_PROJECTION } } + : { _tag: "unavailable" }, + }); + if (recovery === "socket") { + yield* Queue.offer(resumed.inputs, { + kind: "snapshot", + snapshotSequence: 7, + projection: BASE_PROJECTION, + }); + } + const restored = yield* awaitThreadState( + resumed.observed, + (value) => value.status === "live" || value.status === "deleted", + ); + expect(restored.status).toBe("live"); + expect(Option.getOrNull(restored.data)).toEqual(BASE_PROJECTION); + expect(yield* Ref.get(resumed.loaderCalls)).toBe(1); + }), + ); + it.effect("falls back to the socket when the HTTP snapshot is only unavailable", () => Effect.gen(function* () { const harness = yield* makeHarness({ @@ -1935,3 +1987,63 @@ describe("EnvironmentThreads", () => { }), ); }); + +it.effect( + "a pre-creation miss is not retained when the scope closes right after it is remembered", + () => + Effect.gen(function* () { + const rememberedDeleted = yield* Deferred.make(); + let rememberingFiber: Fiber.Fiber | undefined; + let retained: NonNullable[1]>["snapshot"]; + const resumeCache: NonNullable[1]> = { + owner: undefined, + get snapshot() { + return retained; + }, + set snapshot(value) { + retained = value; + if (value?.state.status === "deleted") { + rememberingFiber = Fiber.getCurrent(); + Deferred.doneUnsafe(rememberedDeleted, Effect.void); + } + }, + }; + const scope = yield* Effect.acquireRelease(Scope.make(), (scope) => + Scope.close(scope, Exit.void), + ); + let yieldedAfterRemember = false; + const scheduler = new Scheduler.MixedScheduler(); + const baseShouldYield = scheduler.shouldYield.bind(scheduler); + scheduler.shouldYield = (fiber) => { + if ( + fiber === rememberingFiber && + !yieldedAfterRemember && + retained?.state.status === "deleted" + ) { + yieldedAfterRemember = true; + return true; + } + return baseShouldYield(fiber); + }; + yield* makeHarness({ httpSnapshot: { _tag: "missing" }, resumeCache }).pipe( + Effect.provideService(Scope.Scope, scope), + Effect.provideService(Scheduler.Scheduler, scheduler), + ); + yield* Deferred.await(rememberedDeleted); + yield* Scope.close(scope, Exit.void); + expect(yieldedAfterRemember).toBe(true); + const resumed = yield* makeHarness({ + resumeCache, + httpSnapshot: { + _tag: "present", + snapshot: { snapshotSequence: 7, projection: BASE_PROJECTION }, + }, + }); + const restored = yield* awaitThreadState( + resumed.observed, + (value) => value.status === "live" || value.status === "deleted", + ); + expect(restored.status).toBe("live"); + expect(yield* Ref.get(resumed.loaderCalls)).toBe(1); + }), +); diff --git a/packages/client-runtime/src/state/threads.ts b/packages/client-runtime/src/state/threads.ts index 62df7cd30dfb..a580d7f336e8 100644 --- a/packages/client-runtime/src/state/threads.ts +++ b/packages/client-runtime/src/state/threads.ts @@ -412,16 +412,27 @@ export const makeEnvironmentThreadState = Effect.fn("EnvironmentThreadState.make history: patch(current.history), })); - const setDeleted = Effect.fn("EnvironmentThreadState.setDeleted")(function* () { - yield* Ref.set(awaitingCompletion, false); - yield* SubscriptionRef.set(state, { - data: Option.none(), - status: "deleted", - error: Option.none(), - history: EMPTY_THREAD_HISTORY_META, - }); - yield* remember; - if (resumeCache !== undefined && resumeCache.owner !== owner) return; + const setDeleted = Effect.fn("EnvironmentThreadState.setDeleted")(function* (options?: { + readonly missing?: boolean; + }) { + const ownsCache = yield* Effect.uninterruptible( + Effect.gen(function* () { + yield* Ref.set(awaitingCompletion, false); + yield* SubscriptionRef.set(state, { + data: Option.none(), + status: "deleted", + error: Option.none(), + history: EMPTY_THREAD_HISTORY_META, + }); + yield* remember; + if (resumeCache !== undefined && resumeCache.owner !== owner) return false; + // An HTTP miss can race creation. Recheck on reopen, even if leaving the + // screen interrupts disk cleanup; retain actual thread.deleted events. + if (options?.missing && resumeCache !== undefined) resumeCache.snapshot = undefined; + return true; + }), + ); + if (!ownsCache) return; yield* cache.removeThread(environmentId, threadId).pipe( Effect.catch((error) => Effect.logWarning("Could not remove the cached thread.").pipe( @@ -894,7 +905,7 @@ export const makeEnvironmentThreadState = Effect.fn("EnvironmentThreadState.make case "missing": { // Definitive HTTP 404: clear any stale cache and do not open or // retry a socket subscription for this attempt. - yield* setDeleted(); + yield* setDeleted({ missing: true }); return yield* Effect.never; } case "unavailable": {