From fad44e178d5c0f7eb1fff0c1eaf54d1c4f086575 Mon Sep 17 00:00:00 2001 From: Julius Marminge Date: Sun, 5 Jul 2026 12:17:27 -0700 Subject: [PATCH 1/6] Load thread snapshots over HTTP before live sync - Fetch initial thread snapshots via HTTP and resume subscriptions after the snapshot sequence - Add not-found handling for thread snapshot requests --- apps/mobile/src/connection/runtime.ts | 4 +- apps/server/src/auth/http.ts | 10 ++ apps/server/src/orchestration/http.ts | 21 +++ apps/server/src/ws.ts | 48 ++++-- apps/web/src/connection/runtime.ts | 4 +- apps/web/src/environments/primary/auth.ts | 2 + .../client-runtime/src/connection/errors.ts | 9 ++ packages/client-runtime/src/rpc/http.ts | 2 + .../src/state/threadSnapshotHttp.ts | 137 ++++++++++++++++++ .../src/state/threads-atoms.test.ts | 4 +- .../src/state/threads-sync.test.ts | 46 +++++- packages/client-runtime/src/state/threads.ts | 53 +++++-- packages/contracts/src/environmentHttp.ts | 39 ++++- packages/contracts/src/orchestration.ts | 8 + 14 files changed, 359 insertions(+), 28 deletions(-) create mode 100644 packages/client-runtime/src/state/threadSnapshotHttp.ts diff --git a/apps/mobile/src/connection/runtime.ts b/apps/mobile/src/connection/runtime.ts index 3698a0a5fc7..e97df575096 100644 --- a/apps/mobile/src/connection/runtime.ts +++ b/apps/mobile/src/connection/runtime.ts @@ -1,4 +1,5 @@ import { Connection } from "@t3tools/client-runtime/connection"; +import { threadSnapshotLoaderLayer } from "@t3tools/client-runtime/state/threads"; import * as Layer from "effect/Layer"; import { Atom } from "effect/unstable/reactivity"; @@ -11,10 +12,11 @@ const providedConnectionPlatformLayer = connectionPlatformLayer.pipe( type ConnectionLayerSource = | typeof Connection.layer + | typeof threadSnapshotLoaderLayer | typeof runtimeContextLayer | typeof connectionPlatformLayer; -const connectionLayer = Connection.layer.pipe( +const connectionLayer = Layer.merge(Connection.layer, threadSnapshotLoaderLayer).pipe( Layer.provideMerge(Layer.mergeAll(runtimeContextLayer, providedConnectionPlatformLayer)), ); diff --git a/apps/server/src/auth/http.ts b/apps/server/src/auth/http.ts index 71fb00b970a..780aaabde25 100644 --- a/apps/server/src/auth/http.ts +++ b/apps/server/src/auth/http.ts @@ -16,6 +16,8 @@ import { EnvironmentOperationForbiddenError, EnvironmentRequestInvalidError, type EnvironmentRequestInvalidReason, + EnvironmentResourceNotFoundError, + type EnvironmentResourceNotFoundReason, EnvironmentScopeRequiredError, EnvironmentAuthenticatedAuth, EnvironmentAuthenticatedPrincipal, @@ -137,6 +139,14 @@ function failEnvironmentOperationForbidden(reason: "current_session_revoke_not_a ); } +export function failEnvironmentNotFound(reason: EnvironmentResourceNotFoundReason) { + return currentEnvironmentTraceId.pipe( + Effect.flatMap((traceId) => + Effect.fail(new EnvironmentResourceNotFoundError({ code: "not_found", reason, traceId })), + ), + ); +} + export function failEnvironmentInternal(reason: EnvironmentInternalErrorReason, error?: unknown) { return Effect.gen(function* () { const traceId = yield* currentEnvironmentTraceId; diff --git a/apps/server/src/orchestration/http.ts b/apps/server/src/orchestration/http.ts index a148f98474b..6a89ffcd6b8 100644 --- a/apps/server/src/orchestration/http.ts +++ b/apps/server/src/orchestration/http.ts @@ -4,6 +4,7 @@ import { EnvironmentHttpApi, } from "@t3tools/contracts"; import * as Effect from "effect/Effect"; +import * as Option from "effect/Option"; import * as HttpApiBuilder from "effect/unstable/httpapi/HttpApiBuilder"; import { normalizeDispatchCommand } from "./Normalizer.ts"; @@ -11,6 +12,7 @@ import { annotateEnvironmentRequest, failEnvironmentInternal, failEnvironmentInvalidRequest, + failEnvironmentNotFound, requireEnvironmentScope, } from "../auth/http.ts"; import { OrchestrationEngineService } from "./Services/OrchestrationEngine.ts"; @@ -38,6 +40,25 @@ export const orchestrationHttpApiLayer = HttpApiBuilder.group( ); }), ) + .handle( + "threadSnapshot", + Effect.fn("environment.orchestration.threadSnapshot")(function* (args) { + yield* annotateEnvironmentRequest(args.endpoint.name); + yield* requireEnvironmentScope(AuthOrchestrationReadScope); + const [threadDetail, { snapshotSequence }] = yield* Effect.all([ + projectionSnapshotQuery.getThreadDetailById(args.params.threadId), + projectionSnapshotQuery.getSnapshotSequence(), + ]).pipe( + Effect.catch((cause) => + failEnvironmentInternal("orchestration_thread_snapshot_failed", cause), + ), + ); + if (Option.isNone(threadDetail)) { + return yield* failEnvironmentNotFound("thread_not_found"); + } + return { snapshotSequence, thread: threadDetail.value }; + }), + ) .handle( "dispatch", Effect.fn("environment.orchestration.dispatch")(function* (args) { diff --git a/apps/server/src/ws.ts b/apps/server/src/ws.ts index 9020e99f670..de911658591 100644 --- a/apps/server/src/ws.ts +++ b/apps/server/src/ws.ts @@ -1114,6 +1114,41 @@ const makeWsRpcLayer = ( observeRpcStreamEffect( ORCHESTRATION_WS_METHODS.subscribeThread, Effect.gen(function* () { + const isThisThreadDetailEvent = (event: OrchestrationEvent) => + event.aggregateKind === "thread" && + event.aggregateId === input.threadId && + isThreadDetailEvent(event); + + const liveStream = orchestrationEngine.streamDomainEvents.pipe( + Stream.filter(isThisThreadDetailEvent), + Stream.map((event) => ({ + kind: "event" as const, + event, + })), + ); + + // When the client already loaded the snapshot over HTTP it passes + // that snapshot's sequence, and we resume the live subscription by + // replaying persisted events after it instead of re-sending the + // (potentially multi-KB) snapshot frame over the socket. The + // catch-up replay + live stream carry the same ordering guarantees + // as the snapshot-then-live path below; overlapping events are + // deduped by sequence on the client. + if (input.afterSequence !== undefined) { + const catchUpStream = orchestrationEngine.readEvents(input.afterSequence).pipe( + Stream.filter(isThisThreadDetailEvent), + Stream.map((event) => ({ kind: "event" as const, event })), + Stream.mapError( + (cause) => + new OrchestrationGetSnapshotError({ + message: `Failed to replay thread ${input.threadId} events`, + cause, + }), + ), + ); + return Stream.concat(catchUpStream, liveStream); + } + const [threadDetail, snapshotSequence] = yield* Effect.all([ projectionSnapshotQuery.getThreadDetailById(input.threadId).pipe( Effect.mapError( @@ -1143,19 +1178,6 @@ const makeWsRpcLayer = ( }); } - const liveStream = orchestrationEngine.streamDomainEvents.pipe( - Stream.filter( - (event) => - event.aggregateKind === "thread" && - event.aggregateId === input.threadId && - isThreadDetailEvent(event), - ), - Stream.map((event) => ({ - kind: "event" as const, - event, - })), - ); - return Stream.concat( Stream.make({ kind: "snapshot" as const, diff --git a/apps/web/src/connection/runtime.ts b/apps/web/src/connection/runtime.ts index 3698a0a5fc7..e97df575096 100644 --- a/apps/web/src/connection/runtime.ts +++ b/apps/web/src/connection/runtime.ts @@ -1,4 +1,5 @@ import { Connection } from "@t3tools/client-runtime/connection"; +import { threadSnapshotLoaderLayer } from "@t3tools/client-runtime/state/threads"; import * as Layer from "effect/Layer"; import { Atom } from "effect/unstable/reactivity"; @@ -11,10 +12,11 @@ const providedConnectionPlatformLayer = connectionPlatformLayer.pipe( type ConnectionLayerSource = | typeof Connection.layer + | typeof threadSnapshotLoaderLayer | typeof runtimeContextLayer | typeof connectionPlatformLayer; -const connectionLayer = Connection.layer.pipe( +const connectionLayer = Layer.merge(Connection.layer, threadSnapshotLoaderLayer).pipe( Layer.provideMerge(Layer.mergeAll(runtimeContextLayer, providedConnectionPlatformLayer)), ); diff --git a/apps/web/src/environments/primary/auth.ts b/apps/web/src/environments/primary/auth.ts index 1c5426d953e..f9381bcad71 100644 --- a/apps/web/src/environments/primary/auth.ts +++ b/apps/web/src/environments/primary/auth.ts @@ -221,6 +221,8 @@ function readEnvironmentHttpErrorStatus(error: EnvironmentHttpCommonErrorType): case "EnvironmentScopeRequiredError": case "EnvironmentOperationForbiddenError": return 403; + case "EnvironmentResourceNotFoundError": + return 404; case "EnvironmentInternalError": return 500; } diff --git a/packages/client-runtime/src/connection/errors.ts b/packages/client-runtime/src/connection/errors.ts index f70e41adfe7..fccce583392 100644 --- a/packages/client-runtime/src/connection/errors.ts +++ b/packages/client-runtime/src/connection/errors.ts @@ -132,6 +132,15 @@ export function mapRemoteEnvironmentError( detail: "The environment rejected the authentication request.", traceId: error.traceId, }); + case "EnvironmentResourceNotFoundError": + // Not expected during connection authorization, but the shared request + // error type now includes it (used by resource fetches like the thread + // snapshot). Treat it as a configuration issue with the endpoint. + return new ConnectionBlockedError({ + reason: "configuration", + detail: "The environment endpoint could not be found.", + traceId: error.traceId, + }); case "RemoteEnvironmentAuthTimeoutError": return new ConnectionTransientError({ reason: "timeout", diff --git a/packages/client-runtime/src/rpc/http.ts b/packages/client-runtime/src/rpc/http.ts index 11bfa794fa7..87495918598 100644 --- a/packages/client-runtime/src/rpc/http.ts +++ b/packages/client-runtime/src/rpc/http.ts @@ -5,6 +5,7 @@ import { type EnvironmentInternalError, type EnvironmentOperationForbiddenError, type EnvironmentRequestInvalidError, + type EnvironmentResourceNotFoundError, type EnvironmentScopeRequiredError, } from "@t3tools/contracts"; import { httpHeaderRedactionLayer } from "@t3tools/shared/httpObservability"; @@ -70,6 +71,7 @@ export type RemoteEnvironmentRequestError = | EnvironmentAuthInvalidError | EnvironmentScopeRequiredError | EnvironmentOperationForbiddenError + | EnvironmentResourceNotFoundError | EnvironmentInternalError | RemoteEnvironmentAuthFetchError | RemoteEnvironmentAuthInvalidJsonError diff --git a/packages/client-runtime/src/state/threadSnapshotHttp.ts b/packages/client-runtime/src/state/threadSnapshotHttp.ts new file mode 100644 index 00000000000..be00f782ad8 --- /dev/null +++ b/packages/client-runtime/src/state/threadSnapshotHttp.ts @@ -0,0 +1,137 @@ +import type { OrchestrationThreadDetailSnapshot, ThreadId } from "@t3tools/contracts"; +import * as Cause from "effect/Cause"; +import * as Context from "effect/Context"; +import * as Effect from "effect/Effect"; +import * as Layer from "effect/Layer"; +import * as Option from "effect/Option"; +import { HttpClient, type HttpMethod } from "effect/unstable/http"; + +import type { PreparedConnection, PreparedHttpAuthorization } from "../connection/model.ts"; +import { environmentEndpointUrl } from "../environment/endpoint.ts"; +import { ManagedRelayDpopSigner } from "../relay/managedRelay.ts"; +import { + RemoteEnvironmentAuthFetchError, + executeEnvironmentHttpRequest, + makeEnvironmentHttpApiClient, + type RemoteEnvironmentRequestError, +} from "../rpc/http.ts"; + +const DEFAULT_THREAD_SNAPSHOT_TIMEOUT_MS = 10_000; + +interface EnvironmentHttpAuthHeaders { + readonly authorization?: string; + readonly dpop?: string; +} + +/** + * Build the authorization headers for an authenticated environment HTTP + * request, matching the credential the connection was prepared with: + * - primary/local connections carry no credential, + * - bearer connections send a static `Bearer` token, + * - relay connections send a `DPoP` access token with a freshly signed proof + * bound to this request's method and URL. + */ +const buildAuthHeaders = ( + authorization: PreparedHttpAuthorization | null, + method: HttpMethod.HttpMethod, + url: string, +): Effect.Effect< + EnvironmentHttpAuthHeaders, + RemoteEnvironmentAuthFetchError, + ManagedRelayDpopSigner +> => + Effect.gen(function* () { + if (authorization === null) { + return {}; + } + if (authorization._tag === "Bearer") { + return { authorization: `Bearer ${authorization.token}` }; + } + const signer = yield* ManagedRelayDpopSigner; + const proof = yield* signer + .createProof({ method, url, accessToken: authorization.accessToken }) + .pipe( + Effect.mapError( + (cause) => + new RemoteEnvironmentAuthFetchError({ + message: "Could not create the thread snapshot authorization proof.", + cause, + }), + ), + ); + return { authorization: `DPoP ${authorization.accessToken}`, dpop: proof }; + }); + +/** + * Load a thread's detail snapshot over HTTP instead of embedding it in the + * WebSocket subscription's first frame. The response is gzip-compressible by + * the transport and keeps the (potentially multi-KB) snapshot off the socket. + */ +export const fetchEnvironmentThreadSnapshot = Effect.fn( + "clientRuntime.state.fetchEnvironmentThreadSnapshot", +)(function* (input: { + readonly prepared: PreparedConnection; + readonly threadId: ThreadId; + readonly timeoutMs?: number; +}) { + const requestUrl = environmentEndpointUrl( + input.prepared.httpBaseUrl, + `/api/orchestration/threads/${input.threadId}`, + ); + const client = yield* makeEnvironmentHttpApiClient(input.prepared.httpBaseUrl); + const headers = yield* buildAuthHeaders(input.prepared.httpAuthorization, "GET", requestUrl); + return yield* executeEnvironmentHttpRequest( + requestUrl, + input.timeoutMs ?? DEFAULT_THREAD_SNAPSHOT_TIMEOUT_MS, + client.orchestration.threadSnapshot({ + params: { threadId: input.threadId }, + headers, + }), + ); +}); + +export type FetchEnvironmentThreadSnapshotError = RemoteEnvironmentRequestError; + +/** + * Loads a thread's detail snapshot over HTTP, returning `Option.none()` when it + * cannot be loaded (so the caller falls back to the socket-embedded snapshot). + * Decouples the thread state machine from the underlying HTTP + DPoP details and + * keeps them out of test contexts. + */ +export class ThreadSnapshotLoader extends Context.Service< + ThreadSnapshotLoader, + { + readonly load: ( + prepared: PreparedConnection, + threadId: ThreadId, + ) => Effect.Effect>; + } +>()("@t3tools/client-runtime/state/threadSnapshotHttp/ThreadSnapshotLoader") {} + +export const threadSnapshotLoaderLayer: Layer.Layer< + ThreadSnapshotLoader, + never, + HttpClient.HttpClient | ManagedRelayDpopSigner +> = Layer.effect( + ThreadSnapshotLoader, + Effect.gen(function* () { + const httpClient = yield* HttpClient.HttpClient; + const signer = yield* ManagedRelayDpopSigner; + return ThreadSnapshotLoader.of({ + load: (prepared: PreparedConnection, threadId: ThreadId) => + fetchEnvironmentThreadSnapshot({ prepared, threadId }).pipe( + Effect.map(Option.some), + Effect.provideService(HttpClient.HttpClient, httpClient), + Effect.provideService(ManagedRelayDpopSigner, signer), + Effect.catchCause((cause) => + Effect.logWarning( + "Could not load the thread snapshot over HTTP; using the socket snapshot instead.", + ).pipe( + Effect.annotateLogs({ threadId, cause: Cause.pretty(cause) }), + Effect.as(Option.none()), + ), + ), + ), + }); + }), +); diff --git a/packages/client-runtime/src/state/threads-atoms.test.ts b/packages/client-runtime/src/state/threads-atoms.test.ts index 420f9412b68..0a9b2cd21b1 100644 --- a/packages/client-runtime/src/state/threads-atoms.test.ts +++ b/packages/client-runtime/src/state/threads-atoms.test.ts @@ -6,12 +6,12 @@ import { Atom } from "effect/unstable/reactivity"; import type { EnvironmentRegistry } from "../connection/registry.ts"; import type { EnvironmentCacheStore } from "../platform/persistence.ts"; import { THREAD_STATE_IDLE_TTL_MS } from "./threadRetention.ts"; -import { createEnvironmentThreadStateAtoms } from "./threads.ts"; +import { createEnvironmentThreadStateAtoms, type ThreadSnapshotLoader } from "./threads.ts"; describe("createEnvironmentThreadStateAtoms", () => { it("retains thread state across short subscriber gaps", () => { const runtime = Atom.runtime(Layer.empty) as unknown as Atom.AtomRuntime< - EnvironmentRegistry | EnvironmentCacheStore, + EnvironmentRegistry | EnvironmentCacheStore | ThreadSnapshotLoader, never >; const threads = createEnvironmentThreadStateAtoms(runtime); diff --git a/packages/client-runtime/src/state/threads-sync.test.ts b/packages/client-runtime/src/state/threads-sync.test.ts index 3a5a8b69630..6552088791f 100644 --- a/packages/client-runtime/src/state/threads-sync.test.ts +++ b/packages/client-runtime/src/state/threads-sync.test.ts @@ -6,6 +6,7 @@ import { ProviderInstanceId, ThreadId, type OrchestrationThread, + type OrchestrationThreadDetailSnapshot, type OrchestrationThreadStreamItem, } from "@t3tools/contracts"; import { describe, expect, it } from "@effect/vitest"; @@ -30,6 +31,7 @@ import * as RpcSession from "../rpc/session.ts"; import { EMPTY_ENVIRONMENT_THREAD_STATE, makeEnvironmentThreadState, + ThreadSnapshotLoader, type EnvironmentThreadState, } from "./threads.ts"; @@ -40,6 +42,14 @@ const TARGET = new PrimaryConnectionTarget({ wsBaseUrl: "wss://environment.example.test", }); const THREAD_ID = ThreadId.make("thread-1"); +const PREPARED: PreparedConnection = { + environmentId: TARGET.environmentId, + label: TARGET.label, + httpBaseUrl: TARGET.httpBaseUrl, + socketUrl: TARGET.wsBaseUrl, + httpAuthorization: null, + target: TARGET, +}; const BASE_THREAD: OrchestrationThread = { id: THREAD_ID, projectId: ProjectId.make("project-1"), @@ -89,6 +99,7 @@ function awaitThreadState( const makeHarness = Effect.fn("TestEnvironmentThreads.makeHarness")(function* (options?: { readonly cached?: OrchestrationThread; + readonly httpSnapshot?: Option.Option; }) { const inputs = yield* Queue.unbounded(); const observed = yield* Queue.unbounded(); @@ -117,7 +128,17 @@ const makeHarness = Effect.fn("TestEnvironmentThreads.makeHarness")(function* (o const supervisorSession = yield* SubscriptionRef.make>( Option.some(testSession(client)), ); - const prepared = yield* SubscriptionRef.make>(Option.none()); + const prepared = yield* SubscriptionRef.make>( + Option.some(PREPARED), + ); + const snapshotLoader = ThreadSnapshotLoader.of({ + load: (_prepared, threadId) => + Effect.succeed( + threadId === THREAD_ID + ? (options?.httpSnapshot ?? Option.none()) + : Option.none(), + ), + }); const supervisor = EnvironmentSupervisor.EnvironmentSupervisor.of({ target: TARGET, state: supervisorState, @@ -145,6 +166,7 @@ const makeHarness = Effect.fn("TestEnvironmentThreads.makeHarness")(function* (o const threadState = yield* makeEnvironmentThreadState(THREAD_ID).pipe( Effect.provideService(EnvironmentSupervisor.EnvironmentSupervisor, supervisor), Effect.provideService(Persistence.EnvironmentCacheStore, cache), + Effect.provideService(ThreadSnapshotLoader, snapshotLoader), ); yield* SubscriptionRef.changes(threadState).pipe( Stream.runForEach((state) => @@ -251,6 +273,28 @@ describe("EnvironmentThreads", () => { }), ); + it.effect("seeds the thread from the HTTP snapshot and resumes live events", () => + Effect.gen(function* () { + const httpThread: OrchestrationThread = { ...BASE_THREAD, title: "HTTP title" }; + const harness = yield* makeHarness({ + httpSnapshot: Option.some({ snapshotSequence: 1, thread: httpThread }), + }); + // No socket snapshot is pushed; only a live event arrives over the socket. + // It can only be applied if the HTTP snapshot already seeded the thread. + yield* Queue.offer(harness.inputs, titleUpdated("Live title", 2)); + + const state = yield* awaitThreadState( + harness.observed, + (value) => + value.status === "live" && + Option.isSome(value.data) && + value.data.value.title === "Live title", + ); + + expect(Option.getOrThrow(state.data).title).toBe("Live title"); + }), + ); + it.effect("ignores replayed thread events at or below the snapshot sequence", () => Effect.gen(function* () { const harness = yield* makeHarness({ cached: BASE_THREAD }); diff --git a/packages/client-runtime/src/state/threads.ts b/packages/client-runtime/src/state/threads.ts index a01baf1594d..539ff711a5e 100644 --- a/packages/client-runtime/src/state/threads.ts +++ b/packages/client-runtime/src/state/threads.ts @@ -2,6 +2,7 @@ import { ORCHESTRATION_WS_METHODS, type EnvironmentId as EnvironmentIdType, type OrchestrationThread, + type OrchestrationThreadDetailSnapshot, type OrchestrationThreadStreamItem, type ThreadId as ThreadIdType, } from "@t3tools/contracts"; @@ -18,6 +19,7 @@ import { connectionProjectionPhase } from "../connection/model.ts"; import { EnvironmentSupervisor } from "../connection/supervisor.ts"; import { EnvironmentCacheStore } from "../platform/persistence.ts"; import { subscribe } from "../rpc/client.ts"; +import { ThreadSnapshotLoader } from "./threadSnapshotHttp.ts"; import { parseThreadKey, threadKey } from "./entities.ts"; import { applyThreadDetailEvent } from "./threadReducer.ts"; import { THREAD_STATE_IDLE_TTL_MS } from "./threadRetention.ts"; @@ -44,6 +46,7 @@ export const makeEnvironmentThreadState = Effect.fn("EnvironmentThreadState.make ) { const supervisor = yield* EnvironmentSupervisor; const cache = yield* EnvironmentCacheStore; + const snapshotLoader = yield* ThreadSnapshotLoader; const environmentId = supervisor.target.environmentId; const cached = yield* cache.loadThread(environmentId, threadId).pipe( Effect.catch((error) => @@ -187,14 +190,42 @@ export const makeEnvironmentThreadState = Effect.fn("EnvironmentThreadState.make ); yield* setSynchronizing; - yield* subscribe( - ORCHESTRATION_WS_METHODS.subscribeThread, - { threadId }, - { - onExpectedFailure: setStreamError, - retryExpectedFailureAfter: "250 millis", - }, - ).pipe(Stream.runForEach(applyItem), Effect.forkScoped); + yield* Effect.forkScoped( + Effect.gen(function* () { + // Load the initial snapshot over HTTP (gzip-compressible, and off the + // socket) rather than as a multi-KB first WebSocket frame, then subscribe + // to live events resuming after the snapshot's sequence. If the snapshot + // cannot be loaded over HTTP we fall back to the socket-embedded snapshot + // so the thread still synchronizes. Overlapping events are deduped by + // sequence in applyItem. + // + // Wait for a prepared connection so we can authenticate the HTTP request; + // this mirrors the socket path, which likewise waits for a live session. + const prepared = yield* SubscriptionRef.changes(supervisor.prepared).pipe( + Stream.filter(Option.isSome), + Stream.map((current) => current.value), + Stream.runHead, + ); + + const httpSnapshot = Option.isSome(prepared) + ? yield* snapshotLoader.load(prepared.value, threadId) + : Option.none(); + + if (Option.isSome(httpSnapshot)) { + yield* applyItem({ kind: "snapshot", snapshot: httpSnapshot.value }); + } + + const subscribeInput = Option.match(httpSnapshot, { + onNone: () => ({ threadId }), + onSome: (snapshot) => ({ threadId, afterSequence: snapshot.snapshotSequence }), + }); + + yield* subscribe(ORCHESTRATION_WS_METHODS.subscribeThread, subscribeInput, { + onExpectedFailure: setStreamError, + retryExpectedFailureAfter: "250 millis", + }).pipe(Stream.runForEach(applyItem)); + }), + ); yield* Effect.addFinalizer(() => SubscriptionRef.get(state).pipe( @@ -218,7 +249,10 @@ export function threadStateChanges(environmentId: EnvironmentIdType, threadId: T } export function createEnvironmentThreadStateAtoms( - runtime: Atom.AtomRuntime, + runtime: Atom.AtomRuntime< + EnvironmentRegistry | EnvironmentCacheStore | ThreadSnapshotLoader | R, + E + >, ) { const family = Atom.family((key: string) => { const { environmentId, threadId } = parseThreadKey(key); @@ -240,6 +274,7 @@ export function createEnvironmentThreadStateAtoms( export * from "./archivedThreads.ts"; export * from "./checkpointDiff.ts"; +export * from "./threadSnapshotHttp.ts"; export * from "./composerPathSearch.ts"; export * from "./threadCommands.ts"; export * from "./threadDetail.ts"; diff --git a/packages/contracts/src/environmentHttp.ts b/packages/contracts/src/environmentHttp.ts index adc5f149cba..353ed91accf 100644 --- a/packages/contracts/src/environmentHttp.ts +++ b/packages/contracts/src/environmentHttp.ts @@ -24,12 +24,13 @@ import { AuthWebSocketTicketResult, ServerAuthSessionMethod, } from "./auth.ts"; -import { AuthSessionId, TrimmedNonEmptyString } from "./baseSchemas.ts"; +import { AuthSessionId, ThreadId, TrimmedNonEmptyString } from "./baseSchemas.ts"; import { ExecutionEnvironmentDescriptor } from "./environment.ts"; import { ClientOrchestrationCommand, DispatchResult, OrchestrationReadModel, + OrchestrationThreadDetailSnapshot, } from "./orchestration.ts"; import { RelayCloudEnvironmentHealthRequest, @@ -80,6 +81,7 @@ export const EnvironmentInternalErrorReason = Schema.Literals([ "client_sessions_load_failed", "client_session_revoke_failed", "orchestration_snapshot_failed", + "orchestration_thread_snapshot_failed", "orchestration_dispatch_failed", "internal_error", ]); @@ -155,11 +157,29 @@ export class EnvironmentInternalError extends Schema.TaggedErrorClass()( + "EnvironmentResourceNotFoundError", + { + code: Schema.Literal("not_found"), + reason: EnvironmentResourceNotFoundReason, + traceId: TrimmedNonEmptyString, + }, + { httpApiStatus: 404 }, +) { + [HttpServerRespondable.symbol]() { + return HttpServerResponse.schemaJson(EnvironmentResourceNotFoundError)(this, { status: 404 }); + } +} + export const EnvironmentHttpCommonError = Schema.Union([ EnvironmentRequestInvalidError, EnvironmentAuthInvalidError, EnvironmentScopeRequiredError, EnvironmentOperationForbiddenError, + EnvironmentResourceNotFoundError, EnvironmentInternalError, ]); export type EnvironmentHttpCommonError = typeof EnvironmentHttpCommonError.Type; @@ -269,6 +289,11 @@ const EnvironmentOrchestrationSnapshotErrors = [ EnvironmentScopeRequiredError, EnvironmentInternalError, ] as const; +const EnvironmentOrchestrationThreadSnapshotErrors = [ + EnvironmentScopeRequiredError, + EnvironmentResourceNotFoundError, + EnvironmentInternalError, +] as const; const EnvironmentOrchestrationDispatchErrors = [ EnvironmentRequestInvalidError, EnvironmentScopeRequiredError, @@ -422,6 +447,10 @@ export class EnvironmentAuthHttpApi extends HttpApiGroup.make("auth") }).middleware(EnvironmentAuthenticatedAuth), ) {} +const EnvironmentOrchestrationThreadSnapshotParams = Schema.Struct({ + threadId: ThreadId, +}); + export class EnvironmentOrchestrationHttpApi extends HttpApiGroup.make("orchestration") .add( HttpApiEndpoint.get("snapshot", "/api/orchestration/snapshot", { @@ -430,6 +459,14 @@ export class EnvironmentOrchestrationHttpApi extends HttpApiGroup.make("orchestr error: EnvironmentOrchestrationSnapshotErrors, }).middleware(EnvironmentAuthenticatedAuth), ) + .add( + HttpApiEndpoint.get("threadSnapshot", "/api/orchestration/threads/:threadId", { + headers: OptionalBearerHeaders, + params: EnvironmentOrchestrationThreadSnapshotParams, + success: OrchestrationThreadDetailSnapshot, + error: EnvironmentOrchestrationThreadSnapshotErrors, + }).middleware(EnvironmentAuthenticatedAuth), + ) .add( HttpApiEndpoint.post("dispatch", "/api/orchestration/dispatch", { headers: OptionalBearerHeaders, diff --git a/packages/contracts/src/orchestration.ts b/packages/contracts/src/orchestration.ts index 623fed0917b..c7c768fabb7 100644 --- a/packages/contracts/src/orchestration.ts +++ b/packages/contracts/src/orchestration.ts @@ -453,6 +453,14 @@ export type OrchestrationShellStreamItem = typeof OrchestrationShellStreamItem.T export const OrchestrationSubscribeThreadInput = Schema.Struct({ threadId: ThreadId, + /** + * When provided, the server skips the initial snapshot frame and instead + * replays events after this sequence before streaming live events. Clients + * that load the snapshot over HTTP pass the snapshot's sequence here so the + * live subscription resumes without a gap (overlapping events are deduped by + * sequence on the client). + */ + afterSequence: Schema.optionalKey(NonNegativeInt), }); export type OrchestrationSubscribeThreadInput = typeof OrchestrationSubscribeThreadInput.Type; From c420e9ebb77280559ba9c17eecbeb74404d49f20 Mon Sep 17 00:00:00 2001 From: Julius Marminge Date: Sun, 5 Jul 2026 13:53:07 -0700 Subject: [PATCH 2/6] Address review feedback on HTTP thread snapshot loading - ws.ts (afterSequence resume): attach the live PubSub subscription into a scope-bound buffer BEFORE draining the catch-up replay, so events published during the replay window are not lost; read the full range after the cursor (readEvents limit) so the per-thread filter can't be starved by a global cap. - ProjectionSnapshotQuery: add getThreadDetailSnapshot that reads thread detail + snapshot sequence in one transaction, so the sequence is consistent with the returned state; use it in the HTTP handler and the WS snapshot path. - OrchestrationEngine.readEvents: accept an optional limit. - client loader: bound the HTTP snapshot timeout (cached data renders meanwhile) and treat a 404 as an expected defer-to-socket case rather than a warning. Co-Authored-By: Claude Opus 4.8 --- .../checkpointing/CheckpointDiffQuery.test.ts | 5 ++ .../Layers/OrchestrationEngine.test.ts | 1 + .../Layers/OrchestrationEngine.ts | 4 +- .../Layers/ProjectionSnapshotQuery.ts | 31 ++++++++ .../Services/OrchestrationEngine.ts | 5 ++ .../Services/ProjectionSnapshotQuery.ts | 11 +++ apps/server/src/orchestration/http.ts | 19 +++-- .../project/ProjectSetupScriptRunner.test.ts | 1 + .../Layers/ProviderSessionReaper.test.ts | 1 + apps/server/src/server.test.ts | 1 + apps/server/src/serverRuntimeStartup.test.ts | 4 + apps/server/src/ws.ts | 77 +++++++++++-------- .../src/state/threadSnapshotHttp.ts | 17 +++- 13 files changed, 130 insertions(+), 47 deletions(-) diff --git a/apps/server/src/checkpointing/CheckpointDiffQuery.test.ts b/apps/server/src/checkpointing/CheckpointDiffQuery.test.ts index c1dbc833718..8e0e5fb74d5 100644 --- a/apps/server/src/checkpointing/CheckpointDiffQuery.test.ts +++ b/apps/server/src/checkpointing/CheckpointDiffQuery.test.ts @@ -107,6 +107,7 @@ describe("CheckpointDiffQuery.layer", () => { }), getThreadShellById: () => Effect.succeed(Option.none()), getThreadDetailById: () => Effect.succeed(Option.none()), + getThreadDetailSnapshot: () => Effect.succeed(Option.none()), }), ), ); @@ -199,6 +200,7 @@ describe("CheckpointDiffQuery.layer", () => { getFullThreadDiffContext: () => Effect.die("unused"), getThreadShellById: () => Effect.succeed(Option.none()), getThreadDetailById: () => Effect.succeed(Option.none()), + getThreadDetailSnapshot: () => Effect.succeed(Option.none()), }), ), ); @@ -281,6 +283,7 @@ describe("CheckpointDiffQuery.layer", () => { getFullThreadDiffContext: () => Effect.die("unused"), getThreadShellById: () => Effect.succeed(Option.none()), getThreadDetailById: () => Effect.succeed(Option.none()), + getThreadDetailSnapshot: () => Effect.succeed(Option.none()), }), ), ); @@ -348,6 +351,7 @@ describe("CheckpointDiffQuery.layer", () => { getFullThreadDiffContext: () => Effect.die("unused"), getThreadShellById: () => Effect.succeed(Option.none()), getThreadDetailById: () => Effect.succeed(Option.none()), + getThreadDetailSnapshot: () => Effect.succeed(Option.none()), }), ), ); @@ -400,6 +404,7 @@ describe("CheckpointDiffQuery.layer", () => { getFullThreadDiffContext: () => Effect.succeed(Option.none()), getThreadShellById: () => Effect.succeed(Option.none()), getThreadDetailById: () => Effect.succeed(Option.none()), + getThreadDetailSnapshot: () => Effect.succeed(Option.none()), }), ), ); diff --git a/apps/server/src/orchestration/Layers/OrchestrationEngine.test.ts b/apps/server/src/orchestration/Layers/OrchestrationEngine.test.ts index b2ef0fed0f9..edfda7c837a 100644 --- a/apps/server/src/orchestration/Layers/OrchestrationEngine.test.ts +++ b/apps/server/src/orchestration/Layers/OrchestrationEngine.test.ts @@ -200,6 +200,7 @@ describe("OrchestrationEngine", () => { getFullThreadDiffContext: () => Effect.succeed(Option.none()), getThreadShellById: () => Effect.succeed(Option.none()), getThreadDetailById: () => Effect.succeed(Option.none()), + getThreadDetailSnapshot: () => Effect.succeed(Option.none()), }), ), Layer.provide( diff --git a/apps/server/src/orchestration/Layers/OrchestrationEngine.ts b/apps/server/src/orchestration/Layers/OrchestrationEngine.ts index 7277663e948..9598d1c3ef5 100644 --- a/apps/server/src/orchestration/Layers/OrchestrationEngine.ts +++ b/apps/server/src/orchestration/Layers/OrchestrationEngine.ts @@ -306,8 +306,8 @@ const makeOrchestrationEngine = Effect.gen(function* () { Effect.annotateLogs({ sequence: commandReadModel.snapshotSequence }), ); - const readEvents: OrchestrationEngineShape["readEvents"] = (fromSequenceExclusive) => - eventStore.readFromSequence(fromSequenceExclusive); + const readEvents: OrchestrationEngineShape["readEvents"] = (fromSequenceExclusive, limit) => + eventStore.readFromSequence(fromSequenceExclusive, limit); const dispatch: OrchestrationEngineShape["dispatch"] = (command) => Effect.gen(function* () { diff --git a/apps/server/src/orchestration/Layers/ProjectionSnapshotQuery.ts b/apps/server/src/orchestration/Layers/ProjectionSnapshotQuery.ts index e36db35b107..32210436e67 100644 --- a/apps/server/src/orchestration/Layers/ProjectionSnapshotQuery.ts +++ b/apps/server/src/orchestration/Layers/ProjectionSnapshotQuery.ts @@ -9,6 +9,7 @@ import { OrchestrationReadModel, OrchestrationShellSnapshot, OrchestrationThread, + OrchestrationThreadDetailSnapshot, ProjectScript, TurnId, type OrchestrationCheckpointSummary, @@ -2033,6 +2034,35 @@ const makeProjectionSnapshotQuery = Effect.gen(function* () { ); }); + const getThreadDetailSnapshot: ProjectionSnapshotQueryShape["getThreadDetailSnapshot"] = ( + threadId, + ) => + // Read the thread detail and the snapshot sequence within a single + // transaction so the sequence is consistent with the returned state; a + // projector update landing between two separate reads could otherwise return + // a sequence ahead of the thread detail, causing the client to resume from + // too far and drop events. + sql + .withTransaction( + Effect.gen(function* () { + const thread = yield* getThreadDetailById(threadId); + if (Option.isNone(thread)) { + return Option.none(); + } + const { snapshotSequence } = yield* getSnapshotSequence(); + return Option.some({ snapshotSequence, thread: thread.value }); + }), + ) + .pipe( + Effect.mapError((error) => + isPersistenceError(error) + ? error + : toPersistenceSqlError("ProjectionSnapshotQuery.getThreadDetailSnapshot:transaction")( + error, + ), + ), + ); + return { getCommandReadModel, getSnapshot, @@ -2047,6 +2077,7 @@ const makeProjectionSnapshotQuery = Effect.gen(function* () { getFullThreadDiffContext, getThreadShellById, getThreadDetailById, + getThreadDetailSnapshot, } satisfies ProjectionSnapshotQueryShape; }); diff --git a/apps/server/src/orchestration/Services/OrchestrationEngine.ts b/apps/server/src/orchestration/Services/OrchestrationEngine.ts index acb2b7b042d..dc1b5fd3003 100644 --- a/apps/server/src/orchestration/Services/OrchestrationEngine.ts +++ b/apps/server/src/orchestration/Services/OrchestrationEngine.ts @@ -26,10 +26,15 @@ export interface OrchestrationEngineShape { * Replay persisted orchestration events from an exclusive sequence cursor. * * @param fromSequenceExclusive - Sequence cursor (exclusive). + * @param limit - Maximum number of events to read. Defaults to the event + * store's page-bounded default; pass a higher value when the caller must + * read every event after the cursor (e.g. per-thread catch-up that filters + * a small subset out of a potentially larger global range). * @returns Stream containing ordered events. */ readonly readEvents: ( fromSequenceExclusive: number, + limit?: number, ) => Stream.Stream; /** diff --git a/apps/server/src/orchestration/Services/ProjectionSnapshotQuery.ts b/apps/server/src/orchestration/Services/ProjectionSnapshotQuery.ts index 7d85f0240f7..23b291d8778 100644 --- a/apps/server/src/orchestration/Services/ProjectionSnapshotQuery.ts +++ b/apps/server/src/orchestration/Services/ProjectionSnapshotQuery.ts @@ -14,6 +14,7 @@ import type { OrchestrationReadModel, OrchestrationShellSnapshot, OrchestrationThread, + OrchestrationThreadDetailSnapshot, OrchestrationThreadShell, ProjectId, ThreadId, @@ -157,6 +158,16 @@ export interface ProjectionSnapshotQueryShape { readonly getThreadDetailById: ( threadId: ThreadId, ) => Effect.Effect, ProjectionRepositoryError>; + + /** + * Read a single active thread detail together with the projection snapshot + * sequence in one consistent transaction, so the returned `snapshotSequence` + * exactly matches the state reflected in `thread` (no interleaving projector + * update between the two reads). + */ + readonly getThreadDetailSnapshot: ( + threadId: ThreadId, + ) => Effect.Effect, ProjectionRepositoryError>; } /** diff --git a/apps/server/src/orchestration/http.ts b/apps/server/src/orchestration/http.ts index 6a89ffcd6b8..9ec5565650c 100644 --- a/apps/server/src/orchestration/http.ts +++ b/apps/server/src/orchestration/http.ts @@ -45,18 +45,17 @@ export const orchestrationHttpApiLayer = HttpApiBuilder.group( Effect.fn("environment.orchestration.threadSnapshot")(function* (args) { yield* annotateEnvironmentRequest(args.endpoint.name); yield* requireEnvironmentScope(AuthOrchestrationReadScope); - const [threadDetail, { snapshotSequence }] = yield* Effect.all([ - projectionSnapshotQuery.getThreadDetailById(args.params.threadId), - projectionSnapshotQuery.getSnapshotSequence(), - ]).pipe( - Effect.catch((cause) => - failEnvironmentInternal("orchestration_thread_snapshot_failed", cause), - ), - ); - if (Option.isNone(threadDetail)) { + const snapshot = yield* projectionSnapshotQuery + .getThreadDetailSnapshot(args.params.threadId) + .pipe( + Effect.catch((cause) => + failEnvironmentInternal("orchestration_thread_snapshot_failed", cause), + ), + ); + if (Option.isNone(snapshot)) { return yield* failEnvironmentNotFound("thread_not_found"); } - return { snapshotSequence, thread: threadDetail.value }; + return snapshot.value; }), ) .handle( diff --git a/apps/server/src/project/ProjectSetupScriptRunner.test.ts b/apps/server/src/project/ProjectSetupScriptRunner.test.ts index fdf95df0b99..15612908079 100644 --- a/apps/server/src/project/ProjectSetupScriptRunner.test.ts +++ b/apps/server/src/project/ProjectSetupScriptRunner.test.ts @@ -43,6 +43,7 @@ const makeProjectionSnapshotQueryLayer = (project: OrchestrationProject) => getFullThreadDiffContext: () => Effect.die("unused"), getThreadShellById: () => Effect.die("unused"), getThreadDetailById: () => Effect.die("unused"), + getThreadDetailSnapshot: () => Effect.die("unused"), }); const makeTerminalManagerLayer = ( diff --git a/apps/server/src/provider/Layers/ProviderSessionReaper.test.ts b/apps/server/src/provider/Layers/ProviderSessionReaper.test.ts index e976c183a43..166a88ef17b 100644 --- a/apps/server/src/provider/Layers/ProviderSessionReaper.test.ts +++ b/apps/server/src/provider/Layers/ProviderSessionReaper.test.ts @@ -209,6 +209,7 @@ describe("ProviderSessionReaper", () => { : Option.none(), ), getThreadDetailById: () => Effect.die("unused"), + getThreadDetailSnapshot: () => Effect.die("unused"), }), ), Layer.provideMerge(NodeServices.layer), diff --git a/apps/server/src/server.test.ts b/apps/server/src/server.test.ts index 26528c84d34..6a776afe48d 100644 --- a/apps/server/src/server.test.ts +++ b/apps/server/src/server.test.ts @@ -702,6 +702,7 @@ const buildAppUnderTest = (options?: { getProjectShellById: () => Effect.succeed(Option.none()), getThreadShellById: () => Effect.succeed(Option.none()), getThreadDetailById: () => Effect.succeed(Option.none()), + getThreadDetailSnapshot: () => Effect.succeed(Option.none()), getCounts: () => Effect.succeed({ projectCount: 0, threadCount: 0 }), getActiveProjectByWorkspaceRoot: () => Effect.succeed(Option.none()), getFirstActiveThreadIdByProjectId: () => Effect.succeed(Option.none()), diff --git a/apps/server/src/serverRuntimeStartup.test.ts b/apps/server/src/serverRuntimeStartup.test.ts index e331f0cd4d6..1a331bc717f 100644 --- a/apps/server/src/serverRuntimeStartup.test.ts +++ b/apps/server/src/serverRuntimeStartup.test.ts @@ -96,6 +96,7 @@ it.effect("launchStartupHeartbeat does not block the caller while counts are loa getFullThreadDiffContext: () => Effect.succeed(Option.none()), getThreadShellById: () => Effect.succeed(Option.none()), getThreadDetailById: () => Effect.succeed(Option.none()), + getThreadDetailSnapshot: () => Effect.succeed(Option.none()), }), Effect.provideService(AnalyticsService.AnalyticsService, { record: () => Effect.void, @@ -158,6 +159,7 @@ it.effect("resolveAutoBootstrapWelcomeTargets returns existing project and threa getFullThreadDiffContext: () => Effect.succeed(Option.none()), getThreadShellById: () => Effect.die("unused"), getThreadDetailById: () => Effect.die("unused"), + getThreadDetailSnapshot: () => Effect.die("unused"), }), Effect.provideService(OrchestrationEngine.OrchestrationEngineService, { readEvents: () => Stream.empty, @@ -200,6 +202,7 @@ it.effect("resolveAutoBootstrapWelcomeTargets creates a project and thread when getFullThreadDiffContext: () => Effect.succeed(Option.none()), getThreadShellById: () => Effect.die("unused"), getThreadDetailById: () => Effect.die("unused"), + getThreadDetailSnapshot: () => Effect.die("unused"), }), Effect.provideService(OrchestrationEngine.OrchestrationEngineService, { readEvents: () => Stream.empty, @@ -248,6 +251,7 @@ it.effect("resolveAutoBootstrapWelcomeTargets preserves typed UUID generation fa getFullThreadDiffContext: () => Effect.succeed(Option.none()), getThreadShellById: () => Effect.die("unused"), getThreadDetailById: () => Effect.die("unused"), + getThreadDetailSnapshot: () => Effect.die("unused"), }), Effect.provideService(OrchestrationEngine.OrchestrationEngineService, { readEvents: () => Stream.empty, diff --git a/apps/server/src/ws.ts b/apps/server/src/ws.ts index de911658591..91a14b8442a 100644 --- a/apps/server/src/ws.ts +++ b/apps/server/src/ws.ts @@ -30,6 +30,7 @@ import { OrchestrationDispatchCommandError, type OrchestrationEvent, type OrchestrationShellStreamEvent, + type OrchestrationThreadStreamItem, OrchestrationGetFullThreadDiffError, OrchestrationGetSnapshotError, OrchestrationGetTurnDiffError, @@ -1130,27 +1131,49 @@ const makeWsRpcLayer = ( // When the client already loaded the snapshot over HTTP it passes // that snapshot's sequence, and we resume the live subscription by // replaying persisted events after it instead of re-sending the - // (potentially multi-KB) snapshot frame over the socket. The - // catch-up replay + live stream carry the same ordering guarantees - // as the snapshot-then-live path below; overlapping events are - // deduped by sequence on the client. + // (potentially multi-KB) snapshot frame over the socket. + // + // The live PubSub subscription must be attached *before* draining + // the catch-up replay, otherwise events published during the replay + // window are dropped (they are past the persisted tail the replay + // read, but the live stream is not yet subscribed). So fork the + // live stream into a buffer bound to this stream's scope, then emit + // catch-up followed by the buffered/ongoing live events. Overlapping + // events are deduped by sequence on the client. + // + // Read the full range after the cursor (not the store's default + // page-bounded limit): the range is normally tiny (a fresh HTTP + // snapshot sequence) and the per-thread filter runs after reading, + // so a global cap could otherwise omit this thread's events. if (input.afterSequence !== undefined) { - const catchUpStream = orchestrationEngine.readEvents(input.afterSequence).pipe( - Stream.filter(isThisThreadDetailEvent), - Stream.map((event) => ({ kind: "event" as const, event })), - Stream.mapError( - (cause) => - new OrchestrationGetSnapshotError({ - message: `Failed to replay thread ${input.threadId} events`, - cause, - }), - ), + const afterSequence = input.afterSequence; + return Stream.unwrap( + Effect.gen(function* () { + const liveBuffer = yield* Queue.unbounded(); + yield* Effect.forkScoped( + liveStream.pipe(Stream.runForEach((item) => Queue.offer(liveBuffer, item))), + ); + const catchUpStream = orchestrationEngine + .readEvents(afterSequence, Number.MAX_SAFE_INTEGER) + .pipe( + Stream.filter(isThisThreadDetailEvent), + Stream.map((event) => ({ kind: "event" as const, event })), + Stream.mapError( + (cause) => + new OrchestrationGetSnapshotError({ + message: `Failed to replay thread ${input.threadId} events`, + cause, + }), + ), + ); + return Stream.concat(catchUpStream, Stream.fromQueue(liveBuffer)); + }), ); - return Stream.concat(catchUpStream, liveStream); } - const [threadDetail, snapshotSequence] = yield* Effect.all([ - projectionSnapshotQuery.getThreadDetailById(input.threadId).pipe( + const snapshot = yield* projectionSnapshotQuery + .getThreadDetailSnapshot(input.threadId) + .pipe( Effect.mapError( (cause) => new OrchestrationGetSnapshotError({ @@ -1158,20 +1181,9 @@ const makeWsRpcLayer = ( cause, }), ), - ), - projectionSnapshotQuery.getSnapshotSequence().pipe( - Effect.map(({ snapshotSequence }) => snapshotSequence), - Effect.mapError( - (cause) => - new OrchestrationGetSnapshotError({ - message: "Failed to load orchestration snapshot sequence", - cause, - }), - ), - ), - ]); + ); - if (Option.isNone(threadDetail)) { + if (Option.isNone(snapshot)) { return yield* new OrchestrationGetSnapshotError({ message: `Thread ${input.threadId} was not found`, cause: input.threadId, @@ -1181,10 +1193,7 @@ const makeWsRpcLayer = ( return Stream.concat( Stream.make({ kind: "snapshot" as const, - snapshot: { - snapshotSequence, - thread: threadDetail.value, - }, + snapshot: snapshot.value, }), liveStream, ); diff --git a/packages/client-runtime/src/state/threadSnapshotHttp.ts b/packages/client-runtime/src/state/threadSnapshotHttp.ts index be00f782ad8..a9a045ce7de 100644 --- a/packages/client-runtime/src/state/threadSnapshotHttp.ts +++ b/packages/client-runtime/src/state/threadSnapshotHttp.ts @@ -16,7 +16,10 @@ import { type RemoteEnvironmentRequestError, } from "../rpc/http.ts"; -const DEFAULT_THREAD_SNAPSHOT_TIMEOUT_MS = 10_000; +// Bounded so a pathologically slow endpoint cannot block the (cheaper) socket +// fallback for long. The cached thread renders while this runs, so the wait only +// delays the transition to live data on the first open, not the initial paint. +const DEFAULT_THREAD_SNAPSHOT_TIMEOUT_MS = 6_000; interface EnvironmentHttpAuthHeaders { readonly authorization?: string; @@ -123,6 +126,18 @@ export const threadSnapshotLoaderLayer: Layer.Layer< Effect.map(Option.some), Effect.provideService(HttpClient.HttpClient, httpClient), Effect.provideService(ManagedRelayDpopSigner, signer), + // A genuinely missing thread (404) is expected — the socket + // subscription is the source of truth for thread existence and will + // surface the deletion — so don't treat it as an error worth warning + // about; just defer to the socket path. + Effect.catchTag("EnvironmentResourceNotFoundError", () => + Effect.logDebug( + "Thread snapshot not found over HTTP; deferring to the socket subscription.", + ).pipe( + Effect.annotateLogs({ threadId }), + Effect.as(Option.none()), + ), + ), Effect.catchCause((cause) => Effect.logWarning( "Could not load the thread snapshot over HTTP; using the socket snapshot instead.", From 2368eb9bba8f5ffb9ebc55d4ce5fff953f14352f Mon Sep 17 00:00:00 2001 From: Julius Marminge Date: Sun, 5 Jul 2026 14:01:28 -0700 Subject: [PATCH 3/6] Make thread-snapshot loader's DPoP signer optional The loader layer resolved ManagedRelayDpopSigner eagerly at build, hard-requiring it even though only relay/DPoP connections need it. Resolve it via Effect.serviceOption and pass it through to buildAuthHeaders, so the layer only requires HttpClient; bearer/primary connections no longer depend on a signer. Co-Authored-By: Claude Opus 4.8 --- .../src/state/threadSnapshotHttp.ts | 38 +++++++++++++------ 1 file changed, 26 insertions(+), 12 deletions(-) diff --git a/packages/client-runtime/src/state/threadSnapshotHttp.ts b/packages/client-runtime/src/state/threadSnapshotHttp.ts index a9a045ce7de..2214c6d4b36 100644 --- a/packages/client-runtime/src/state/threadSnapshotHttp.ts +++ b/packages/client-runtime/src/state/threadSnapshotHttp.ts @@ -33,16 +33,17 @@ interface EnvironmentHttpAuthHeaders { * - bearer connections send a static `Bearer` token, * - relay connections send a `DPoP` access token with a freshly signed proof * bound to this request's method and URL. + * + * The DPoP signer is passed in (not resolved from context) and is only required + * for relay/DPoP connections, so bearer/primary connections work even when no + * signer is available. */ const buildAuthHeaders = ( authorization: PreparedHttpAuthorization | null, method: HttpMethod.HttpMethod, url: string, -): Effect.Effect< - EnvironmentHttpAuthHeaders, - RemoteEnvironmentAuthFetchError, - ManagedRelayDpopSigner -> => + signer: Option.Option, +): Effect.Effect => Effect.gen(function* () { if (authorization === null) { return {}; @@ -50,8 +51,13 @@ const buildAuthHeaders = ( if (authorization._tag === "Bearer") { return { authorization: `Bearer ${authorization.token}` }; } - const signer = yield* ManagedRelayDpopSigner; - const proof = yield* signer + if (Option.isNone(signer)) { + return yield* new RemoteEnvironmentAuthFetchError({ + message: "No DPoP signer is available to authorize the thread snapshot request.", + cause: authorization._tag, + }); + } + const proof = yield* signer.value .createProof({ method, url, accessToken: authorization.accessToken }) .pipe( Effect.mapError( @@ -75,6 +81,7 @@ export const fetchEnvironmentThreadSnapshot = Effect.fn( )(function* (input: { readonly prepared: PreparedConnection; readonly threadId: ThreadId; + readonly signer: Option.Option; readonly timeoutMs?: number; }) { const requestUrl = environmentEndpointUrl( @@ -82,7 +89,12 @@ export const fetchEnvironmentThreadSnapshot = Effect.fn( `/api/orchestration/threads/${input.threadId}`, ); const client = yield* makeEnvironmentHttpApiClient(input.prepared.httpBaseUrl); - const headers = yield* buildAuthHeaders(input.prepared.httpAuthorization, "GET", requestUrl); + const headers = yield* buildAuthHeaders( + input.prepared.httpAuthorization, + "GET", + requestUrl, + input.signer, + ); return yield* executeEnvironmentHttpRequest( requestUrl, input.timeoutMs ?? DEFAULT_THREAD_SNAPSHOT_TIMEOUT_MS, @@ -114,18 +126,20 @@ export class ThreadSnapshotLoader extends Context.Service< export const threadSnapshotLoaderLayer: Layer.Layer< ThreadSnapshotLoader, never, - HttpClient.HttpClient | ManagedRelayDpopSigner + HttpClient.HttpClient > = Layer.effect( ThreadSnapshotLoader, Effect.gen(function* () { const httpClient = yield* HttpClient.HttpClient; - const signer = yield* ManagedRelayDpopSigner; + // Resolve the DPoP signer optionally: it is only needed for relay/DPoP + // connections, so the loader must not hard-require it (bearer/primary + // connections work without one). + const signer = yield* Effect.serviceOption(ManagedRelayDpopSigner); return ThreadSnapshotLoader.of({ load: (prepared: PreparedConnection, threadId: ThreadId) => - fetchEnvironmentThreadSnapshot({ prepared, threadId }).pipe( + fetchEnvironmentThreadSnapshot({ prepared, threadId, signer }).pipe( Effect.map(Option.some), Effect.provideService(HttpClient.HttpClient, httpClient), - Effect.provideService(ManagedRelayDpopSigner, signer), // A genuinely missing thread (404) is expected — the socket // subscription is the source of truth for thread existence and will // surface the deletion — so don't treat it as an error worth warning From 6731ade8b8e5133f22d204a7bd33bdaff1e6c543 Mon Sep 17 00:00:00 2001 From: Julius Marminge Date: Sun, 5 Jul 2026 15:35:55 -0700 Subject: [PATCH 4/6] Use Effect.catchTags for the thread-snapshot not-found handler Effect service conventions require catchTags (object form) over catchTag even for a single tag. Co-Authored-By: Claude Opus 4.8 --- .../src/state/threadSnapshotHttp.ts | 17 +++++++++-------- 1 file changed, 9 insertions(+), 8 deletions(-) diff --git a/packages/client-runtime/src/state/threadSnapshotHttp.ts b/packages/client-runtime/src/state/threadSnapshotHttp.ts index 2214c6d4b36..34a03ed9455 100644 --- a/packages/client-runtime/src/state/threadSnapshotHttp.ts +++ b/packages/client-runtime/src/state/threadSnapshotHttp.ts @@ -144,14 +144,15 @@ export const threadSnapshotLoaderLayer: Layer.Layer< // subscription is the source of truth for thread existence and will // surface the deletion — so don't treat it as an error worth warning // about; just defer to the socket path. - Effect.catchTag("EnvironmentResourceNotFoundError", () => - Effect.logDebug( - "Thread snapshot not found over HTTP; deferring to the socket subscription.", - ).pipe( - Effect.annotateLogs({ threadId }), - Effect.as(Option.none()), - ), - ), + Effect.catchTags({ + EnvironmentResourceNotFoundError: () => + Effect.logDebug( + "Thread snapshot not found over HTTP; deferring to the socket subscription.", + ).pipe( + Effect.annotateLogs({ threadId }), + Effect.as(Option.none()), + ), + }), Effect.catchCause((cause) => Effect.logWarning( "Could not load the thread snapshot over HTTP; using the socket snapshot instead.", From 9bc191b188b58edb26bf3b7e01f3777627d6511b Mon Sep 17 00:00:00 2001 From: Julius Marminge Date: Sun, 5 Jul 2026 18:08:17 -0700 Subject: [PATCH 5/6] Minimize catch-up bytes: afterSequence from warm cache + HTTP shell snapshot MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Part 1 (thread warm-cache): persist the snapshot sequence alongside the cached thread (StoredThreadSnapshot v2) so a re-opened thread resumes from the cached sequence via afterSequence — receiving only events since then — instead of re-downloading its full body. Cold cache still loads the full snapshot over HTTP. Part 2 (shell): the shell subscription now resumes the same way. Added an afterSequence input to subscribeShell (server replays shell events after the cursor, buffering live events before the catch-up replay, as in the thread path) and a GET /api/orchestration/shell endpoint so a cold-cache full shell rides HTTP instead of the socket. The client passes the cached/loaded shell sequence. Shared the environment auth-header builder across the thread and shell HTTP loaders. Co-Authored-By: Claude Opus 4.8 --- apps/mobile/src/connection/runtime.ts | 7 +- apps/mobile/src/connection/storage.ts | 19 ++-- apps/server/src/orchestration/http.ts | 14 +++ apps/server/src/ws.ts | 54 +++++++++-- apps/web/src/connection/runtime.ts | 7 +- apps/web/src/connection/storage.ts | 21 +++-- .../src/platform/persistence.ts | 9 +- .../src/state/environmentHttpAuth.ts | 57 ++++++++++++ .../src/state/shell-sync.test.ts | 80 ++++++++++++++++- packages/client-runtime/src/state/shell.ts | 54 +++++++++-- .../src/state/shellSnapshotHttp.ts | 89 +++++++++++++++++++ .../src/state/threadSnapshotHttp.ts | 58 +----------- .../src/state/threads-sync.test.ts | 63 ++++++++++--- packages/client-runtime/src/state/threads.ts | 78 +++++++++------- packages/contracts/src/environmentHttp.ts | 8 ++ packages/contracts/src/orchestration.ts | 15 +++- 16 files changed, 490 insertions(+), 143 deletions(-) create mode 100644 packages/client-runtime/src/state/environmentHttpAuth.ts create mode 100644 packages/client-runtime/src/state/shellSnapshotHttp.ts diff --git a/apps/mobile/src/connection/runtime.ts b/apps/mobile/src/connection/runtime.ts index e97df575096..3d4bf4944f1 100644 --- a/apps/mobile/src/connection/runtime.ts +++ b/apps/mobile/src/connection/runtime.ts @@ -1,4 +1,5 @@ import { Connection } from "@t3tools/client-runtime/connection"; +import { shellSnapshotLoaderLayer } from "@t3tools/client-runtime/state/shell"; import { threadSnapshotLoaderLayer } from "@t3tools/client-runtime/state/threads"; import * as Layer from "effect/Layer"; import { Atom } from "effect/unstable/reactivity"; @@ -10,13 +11,15 @@ const providedConnectionPlatformLayer = connectionPlatformLayer.pipe( Layer.provide(runtimeContextLayer), ); +const snapshotLoaderLayer = Layer.merge(threadSnapshotLoaderLayer, shellSnapshotLoaderLayer); + type ConnectionLayerSource = | typeof Connection.layer - | typeof threadSnapshotLoaderLayer + | typeof snapshotLoaderLayer | typeof runtimeContextLayer | typeof connectionPlatformLayer; -const connectionLayer = Layer.merge(Connection.layer, threadSnapshotLoaderLayer).pipe( +const connectionLayer = Layer.merge(Connection.layer, snapshotLoaderLayer).pipe( Layer.provideMerge(Layer.mergeAll(runtimeContextLayer, providedConnectionPlatformLayer)), ); diff --git a/apps/mobile/src/connection/storage.ts b/apps/mobile/src/connection/storage.ts index 276ea3c5c08..73f4307fb23 100644 --- a/apps/mobile/src/connection/storage.ts +++ b/apps/mobile/src/connection/storage.ts @@ -16,8 +16,8 @@ import { } from "@t3tools/client-runtime/connection"; import { EnvironmentId, - OrchestrationThread, OrchestrationShellSnapshot, + OrchestrationThreadDetailSnapshot, ThreadId, } from "@t3tools/contracts"; import * as Context from "effect/Context"; @@ -32,7 +32,10 @@ import { makeCatalogStore, type SecureCatalogStorage } from "./catalog-store"; const SHELL_SNAPSHOT_CACHE_SCHEMA_VERSION = 1; const SHELL_SNAPSHOT_CACHE_DIRECTORY = "connection-shell-snapshots"; const LEGACY_SHELL_SNAPSHOT_CACHE_DIRECTORY = "shell-snapshots"; -const THREAD_SNAPSHOT_CACHE_SCHEMA_VERSION = 1; +// v2 stores the snapshot sequence alongside the thread so a warm cache can +// resume via `afterSequence` instead of re-downloading the full thread body. +// Older v1 entries (no sequence) fail to decode and are treated as a cold cache. +const THREAD_SNAPSHOT_CACHE_SCHEMA_VERSION = 2; const THREAD_SNAPSHOT_CACHE_DIRECTORY = "connection-thread-snapshots"; const StoredShellSnapshot = Schema.Struct({ @@ -45,7 +48,7 @@ const StoredThreadSnapshot = Schema.Struct({ schemaVersion: Schema.Literal(THREAD_SNAPSHOT_CACHE_SCHEMA_VERSION), environmentId: EnvironmentId, threadId: ThreadId, - thread: OrchestrationThread, + snapshot: OrchestrationThreadDetailSnapshot, }); const LegacyStoredShellSnapshot = Schema.Struct({ @@ -355,18 +358,18 @@ export const connectionStorageLayer = Layer.effectContext( Schema.decodeUnknownResult(StoredThreadSnapshot)(parsed), ).pipe(Effect.mapError((cause) => shellPersistenceError("load-thread", cause))); return stored.environmentId === environmentId && stored.threadId === threadId - ? Option.some(stored.thread) + ? Option.some(stored.snapshot) : Option.none(); }), - saveThread: (environmentId, thread) => + saveThread: (environmentId, snapshot) => Effect.gen(function* () { - const file = yield* threadSnapshotFile(environmentId, thread.id, "save-thread"); + const file = yield* threadSnapshotFile(environmentId, snapshot.thread.id, "save-thread"); const encoded = yield* Effect.fromResult( Schema.encodeUnknownResult(StoredThreadSnapshot)({ schemaVersion: THREAD_SNAPSHOT_CACHE_SCHEMA_VERSION, environmentId, - threadId: thread.id, - thread, + threadId: snapshot.thread.id, + snapshot, }), ).pipe(Effect.mapError((cause) => shellPersistenceError("save-thread", cause))); yield* Effect.try({ diff --git a/apps/server/src/orchestration/http.ts b/apps/server/src/orchestration/http.ts index 9ec5565650c..016c3d508ec 100644 --- a/apps/server/src/orchestration/http.ts +++ b/apps/server/src/orchestration/http.ts @@ -40,6 +40,20 @@ export const orchestrationHttpApiLayer = HttpApiBuilder.group( ); }), ) + .handle( + "shellSnapshot", + Effect.fn("environment.orchestration.shellSnapshot")(function* (args) { + yield* annotateEnvironmentRequest(args.endpoint.name); + yield* requireEnvironmentScope(AuthOrchestrationReadScope); + return yield* projectionSnapshotQuery + .getShellSnapshot() + .pipe( + Effect.catch((cause) => + failEnvironmentInternal("orchestration_snapshot_failed", cause), + ), + ); + }), + ) .handle( "threadSnapshot", Effect.fn("environment.orchestration.threadSnapshot")(function* (args) { diff --git a/apps/server/src/ws.ts b/apps/server/src/ws.ts index 91a14b8442a..fafdbfa6814 100644 --- a/apps/server/src/ws.ts +++ b/apps/server/src/ws.ts @@ -30,6 +30,7 @@ import { OrchestrationDispatchCommandError, type OrchestrationEvent, type OrchestrationShellStreamEvent, + type OrchestrationShellStreamItem, type OrchestrationThreadStreamItem, OrchestrationGetFullThreadDiffError, OrchestrationGetSnapshotError, @@ -1060,10 +1061,54 @@ const makeWsRpcLayer = ( ), { "rpc.aggregate": "orchestration" }, ), - [ORCHESTRATION_WS_METHODS.subscribeShell]: (_input) => + [ORCHESTRATION_WS_METHODS.subscribeShell]: (input) => observeRpcStreamEffect( ORCHESTRATION_WS_METHODS.subscribeShell, Effect.gen(function* () { + const liveStream = orchestrationEngine.streamDomainEvents.pipe( + Stream.mapEffect(toShellStreamEvent), + Stream.flatMap((event) => + Option.isSome(event) ? Stream.succeed(event.value) : Stream.empty, + ), + ); + + // When the client already holds a shell snapshot (cached, or loaded + // over HTTP) it passes that snapshot's sequence, and we resume by + // replaying shell events after it instead of re-sending the whole + // projects/threads list over the socket. As in the thread path, the + // live subscription is attached (into a scope-bound buffer) before + // draining the catch-up replay so no event published during the + // replay window is lost; overlapping events are deduped by sequence + // on the client. The full range is read (not the store's default + // page limit) since the shell filter runs after reading. + if (input.afterSequence !== undefined) { + const afterSequence = input.afterSequence; + return Stream.unwrap( + Effect.gen(function* () { + const liveBuffer = yield* Queue.unbounded(); + yield* Effect.forkScoped( + liveStream.pipe(Stream.runForEach((item) => Queue.offer(liveBuffer, item))), + ); + const catchUpStream = orchestrationEngine + .readEvents(afterSequence, Number.MAX_SAFE_INTEGER) + .pipe( + Stream.mapEffect(toShellStreamEvent), + Stream.flatMap((event) => + Option.isSome(event) ? Stream.succeed(event.value) : Stream.empty, + ), + Stream.mapError( + (cause) => + new OrchestrationGetSnapshotError({ + message: "Failed to replay orchestration shell events", + cause, + }), + ), + ); + return Stream.concat(catchUpStream, Stream.fromQueue(liveBuffer)); + }), + ); + } + const snapshot = yield* projectionSnapshotQuery.getShellSnapshot().pipe( Effect.tapError((cause) => Effect.logError("orchestration shell snapshot load failed", { cause }), @@ -1077,13 +1122,6 @@ const makeWsRpcLayer = ( ), ); - const liveStream = orchestrationEngine.streamDomainEvents.pipe( - Stream.mapEffect(toShellStreamEvent), - Stream.flatMap((event) => - Option.isSome(event) ? Stream.succeed(event.value) : Stream.empty, - ), - ); - return Stream.concat( Stream.make({ kind: "snapshot" as const, diff --git a/apps/web/src/connection/runtime.ts b/apps/web/src/connection/runtime.ts index e97df575096..3d4bf4944f1 100644 --- a/apps/web/src/connection/runtime.ts +++ b/apps/web/src/connection/runtime.ts @@ -1,4 +1,5 @@ import { Connection } from "@t3tools/client-runtime/connection"; +import { shellSnapshotLoaderLayer } from "@t3tools/client-runtime/state/shell"; import { threadSnapshotLoaderLayer } from "@t3tools/client-runtime/state/threads"; import * as Layer from "effect/Layer"; import { Atom } from "effect/unstable/reactivity"; @@ -10,13 +11,15 @@ const providedConnectionPlatformLayer = connectionPlatformLayer.pipe( Layer.provide(runtimeContextLayer), ); +const snapshotLoaderLayer = Layer.merge(threadSnapshotLoaderLayer, shellSnapshotLoaderLayer); + type ConnectionLayerSource = | typeof Connection.layer - | typeof threadSnapshotLoaderLayer + | typeof snapshotLoaderLayer | typeof runtimeContextLayer | typeof connectionPlatformLayer; -const connectionLayer = Layer.merge(Connection.layer, threadSnapshotLoaderLayer).pipe( +const connectionLayer = Layer.merge(Connection.layer, snapshotLoaderLayer).pipe( Layer.provideMerge(Layer.mergeAll(runtimeContextLayer, providedConnectionPlatformLayer)), ); diff --git a/apps/web/src/connection/storage.ts b/apps/web/src/connection/storage.ts index d118a428ed7..df311d07615 100644 --- a/apps/web/src/connection/storage.ts +++ b/apps/web/src/connection/storage.ts @@ -20,7 +20,7 @@ import { import { EnvironmentId, OrchestrationShellSnapshot, - OrchestrationThread, + OrchestrationThreadDetailSnapshot, ThreadId, } from "@t3tools/contracts"; import * as Context from "effect/Context"; @@ -45,11 +45,14 @@ const StoredShellSnapshot = Schema.Struct({ snapshot: OrchestrationShellSnapshot, }); const StoredShellSnapshotJson = Schema.fromJsonString(StoredShellSnapshot); +// v2 stores the snapshot sequence alongside the thread so a warm cache can +// resume via `afterSequence` instead of re-downloading the full thread body. +// Older v1 entries (no sequence) fail to decode and are treated as a cold cache. const StoredThreadSnapshot = Schema.Struct({ - schemaVersion: Schema.Literal(1), + schemaVersion: Schema.Literal(2), environmentId: EnvironmentId, threadId: ThreadId, - thread: OrchestrationThread, + snapshot: OrchestrationThreadDetailSnapshot, }); const StoredThreadSnapshotJson = Schema.fromJsonString(StoredThreadSnapshot); const ConnectionCatalogDocumentJson = Schema.fromJsonString(ConnectionCatalogDocument); @@ -473,7 +476,7 @@ export const connectionStorageLayer = Layer.effectContext( Effect.mapError((cause) => persistenceError("load-thread", cause)), Effect.map((stored) => stored.environmentId === environmentId && stored.threadId === threadId - ? Option.some(stored.thread) + ? Option.some(stored.snapshot) : Option.none(), ), ); @@ -484,18 +487,18 @@ export const connectionStorageLayer = Layer.effectContext( : persistenceError("load-thread", cause), ), ), - saveThread: (environmentId, thread) => + saveThread: (environmentId, snapshot) => Effect.gen(function* () { const encoded = yield* encodeStoredThreadSnapshot({ - schemaVersion: 1, + schemaVersion: 2, environmentId, - threadId: thread.id, - thread, + threadId: snapshot.thread.id, + snapshot, }).pipe(Effect.mapError((cause) => persistenceError("save-thread", cause))); yield* writeDatabaseValue( database, THREAD_STORE_NAME, - threadCacheKey(environmentId, thread.id), + threadCacheKey(environmentId, snapshot.thread.id), encoded, ); }).pipe( diff --git a/packages/client-runtime/src/platform/persistence.ts b/packages/client-runtime/src/platform/persistence.ts index 71664bf4601..047e7793611 100644 --- a/packages/client-runtime/src/platform/persistence.ts +++ b/packages/client-runtime/src/platform/persistence.ts @@ -1,7 +1,7 @@ import { type EnvironmentId, - type OrchestrationThread, type OrchestrationShellSnapshot, + type OrchestrationThreadDetailSnapshot, type ThreadId, } from "@t3tools/contracts"; import * as Context from "effect/Context"; @@ -60,10 +60,13 @@ export class EnvironmentCacheStore extends Context.Service< readonly loadThread: ( environmentId: EnvironmentId, threadId: ThreadId, - ) => Effect.Effect, ConnectionPersistenceError>; + ) => Effect.Effect< + Option.Option, + ConnectionPersistenceError + >; readonly saveThread: ( environmentId: EnvironmentId, - thread: OrchestrationThread, + snapshot: OrchestrationThreadDetailSnapshot, ) => Effect.Effect; readonly removeThread: ( environmentId: EnvironmentId, diff --git a/packages/client-runtime/src/state/environmentHttpAuth.ts b/packages/client-runtime/src/state/environmentHttpAuth.ts new file mode 100644 index 00000000000..8161a69e045 --- /dev/null +++ b/packages/client-runtime/src/state/environmentHttpAuth.ts @@ -0,0 +1,57 @@ +import * as Effect from "effect/Effect"; +import * as Option from "effect/Option"; +import type { HttpMethod } from "effect/unstable/http"; + +import type { PreparedHttpAuthorization } from "../connection/model.ts"; +import type { ManagedRelayDpopSigner } from "../relay/managedRelay.ts"; +import { RemoteEnvironmentAuthFetchError } from "../rpc/http.ts"; + +export interface EnvironmentHttpAuthHeaders { + readonly authorization?: string; + readonly dpop?: string; +} + +/** + * Build the authorization headers for an authenticated environment HTTP + * request, matching the credential the connection was prepared with: + * - primary/local connections carry no credential, + * - bearer connections send a static `Bearer` token, + * - relay connections send a `DPoP` access token with a freshly signed proof + * bound to this request's method and URL. + * + * The DPoP signer is passed in (not resolved from context) and is only required + * for relay/DPoP connections, so bearer/primary connections work even when no + * signer is available. + */ +export const buildEnvironmentAuthHeaders = ( + authorization: PreparedHttpAuthorization | null, + method: HttpMethod.HttpMethod, + url: string, + signer: Option.Option, +): Effect.Effect => + Effect.gen(function* () { + if (authorization === null) { + return {}; + } + if (authorization._tag === "Bearer") { + return { authorization: `Bearer ${authorization.token}` }; + } + if (Option.isNone(signer)) { + return yield* new RemoteEnvironmentAuthFetchError({ + message: "No DPoP signer is available to authorize the environment request.", + cause: authorization._tag, + }); + } + const proof = yield* signer.value + .createProof({ method, url, accessToken: authorization.accessToken }) + .pipe( + Effect.mapError( + (cause) => + new RemoteEnvironmentAuthFetchError({ + message: "Could not create the environment request authorization proof.", + cause, + }), + ), + ); + return { authorization: `DPoP ${authorization.accessToken}`, dpop: proof }; + }); diff --git a/packages/client-runtime/src/state/shell-sync.test.ts b/packages/client-runtime/src/state/shell-sync.test.ts index 2eab7214225..de240d58bbc 100644 --- a/packages/client-runtime/src/state/shell-sync.test.ts +++ b/packages/client-runtime/src/state/shell-sync.test.ts @@ -20,7 +20,7 @@ import * as EnvironmentSupervisor from "../connection/supervisor.ts"; import * as Persistence from "../platform/persistence.ts"; import * as RpcSession from "../rpc/session.ts"; import type { WsRpcProtocolClient } from "../rpc/protocol.ts"; -import { makeEnvironmentShellState } from "./shell.ts"; +import { makeEnvironmentShellState, ShellSnapshotLoader } from "./shell.ts"; const TARGET = new PrimaryConnectionTarget({ environmentId: EnvironmentId.make("environment-1"), @@ -29,6 +29,15 @@ const TARGET = new PrimaryConnectionTarget({ wsBaseUrl: "wss://environment.example.test", }); +const PREPARED: PreparedConnection = { + environmentId: TARGET.environmentId, + label: TARGET.label, + httpBaseUrl: TARGET.httpBaseUrl, + socketUrl: TARGET.wsBaseUrl, + httpAuthorization: null, + target: TARGET, +}; + const LIVE_SHELL_SNAPSHOT: OrchestrationShellSnapshot = { snapshotSequence: 1, projects: [], @@ -61,7 +70,7 @@ describe("environment shell synchronization", () => { target: TARGET, state: supervisorState, session: activeSession, - prepared: yield* SubscriptionRef.make(Option.none()), + prepared: yield* SubscriptionRef.make(Option.some(PREPARED)), connect: Effect.void, disconnect: Effect.void, retryNow: Effect.void, @@ -74,9 +83,15 @@ describe("environment shell synchronization", () => { removeThread: () => Effect.void, clear: () => Effect.void, }); + // Cold cache with no HTTP snapshot available → falls back to the + // socket-embedded snapshot. + const snapshotLoader = ShellSnapshotLoader.of({ + load: () => Effect.succeed(Option.none()), + }); const shellState = yield* makeEnvironmentShellState().pipe( Effect.provideService(EnvironmentSupervisor.EnvironmentSupervisor, supervisor), Effect.provideService(Persistence.EnvironmentCacheStore, cache), + Effect.provideService(ShellSnapshotLoader, snapshotLoader), ); yield* SubscriptionRef.set(supervisorState, { @@ -117,4 +132,65 @@ describe("environment shell synchronization", () => { expect(Option.getOrThrow(state.snapshot)).toEqual(LIVE_SHELL_SNAPSHOT); }), ); + + it.effect("resumes a warm shell cache via afterSequence without an HTTP fetch", () => + Effect.gen(function* () { + const cachedSnapshot: OrchestrationShellSnapshot = { + snapshotSequence: 5, + projects: [], + threads: [], + updatedAt: "2026-06-06T00:00:00.000Z", + }; + const events = yield* Queue.unbounded(); + const capturedAfterSequence = yield* SubscriptionRef.make(undefined); + const loaderCalls = yield* SubscriptionRef.make(0); + const client = { + [ORCHESTRATION_WS_METHODS.subscribeShell]: (input: { readonly afterSequence?: number }) => + Stream.unwrap( + SubscriptionRef.set(capturedAfterSequence, input.afterSequence).pipe( + Effect.as(Stream.fromQueue(events)), + ), + ), + } as unknown as WsRpcProtocolClient; + const supervisorState = yield* SubscriptionRef.make(AVAILABLE_CONNECTION_STATE); + const activeSession = yield* SubscriptionRef.make>( + Option.some(session(client)), + ); + const supervisor = EnvironmentSupervisor.EnvironmentSupervisor.of({ + target: TARGET, + state: supervisorState, + session: activeSession, + prepared: yield* SubscriptionRef.make(Option.some(PREPARED)), + connect: Effect.void, + disconnect: Effect.void, + retryNow: Effect.void, + } satisfies EnvironmentSupervisor.EnvironmentSupervisor["Service"]); + const cache = Persistence.EnvironmentCacheStore.of({ + loadShell: () => Effect.succeed(Option.some(cachedSnapshot)), + saveShell: () => Effect.void, + loadThread: () => Effect.succeed(Option.none()), + saveThread: () => Effect.void, + removeThread: () => Effect.void, + clear: () => Effect.void, + }); + const snapshotLoader = ShellSnapshotLoader.of({ + load: () => + SubscriptionRef.update(loaderCalls, (count) => count + 1).pipe(Effect.as(Option.none())), + }); + yield* makeEnvironmentShellState().pipe( + Effect.provideService(EnvironmentSupervisor.EnvironmentSupervisor, supervisor), + Effect.provideService(Persistence.EnvironmentCacheStore, cache), + Effect.provideService(ShellSnapshotLoader, snapshotLoader), + ); + + // Wait until the subscription is established from the warm cache. + yield* SubscriptionRef.changes(capturedAfterSequence).pipe( + Stream.filter((value) => value !== undefined), + Stream.runHead, + ); + + expect(yield* SubscriptionRef.get(capturedAfterSequence)).toBe(5); + expect(yield* SubscriptionRef.get(loaderCalls)).toBe(0); + }), + ); }); diff --git a/packages/client-runtime/src/state/shell.ts b/packages/client-runtime/src/state/shell.ts index 2b0ba6346f5..faa70bc4f3a 100644 --- a/packages/client-runtime/src/state/shell.ts +++ b/packages/client-runtime/src/state/shell.ts @@ -19,6 +19,7 @@ import { EnvironmentSupervisor } from "../connection/supervisor.ts"; import { safeErrorLogAttributes } from "../errors/safeLog.ts"; import { EnvironmentCacheStore } from "../platform/persistence.ts"; import { subscribe } from "../rpc/client.ts"; +import { ShellSnapshotLoader } from "./shellSnapshotHttp.ts"; import { applyShellStreamEvent } from "./shellReducer.ts"; import type { EnvironmentCatalogState } from "./connections.ts"; import { followStreamInEnvironment } from "./runtime.ts"; @@ -48,6 +49,7 @@ const SHELL_SYNCHRONIZATION_ERROR_MESSAGE = "Could not synchronize environment d export const makeEnvironmentShellState = Effect.fn("EnvironmentShellState.make")(function* () { const supervisor = yield* EnvironmentSupervisor; const cache = yield* EnvironmentCacheStore; + const snapshotLoader = yield* ShellSnapshotLoader; const environmentId = supervisor.target.environmentId; const cachedSnapshot = yield* cache.loadShell(environmentId).pipe( Effect.catch((error) => @@ -147,13 +149,45 @@ export const makeEnvironmentShellState = Effect.fn("EnvironmentShellState.make") yield* Queue.offer(persistence, nextSnapshot); }); - yield* subscribe( - ORCHESTRATION_WS_METHODS.subscribeShell, - {}, - { - onExpectedFailure: (cause) => setStreamError(Cause.squash(cause)), - }, - ).pipe(Stream.runForEach(applyItem), Effect.forkScoped); + yield* Effect.forkScoped( + Effect.gen(function* () { + // Establish the base shell snapshot to resume from, minimizing bytes over + // the wire: + // - Warm cache: reuse the cached snapshot (zero network) and resume via + // `afterSequence` so we only receive shell events since the cached + // sequence. + // - Cold cache: load the full shell snapshot over HTTP (gzip-compressible, + // and off the socket), then resume via `afterSequence`. + // If no base can be established we fall back to the socket-embedded + // snapshot so the shell still synchronizes. Overlapping/replayed events are + // deduped by sequence in applyItem. + const base = Option.isSome(cachedSnapshot) + ? cachedSnapshot + : yield* Effect.gen(function* () { + const prepared = yield* SubscriptionRef.changes(supervisor.prepared).pipe( + Stream.filter(Option.isSome), + Stream.map((current) => current.value), + Stream.runHead, + ); + return Option.isSome(prepared) + ? yield* snapshotLoader.load(prepared.value) + : Option.none(); + }); + + if (Option.isSome(base)) { + yield* applyItem({ kind: "snapshot", snapshot: base.value }); + } + + const subscribeInput = Option.match(base, { + onNone: () => ({}), + onSome: (snapshot) => ({ afterSequence: snapshot.snapshotSequence }), + }); + + yield* subscribe(ORCHESTRATION_WS_METHODS.subscribeShell, subscribeInput, { + onExpectedFailure: (cause) => setStreamError(Cause.squash(cause)), + }).pipe(Stream.runForEach(applyItem)); + }), + ); yield* SubscriptionRef.changes(supervisor.state).pipe( Stream.runForEach((connectionState) => { switch (connectionProjectionPhase(connectionState)) { @@ -293,7 +327,10 @@ export function createEnvironmentServerConfigsAtom(input: { } export function createEnvironmentShellAtoms( - runtime: Atom.AtomRuntime, + runtime: Atom.AtomRuntime< + EnvironmentRegistry | EnvironmentCacheStore | ShellSnapshotLoader | R, + E + >, ) { const stateAtom = Atom.family((environmentId: EnvironmentId) => runtime.atom(shellStateChanges(environmentId), { @@ -316,4 +353,5 @@ export function createEnvironmentShellAtoms( export * from "./models.ts"; export * from "./shellCommands.ts"; export * from "./shellReducer.ts"; +export * from "./shellSnapshotHttp.ts"; export * from "./snapshots.ts"; diff --git a/packages/client-runtime/src/state/shellSnapshotHttp.ts b/packages/client-runtime/src/state/shellSnapshotHttp.ts new file mode 100644 index 00000000000..0a2b873f46a --- /dev/null +++ b/packages/client-runtime/src/state/shellSnapshotHttp.ts @@ -0,0 +1,89 @@ +import type { OrchestrationShellSnapshot } from "@t3tools/contracts"; +import * as Cause from "effect/Cause"; +import * as Context from "effect/Context"; +import * as Effect from "effect/Effect"; +import * as Layer from "effect/Layer"; +import * as Option from "effect/Option"; +import { HttpClient } from "effect/unstable/http"; + +import type { PreparedConnection } from "../connection/model.ts"; +import { environmentEndpointUrl } from "../environment/endpoint.ts"; +import { ManagedRelayDpopSigner } from "../relay/managedRelay.ts"; +import { executeEnvironmentHttpRequest, makeEnvironmentHttpApiClient } from "../rpc/http.ts"; +import { buildEnvironmentAuthHeaders } from "./environmentHttpAuth.ts"; + +// Bounded so a pathologically slow endpoint cannot block the (cheaper) socket +// fallback for long. The cached shell renders while this runs. +const DEFAULT_SHELL_SNAPSHOT_TIMEOUT_MS = 6_000; + +/** + * Load the environment shell snapshot (projects + thread shells) over HTTP + * instead of as the WebSocket subscription's first frame. The response is + * gzip-compressible by the transport and keeps the (potentially large) list off + * the socket. + */ +export const fetchEnvironmentShellSnapshot = Effect.fn( + "clientRuntime.state.fetchEnvironmentShellSnapshot", +)(function* (input: { + readonly prepared: PreparedConnection; + readonly signer: Option.Option; + readonly timeoutMs?: number; +}) { + const requestUrl = environmentEndpointUrl(input.prepared.httpBaseUrl, "/api/orchestration/shell"); + const client = yield* makeEnvironmentHttpApiClient(input.prepared.httpBaseUrl); + const headers = yield* buildEnvironmentAuthHeaders( + input.prepared.httpAuthorization, + "GET", + requestUrl, + input.signer, + ); + return yield* executeEnvironmentHttpRequest( + requestUrl, + input.timeoutMs ?? DEFAULT_SHELL_SNAPSHOT_TIMEOUT_MS, + client.orchestration.shellSnapshot({ headers }), + ); +}); + +/** + * Loads the environment shell snapshot over HTTP, returning `Option.none()` when + * it cannot be loaded (so the caller falls back to the socket-embedded snapshot). + * Decouples the shell state machine from the underlying HTTP + DPoP details and + * keeps them out of test contexts. + */ +export class ShellSnapshotLoader extends Context.Service< + ShellSnapshotLoader, + { + readonly load: ( + prepared: PreparedConnection, + ) => Effect.Effect>; + } +>()("@t3tools/client-runtime/state/shellSnapshotHttp/ShellSnapshotLoader") {} + +export const shellSnapshotLoaderLayer: Layer.Layer< + ShellSnapshotLoader, + never, + HttpClient.HttpClient +> = Layer.effect( + ShellSnapshotLoader, + Effect.gen(function* () { + const httpClient = yield* HttpClient.HttpClient; + // Resolve the DPoP signer optionally: it is only needed for relay/DPoP + // connections, so the loader must not hard-require it. + const signer = yield* Effect.serviceOption(ManagedRelayDpopSigner); + return ShellSnapshotLoader.of({ + load: (prepared: PreparedConnection) => + fetchEnvironmentShellSnapshot({ prepared, signer }).pipe( + Effect.map(Option.some), + Effect.provideService(HttpClient.HttpClient, httpClient), + Effect.catchCause((cause) => + Effect.logWarning( + "Could not load the environment shell snapshot over HTTP; using the socket snapshot instead.", + ).pipe( + Effect.annotateLogs({ cause: Cause.pretty(cause) }), + Effect.as(Option.none()), + ), + ), + ), + }); + }), +); diff --git a/packages/client-runtime/src/state/threadSnapshotHttp.ts b/packages/client-runtime/src/state/threadSnapshotHttp.ts index 34a03ed9455..dad878417c2 100644 --- a/packages/client-runtime/src/state/threadSnapshotHttp.ts +++ b/packages/client-runtime/src/state/threadSnapshotHttp.ts @@ -4,73 +4,23 @@ import * as Context from "effect/Context"; import * as Effect from "effect/Effect"; import * as Layer from "effect/Layer"; import * as Option from "effect/Option"; -import { HttpClient, type HttpMethod } from "effect/unstable/http"; +import { HttpClient } from "effect/unstable/http"; -import type { PreparedConnection, PreparedHttpAuthorization } from "../connection/model.ts"; +import type { PreparedConnection } from "../connection/model.ts"; import { environmentEndpointUrl } from "../environment/endpoint.ts"; import { ManagedRelayDpopSigner } from "../relay/managedRelay.ts"; import { - RemoteEnvironmentAuthFetchError, executeEnvironmentHttpRequest, makeEnvironmentHttpApiClient, type RemoteEnvironmentRequestError, } from "../rpc/http.ts"; +import { buildEnvironmentAuthHeaders } from "./environmentHttpAuth.ts"; // Bounded so a pathologically slow endpoint cannot block the (cheaper) socket // fallback for long. The cached thread renders while this runs, so the wait only // delays the transition to live data on the first open, not the initial paint. const DEFAULT_THREAD_SNAPSHOT_TIMEOUT_MS = 6_000; -interface EnvironmentHttpAuthHeaders { - readonly authorization?: string; - readonly dpop?: string; -} - -/** - * Build the authorization headers for an authenticated environment HTTP - * request, matching the credential the connection was prepared with: - * - primary/local connections carry no credential, - * - bearer connections send a static `Bearer` token, - * - relay connections send a `DPoP` access token with a freshly signed proof - * bound to this request's method and URL. - * - * The DPoP signer is passed in (not resolved from context) and is only required - * for relay/DPoP connections, so bearer/primary connections work even when no - * signer is available. - */ -const buildAuthHeaders = ( - authorization: PreparedHttpAuthorization | null, - method: HttpMethod.HttpMethod, - url: string, - signer: Option.Option, -): Effect.Effect => - Effect.gen(function* () { - if (authorization === null) { - return {}; - } - if (authorization._tag === "Bearer") { - return { authorization: `Bearer ${authorization.token}` }; - } - if (Option.isNone(signer)) { - return yield* new RemoteEnvironmentAuthFetchError({ - message: "No DPoP signer is available to authorize the thread snapshot request.", - cause: authorization._tag, - }); - } - const proof = yield* signer.value - .createProof({ method, url, accessToken: authorization.accessToken }) - .pipe( - Effect.mapError( - (cause) => - new RemoteEnvironmentAuthFetchError({ - message: "Could not create the thread snapshot authorization proof.", - cause, - }), - ), - ); - return { authorization: `DPoP ${authorization.accessToken}`, dpop: proof }; - }); - /** * Load a thread's detail snapshot over HTTP instead of embedding it in the * WebSocket subscription's first frame. The response is gzip-compressible by @@ -89,7 +39,7 @@ export const fetchEnvironmentThreadSnapshot = Effect.fn( `/api/orchestration/threads/${input.threadId}`, ); const client = yield* makeEnvironmentHttpApiClient(input.prepared.httpBaseUrl); - const headers = yield* buildAuthHeaders( + const headers = yield* buildEnvironmentAuthHeaders( input.prepared.httpAuthorization, "GET", requestUrl, diff --git a/packages/client-runtime/src/state/threads-sync.test.ts b/packages/client-runtime/src/state/threads-sync.test.ts index 6552088791f..b4548995d1f 100644 --- a/packages/client-runtime/src/state/threads-sync.test.ts +++ b/packages/client-runtime/src/state/threads-sync.test.ts @@ -42,6 +42,7 @@ const TARGET = new PrimaryConnectionTarget({ wsBaseUrl: "wss://environment.example.test", }); const THREAD_ID = ThreadId.make("thread-1"); +const CACHED_SNAPSHOT_SEQUENCE = 7; const PREPARED: PreparedConnection = { environmentId: TARGET.environmentId, label: TARGET.label, @@ -106,7 +107,9 @@ const makeHarness = Effect.fn("TestEnvironmentThreads.makeHarness")(function* (o const latest = yield* Ref.make(EMPTY_ENVIRONMENT_THREAD_STATE); const retryCount = yield* Ref.make(0); const subscriptionCount = yield* Ref.make(0); - const savedThreads = yield* Ref.make>([]); + const loaderCalls = yield* Ref.make(0); + const lastSubscribeAfterSequence = yield* Ref.make(undefined); + const savedThreads = yield* Ref.make>([]); const removedThreads = yield* Ref.make>([]); const supervisorState = yield* SubscriptionRef.make( AVAILABLE_CONNECTION_STATE, @@ -118,10 +121,11 @@ const makeHarness = Effect.fn("TestEnvironmentThreads.makeHarness")(function* (o ), ); const client = { - [ORCHESTRATION_WS_METHODS.subscribeThread]: () => + [ORCHESTRATION_WS_METHODS.subscribeThread]: (input: { readonly afterSequence?: number }) => Stream.unwrap( Ref.updateAndGet(subscriptionCount, (count) => count + 1).pipe( - Effect.map(() => streamFrom(inputs)), + Effect.andThen(Ref.set(lastSubscribeAfterSequence, input.afterSequence)), + Effect.as(streamFrom(inputs)), ), ), } as unknown as WsRpcProtocolClient; @@ -133,10 +137,12 @@ const makeHarness = Effect.fn("TestEnvironmentThreads.makeHarness")(function* (o ); const snapshotLoader = ThreadSnapshotLoader.of({ load: (_prepared, threadId) => - Effect.succeed( - threadId === THREAD_ID - ? (options?.httpSnapshot ?? Option.none()) - : Option.none(), + Ref.update(loaderCalls, (count) => count + 1).pipe( + Effect.as( + threadId === THREAD_ID + ? (options?.httpSnapshot ?? Option.none()) + : Option.none(), + ), ), }); const supervisor = EnvironmentSupervisor.EnvironmentSupervisor.of({ @@ -154,7 +160,10 @@ const makeHarness = Effect.fn("TestEnvironmentThreads.makeHarness")(function* (o loadThread: (_environmentId, threadId) => Effect.succeed( threadId === THREAD_ID && options?.cached !== undefined - ? Option.some(options.cached) + ? Option.some({ + snapshotSequence: CACHED_SNAPSHOT_SEQUENCE, + thread: options.cached, + }) : Option.none(), ), saveThread: (_environmentId, thread) => @@ -181,6 +190,8 @@ const makeHarness = Effect.fn("TestEnvironmentThreads.makeHarness")(function* (o latest, retryCount, subscriptionCount, + loaderCalls, + lastSubscribeAfterSequence, supervisorState, supervisorSession, savedThreads, @@ -239,19 +250,38 @@ const deleted = (): OrchestrationThreadStreamItem => ({ }); describe("EnvironmentThreads", () => { - it.effect("publishes cached data before a live snapshot arrives", () => + it.effect("publishes cached data immediately from a warm cache", () => Effect.gen(function* () { const harness = yield* makeHarness({ cached: BASE_THREAD }); - const state = yield* awaitThreadState( - harness.observed, - (value) => value.status === "cached" && Option.isSome(value.data), - ); + const state = yield* awaitThreadState(harness.observed, (value) => Option.isSome(value.data)); expect(Option.getOrThrow(state.data)).toEqual(BASE_THREAD); expect(Option.isNone(state.error)).toBe(true); }), ); + it.effect("resumes a warm cache via afterSequence without an HTTP fetch", () => + Effect.gen(function* () { + const harness = yield* makeHarness({ cached: BASE_THREAD }); + + // The warm cache reaches live from the cached data, and a live event + // applies on top of it. + yield* Queue.offer(harness.inputs, titleUpdated("Live title", CACHED_SNAPSHOT_SEQUENCE + 1)); + yield* awaitThreadState( + harness.observed, + (value) => + value.status === "live" && + Option.isSome(value.data) && + value.data.value.title === "Live title", + ); + + // The subscription resumed from the cached sequence and never fetched the + // full snapshot over HTTP. + expect(yield* Ref.get(harness.lastSubscribeAfterSequence)).toBe(CACHED_SNAPSHOT_SEQUENCE); + expect(yield* Ref.get(harness.loaderCalls)).toBe(0); + }), + ); + it.effect("reduces live events and persists the latest thread", () => Effect.gen(function* () { const harness = yield* makeHarness({ cached: BASE_THREAD }); @@ -269,7 +299,8 @@ describe("EnvironmentThreads", () => { yield* Effect.yieldNow; expect(Option.getOrThrow(state.data).title).toBe("Live title"); - expect((yield* Ref.get(harness.savedThreads)).at(-1)?.title).toBe("Live title"); + expect((yield* Ref.get(harness.savedThreads)).at(-1)?.thread.title).toBe("Live title"); + expect((yield* Ref.get(harness.savedThreads)).at(-1)?.snapshotSequence).toBe(2); }), ); @@ -292,6 +323,10 @@ describe("EnvironmentThreads", () => { ); expect(Option.getOrThrow(state.data).title).toBe("Live title"); + // Cold cache: the full snapshot was loaded over HTTP and the socket + // resumed from that snapshot's sequence. + expect(yield* Ref.get(harness.loaderCalls)).toBeGreaterThanOrEqual(1); + expect(yield* Ref.get(harness.lastSubscribeAfterSequence)).toBe(1); }), ); diff --git a/packages/client-runtime/src/state/threads.ts b/packages/client-runtime/src/state/threads.ts index 539ff711a5e..fd5b425fa2a 100644 --- a/packages/client-runtime/src/state/threads.ts +++ b/packages/client-runtime/src/state/threads.ts @@ -56,22 +56,27 @@ export const makeEnvironmentThreadState = Effect.fn("EnvironmentThreadState.make threadId, error: error.message, }), - Effect.as(Option.none()), + Effect.as(Option.none()), ), ), ); + const cachedThread = Option.map(cached, (snapshot) => snapshot.thread); const state = yield* SubscriptionRef.make({ - data: cached, - status: statusWithoutLiveData(cached), + data: cachedThread, + status: statusWithoutLiveData(cachedThread), error: Option.none(), }); - const lastSequence = yield* SubscriptionRef.make(0); - const persistence = yield* Queue.sliding(1); + // Seed the resume cursor from the cached snapshot so a warm cache can catch up + // via `afterSequence` instead of re-downloading the full thread body. + const lastSequence = yield* SubscriptionRef.make( + Option.match(cached, { onNone: () => 0, onSome: (snapshot) => snapshot.snapshotSequence }), + ); + const persistence = yield* Queue.sliding(1); const persist = Effect.fn("EnvironmentThreadState.persist")(function* ( - thread: OrchestrationThread, + snapshot: OrchestrationThreadDetailSnapshot, ) { - yield* cache.saveThread(environmentId, thread).pipe( + yield* cache.saveThread(environmentId, snapshot).pipe( Effect.catch((error) => Effect.logWarning("Could not persist the thread cache.").pipe( Effect.annotateLogs({ @@ -123,7 +128,10 @@ export const makeEnvironmentThreadState = Effect.fn("EnvironmentThreadState.make status: "live", error: Option.none(), }); - yield* Queue.offer(persistence, thread); + // Persist the thread together with the sequence it reflects so the next warm + // cache can resume from exactly here. + const snapshotSequence = yield* SubscriptionRef.get(lastSequence); + yield* Queue.offer(persistence, { snapshotSequence, thread }); }); const setDeleted = Effect.fn("EnvironmentThreadState.setDeleted")(function* () { @@ -192,30 +200,36 @@ export const makeEnvironmentThreadState = Effect.fn("EnvironmentThreadState.make yield* setSynchronizing; yield* Effect.forkScoped( Effect.gen(function* () { - // Load the initial snapshot over HTTP (gzip-compressible, and off the - // socket) rather than as a multi-KB first WebSocket frame, then subscribe - // to live events resuming after the snapshot's sequence. If the snapshot - // cannot be loaded over HTTP we fall back to the socket-embedded snapshot - // so the thread still synchronizes. Overlapping events are deduped by - // sequence in applyItem. - // - // Wait for a prepared connection so we can authenticate the HTTP request; - // this mirrors the socket path, which likewise waits for a live session. - const prepared = yield* SubscriptionRef.changes(supervisor.prepared).pipe( - Stream.filter(Option.isSome), - Stream.map((current) => current.value), - Stream.runHead, - ); - - const httpSnapshot = Option.isSome(prepared) - ? yield* snapshotLoader.load(prepared.value, threadId) - : Option.none(); + // Establish the base snapshot to resume from, minimizing bytes over the + // wire: + // - Warm cache: reuse the cached snapshot (zero network) and resume via + // `afterSequence` so we only receive events since the cached sequence. + // - Cold cache: load the full snapshot over HTTP (gzip-compressible, and + // off the socket), then resume via `afterSequence`. + // If no base can be established we fall back to the socket-embedded + // snapshot so the thread still synchronizes. Overlapping/replayed events + // are deduped by sequence in applyItem. + const base = Option.isSome(cached) + ? cached + : yield* Effect.gen(function* () { + // Cold cache only: wait for a prepared connection so we can + // authenticate the HTTP request; this mirrors the socket path, which + // likewise waits for a live session. + const prepared = yield* SubscriptionRef.changes(supervisor.prepared).pipe( + Stream.filter(Option.isSome), + Stream.map((current) => current.value), + Stream.runHead, + ); + return Option.isSome(prepared) + ? yield* snapshotLoader.load(prepared.value, threadId) + : Option.none(); + }); - if (Option.isSome(httpSnapshot)) { - yield* applyItem({ kind: "snapshot", snapshot: httpSnapshot.value }); + if (Option.isSome(base)) { + yield* applyItem({ kind: "snapshot", snapshot: base.value }); } - const subscribeInput = Option.match(httpSnapshot, { + const subscribeInput = Option.match(base, { onNone: () => ({ threadId }), onSome: (snapshot) => ({ threadId, afterSequence: snapshot.snapshotSequence }), }); @@ -228,11 +242,11 @@ export const makeEnvironmentThreadState = Effect.fn("EnvironmentThreadState.make ); yield* Effect.addFinalizer(() => - SubscriptionRef.get(state).pipe( - Effect.flatMap((current) => + Effect.all([SubscriptionRef.get(state), SubscriptionRef.get(lastSequence)]).pipe( + Effect.flatMap(([current, snapshotSequence]) => Option.match(current.data, { onNone: () => Effect.void, - onSome: persist, + onSome: (thread) => persist({ snapshotSequence, thread }), }), ), ), diff --git a/packages/contracts/src/environmentHttp.ts b/packages/contracts/src/environmentHttp.ts index 353ed91accf..86b9f151f98 100644 --- a/packages/contracts/src/environmentHttp.ts +++ b/packages/contracts/src/environmentHttp.ts @@ -30,6 +30,7 @@ import { ClientOrchestrationCommand, DispatchResult, OrchestrationReadModel, + OrchestrationShellSnapshot, OrchestrationThreadDetailSnapshot, } from "./orchestration.ts"; import { @@ -459,6 +460,13 @@ export class EnvironmentOrchestrationHttpApi extends HttpApiGroup.make("orchestr error: EnvironmentOrchestrationSnapshotErrors, }).middleware(EnvironmentAuthenticatedAuth), ) + .add( + HttpApiEndpoint.get("shellSnapshot", "/api/orchestration/shell", { + headers: OptionalBearerHeaders, + success: OrchestrationShellSnapshot, + error: EnvironmentOrchestrationSnapshotErrors, + }).middleware(EnvironmentAuthenticatedAuth), + ) .add( HttpApiEndpoint.get("threadSnapshot", "/api/orchestration/threads/:threadId", { headers: OptionalBearerHeaders, diff --git a/packages/contracts/src/orchestration.ts b/packages/contracts/src/orchestration.ts index c7c768fabb7..49a626e20af 100644 --- a/packages/contracts/src/orchestration.ts +++ b/packages/contracts/src/orchestration.ts @@ -451,6 +451,19 @@ export const OrchestrationShellStreamItem = Schema.Union([ ]); export type OrchestrationShellStreamItem = typeof OrchestrationShellStreamItem.Type; +export const OrchestrationSubscribeShellInput = Schema.Struct({ + /** + * When provided, the server skips the initial full shell snapshot and instead + * replays shell events after this sequence before streaming live events. + * Clients that already hold a cached (or HTTP-loaded) shell snapshot pass its + * sequence here so the subscription resumes without re-sending the entire + * projects/threads list (overlapping events are deduped by sequence on the + * client). + */ + afterSequence: Schema.optionalKey(NonNegativeInt), +}); +export type OrchestrationSubscribeShellInput = typeof OrchestrationSubscribeShellInput.Type; + export const OrchestrationSubscribeThreadInput = Schema.Struct({ threadId: ThreadId, /** @@ -1252,7 +1265,7 @@ export const OrchestrationRpcSchemas = { output: OrchestrationThreadStreamItem, }, subscribeShell: { - input: Schema.Struct({}), + input: OrchestrationSubscribeShellInput, output: OrchestrationShellStreamItem, }, } as const; From 9254fe5fb93049c65086d0f5c519dce15878a006 Mon Sep 17 00:00:00 2001 From: Julius Marminge Date: Sun, 5 Jul 2026 21:07:03 -0700 Subject: [PATCH 6/6] Send credentials for cookie-authenticated snapshot fetches MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Primary/local environments with no bearer or DPoP credential authenticate the browser via a session cookie, which a cross-origin fetch does not send by default — so the thread/shell snapshot requests 401'd against a primary env on a different origin. Add withEnvironmentCredentials (shared in environmentHttpAuth) which opts those requests into credentials: "include"; bearer/DPoP connections carry their credential in a header and are unaffected. Co-Authored-By: Claude Opus 4.8 --- .../src/state/environmentHttpAuth.ts | 18 +++++++++++++++++- .../src/state/shellSnapshotHttp.ts | 7 +++++-- .../src/state/threadSnapshotHttp.ts | 13 ++++++++----- 3 files changed, 30 insertions(+), 8 deletions(-) diff --git a/packages/client-runtime/src/state/environmentHttpAuth.ts b/packages/client-runtime/src/state/environmentHttpAuth.ts index 8161a69e045..09770806620 100644 --- a/packages/client-runtime/src/state/environmentHttpAuth.ts +++ b/packages/client-runtime/src/state/environmentHttpAuth.ts @@ -1,6 +1,6 @@ import * as Effect from "effect/Effect"; import * as Option from "effect/Option"; -import type { HttpMethod } from "effect/unstable/http"; +import { FetchHttpClient, type HttpMethod } from "effect/unstable/http"; import type { PreparedHttpAuthorization } from "../connection/model.ts"; import type { ManagedRelayDpopSigner } from "../relay/managedRelay.ts"; @@ -11,6 +11,22 @@ export interface EnvironmentHttpAuthHeaders { readonly dpop?: string; } +/** + * Primary/local environments with no bearer or DPoP credential authenticate the + * browser via a session cookie. A cross-origin `fetch` does not send cookies by + * default, so those requests must opt into credentialed mode; bearer/DPoP + * connections carry their credential in a header and need no cookies. Applied + * per-request via `FetchHttpClient.RequestInit`, which the fetch client reads + * from the fiber context at request time. + */ +export const withEnvironmentCredentials = ( + authorization: PreparedHttpAuthorization | null, + request: Effect.Effect, +): Effect.Effect => + authorization === null + ? request.pipe(Effect.provideService(FetchHttpClient.RequestInit, { credentials: "include" })) + : request; + /** * Build the authorization headers for an authenticated environment HTTP * request, matching the credential the connection was prepared with: diff --git a/packages/client-runtime/src/state/shellSnapshotHttp.ts b/packages/client-runtime/src/state/shellSnapshotHttp.ts index 0a2b873f46a..b0a492a1305 100644 --- a/packages/client-runtime/src/state/shellSnapshotHttp.ts +++ b/packages/client-runtime/src/state/shellSnapshotHttp.ts @@ -10,7 +10,7 @@ import type { PreparedConnection } from "../connection/model.ts"; import { environmentEndpointUrl } from "../environment/endpoint.ts"; import { ManagedRelayDpopSigner } from "../relay/managedRelay.ts"; import { executeEnvironmentHttpRequest, makeEnvironmentHttpApiClient } from "../rpc/http.ts"; -import { buildEnvironmentAuthHeaders } from "./environmentHttpAuth.ts"; +import { buildEnvironmentAuthHeaders, withEnvironmentCredentials } from "./environmentHttpAuth.ts"; // Bounded so a pathologically slow endpoint cannot block the (cheaper) socket // fallback for long. The cached shell renders while this runs. @@ -40,7 +40,10 @@ export const fetchEnvironmentShellSnapshot = Effect.fn( return yield* executeEnvironmentHttpRequest( requestUrl, input.timeoutMs ?? DEFAULT_SHELL_SNAPSHOT_TIMEOUT_MS, - client.orchestration.shellSnapshot({ headers }), + withEnvironmentCredentials( + input.prepared.httpAuthorization, + client.orchestration.shellSnapshot({ headers }), + ), ); }); diff --git a/packages/client-runtime/src/state/threadSnapshotHttp.ts b/packages/client-runtime/src/state/threadSnapshotHttp.ts index dad878417c2..874bcc30ebd 100644 --- a/packages/client-runtime/src/state/threadSnapshotHttp.ts +++ b/packages/client-runtime/src/state/threadSnapshotHttp.ts @@ -14,7 +14,7 @@ import { makeEnvironmentHttpApiClient, type RemoteEnvironmentRequestError, } from "../rpc/http.ts"; -import { buildEnvironmentAuthHeaders } from "./environmentHttpAuth.ts"; +import { buildEnvironmentAuthHeaders, withEnvironmentCredentials } from "./environmentHttpAuth.ts"; // Bounded so a pathologically slow endpoint cannot block the (cheaper) socket // fallback for long. The cached thread renders while this runs, so the wait only @@ -48,10 +48,13 @@ export const fetchEnvironmentThreadSnapshot = Effect.fn( return yield* executeEnvironmentHttpRequest( requestUrl, input.timeoutMs ?? DEFAULT_THREAD_SNAPSHOT_TIMEOUT_MS, - client.orchestration.threadSnapshot({ - params: { threadId: input.threadId }, - headers, - }), + withEnvironmentCredentials( + input.prepared.httpAuthorization, + client.orchestration.threadSnapshot({ + params: { threadId: input.threadId }, + headers, + }), + ), ); });