diff --git a/apps/mobile/src/connection/runtime.ts b/apps/mobile/src/connection/runtime.ts index 3698a0a5fc7..3d4bf4944f1 100644 --- a/apps/mobile/src/connection/runtime.ts +++ b/apps/mobile/src/connection/runtime.ts @@ -1,4 +1,6 @@ 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"; @@ -9,12 +11,15 @@ const providedConnectionPlatformLayer = connectionPlatformLayer.pipe( Layer.provide(runtimeContextLayer), ); +const snapshotLoaderLayer = Layer.merge(threadSnapshotLoaderLayer, shellSnapshotLoaderLayer); + type ConnectionLayerSource = | typeof Connection.layer + | typeof snapshotLoaderLayer | typeof runtimeContextLayer | typeof connectionPlatformLayer; -const connectionLayer = Connection.layer.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/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/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 a148f98474b..016c3d508ec 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,38 @@ 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) { + yield* annotateEnvironmentRequest(args.endpoint.name); + yield* requireEnvironmentScope(AuthOrchestrationReadScope); + 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 snapshot.value; + }), + ) .handle( "dispatch", Effect.fn("environment.orchestration.dispatch")(function* (args) { 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 9020e99f670..fafdbfa6814 100644 --- a/apps/server/src/ws.ts +++ b/apps/server/src/ws.ts @@ -30,6 +30,8 @@ import { OrchestrationDispatchCommandError, type OrchestrationEvent, type OrchestrationShellStreamEvent, + type OrchestrationShellStreamItem, + type OrchestrationThreadStreamItem, OrchestrationGetFullThreadDiffError, OrchestrationGetSnapshotError, OrchestrationGetTurnDiffError, @@ -1059,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 }), @@ -1076,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, @@ -1114,8 +1153,65 @@ const makeWsRpcLayer = ( observeRpcStreamEffect( ORCHESTRATION_WS_METHODS.subscribeThread, Effect.gen(function* () { - const [threadDetail, snapshotSequence] = yield* Effect.all([ - projectionSnapshotQuery.getThreadDetailById(input.threadId).pipe( + 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 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 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)); + }), + ); + } + + const snapshot = yield* projectionSnapshotQuery + .getThreadDetailSnapshot(input.threadId) + .pipe( Effect.mapError( (cause) => new OrchestrationGetSnapshotError({ @@ -1123,46 +1219,19 @@ 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, }); } - 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, - snapshot: { - snapshotSequence, - thread: threadDetail.value, - }, + snapshot: snapshot.value, }), liveStream, ); diff --git a/apps/web/src/connection/runtime.ts b/apps/web/src/connection/runtime.ts index 3698a0a5fc7..3d4bf4944f1 100644 --- a/apps/web/src/connection/runtime.ts +++ b/apps/web/src/connection/runtime.ts @@ -1,4 +1,6 @@ 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"; @@ -9,12 +11,15 @@ const providedConnectionPlatformLayer = connectionPlatformLayer.pipe( Layer.provide(runtimeContextLayer), ); +const snapshotLoaderLayer = Layer.merge(threadSnapshotLoaderLayer, shellSnapshotLoaderLayer); + type ConnectionLayerSource = | typeof Connection.layer + | typeof snapshotLoaderLayer | typeof runtimeContextLayer | typeof connectionPlatformLayer; -const connectionLayer = Connection.layer.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/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/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/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/environmentHttpAuth.ts b/packages/client-runtime/src/state/environmentHttpAuth.ts new file mode 100644 index 00000000000..09770806620 --- /dev/null +++ b/packages/client-runtime/src/state/environmentHttpAuth.ts @@ -0,0 +1,73 @@ +import * as Effect from "effect/Effect"; +import * as Option from "effect/Option"; +import { FetchHttpClient, 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; +} + +/** + * 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: + * - 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..b0a492a1305 --- /dev/null +++ b/packages/client-runtime/src/state/shellSnapshotHttp.ts @@ -0,0 +1,92 @@ +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, 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. +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, + withEnvironmentCredentials( + input.prepared.httpAuthorization, + 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 new file mode 100644 index 00000000000..874bcc30ebd --- /dev/null +++ b/packages/client-runtime/src/state/threadSnapshotHttp.ts @@ -0,0 +1,120 @@ +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 } 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, + type RemoteEnvironmentRequestError, +} from "../rpc/http.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 +// delays the transition to live data on the first open, not the initial paint. +const DEFAULT_THREAD_SNAPSHOT_TIMEOUT_MS = 6_000; + +/** + * 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 signer: Option.Option; + readonly timeoutMs?: number; +}) { + const requestUrl = environmentEndpointUrl( + input.prepared.httpBaseUrl, + `/api/orchestration/threads/${input.threadId}`, + ); + 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_THREAD_SNAPSHOT_TIMEOUT_MS, + withEnvironmentCredentials( + input.prepared.httpAuthorization, + 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 +> = Layer.effect( + ThreadSnapshotLoader, + 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 (bearer/primary + // connections work without one). + const signer = yield* Effect.serviceOption(ManagedRelayDpopSigner); + return ThreadSnapshotLoader.of({ + load: (prepared: PreparedConnection, threadId: ThreadId) => + fetchEnvironmentThreadSnapshot({ prepared, threadId, signer }).pipe( + Effect.map(Option.some), + Effect.provideService(HttpClient.HttpClient, httpClient), + // 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.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.", + ).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..b4548995d1f 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,15 @@ 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, + httpBaseUrl: TARGET.httpBaseUrl, + socketUrl: TARGET.wsBaseUrl, + httpAuthorization: null, + target: TARGET, +}; const BASE_THREAD: OrchestrationThread = { id: THREAD_ID, projectId: ProjectId.make("project-1"), @@ -89,13 +100,16 @@ 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(); 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, @@ -107,17 +121,30 @@ 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; 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) => + Ref.update(loaderCalls, (count) => count + 1).pipe( + Effect.as( + threadId === THREAD_ID + ? (options?.httpSnapshot ?? Option.none()) + : Option.none(), + ), + ), + }); const supervisor = EnvironmentSupervisor.EnvironmentSupervisor.of({ target: TARGET, state: supervisorState, @@ -133,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) => @@ -145,6 +175,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) => @@ -159,6 +190,8 @@ const makeHarness = Effect.fn("TestEnvironmentThreads.makeHarness")(function* (o latest, retryCount, subscriptionCount, + loaderCalls, + lastSubscribeAfterSequence, supervisorState, supervisorSession, savedThreads, @@ -217,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 }); @@ -247,7 +299,34 @@ 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); + }), + ); + + 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"); + // 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 a01baf1594d..fd5b425fa2a 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) => @@ -53,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({ @@ -120,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* () { @@ -187,21 +198,55 @@ 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* () { + // 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(base)) { + yield* applyItem({ kind: "snapshot", snapshot: base.value }); + } + + const subscribeInput = Option.match(base, { + 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( - 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 }), }), ), ), @@ -218,7 +263,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 +288,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..86b9f151f98 100644 --- a/packages/contracts/src/environmentHttp.ts +++ b/packages/contracts/src/environmentHttp.ts @@ -24,12 +24,14 @@ 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, + OrchestrationShellSnapshot, + OrchestrationThreadDetailSnapshot, } from "./orchestration.ts"; import { RelayCloudEnvironmentHealthRequest, @@ -80,6 +82,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 +158,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 +290,11 @@ const EnvironmentOrchestrationSnapshotErrors = [ EnvironmentScopeRequiredError, EnvironmentInternalError, ] as const; +const EnvironmentOrchestrationThreadSnapshotErrors = [ + EnvironmentScopeRequiredError, + EnvironmentResourceNotFoundError, + EnvironmentInternalError, +] as const; const EnvironmentOrchestrationDispatchErrors = [ EnvironmentRequestInvalidError, EnvironmentScopeRequiredError, @@ -422,6 +448,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 +460,21 @@ 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, + 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..49a626e20af 100644 --- a/packages/contracts/src/orchestration.ts +++ b/packages/contracts/src/orchestration.ts @@ -451,8 +451,29 @@ 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, + /** + * 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; @@ -1244,7 +1265,7 @@ export const OrchestrationRpcSchemas = { output: OrchestrationThreadStreamItem, }, subscribeShell: { - input: Schema.Struct({}), + input: OrchestrationSubscribeShellInput, output: OrchestrationShellStreamItem, }, } as const;