diff --git a/FORK.md b/FORK.md index 0ea269c1872..92374eff15a 100644 --- a/FORK.md +++ b/FORK.md @@ -10,6 +10,19 @@ This repository is a fork of `pingdotgg/t3code`. Keep this file focused on fork - Exception: upstream actualization PRs may preserve upstream commit structure when that makes future syncs easier to audit. - Staged formatting tolerates chunks containing only ignored files so large upstream actualization commits can pass the pre-commit hook. +### Protocol Compatibility + +- Fork protocol extensions use separate extension RPC methods and additive optional capabilities. +- Existing upstream RPC methods keep their upstream request, response, and behavior contracts so upstream clients remain compatible with the fork backend. +- Fork clients call a fork RPC only when the backend advertises its capability and otherwise retain the upstream RPC path. +- Fork backends keep the upstream RPC handler alongside each fork extension. + +### Thread Delta Subscription + +- Fork backends advertise `threadDeltaSubscription` and expose `orchestration.subscribeThread.withDelta` alongside the upstream thread subscription. +- Fork clients use the fork subscription only when advertised. Upstream backends therefore continue receiving the upstream subscription. +- The fork subscription replays gaps of at most 1,000 orchestration events and sends a fresh thread snapshot for larger or invalid gaps. If the thread no longer exists, it sends `not-found` so clients clear stale cached state. + ### Release And CI - Fork workflows create/update a daily stable release PR while main-branch pushes produce nightly releases. diff --git a/apps/server/src/environment/ServerEnvironment.ts b/apps/server/src/environment/ServerEnvironment.ts index 0eaf5a7c16a..96e61619446 100644 --- a/apps/server/src/environment/ServerEnvironment.ts +++ b/apps/server/src/environment/ServerEnvironment.ts @@ -142,6 +142,7 @@ export const make = Effect.gen(function* () { connectionProbe: true, threadSettlement: true, threadSnooze: true, + threadDeltaSubscription: true, ...(serverSelfUpdate === null ? {} : { serverSelfUpdate }), }, }; diff --git a/apps/server/src/ws.ts b/apps/server/src/ws.ts index 744e48661bf..04e45dd8f68 100644 --- a/apps/server/src/ws.ts +++ b/apps/server/src/ws.ts @@ -31,6 +31,7 @@ import { type OrchestrationEvent, type OrchestrationShellStreamEvent, type OrchestrationShellStreamItem, + type OrchestrationSubscribeThreadInput, type OrchestrationThreadStreamItem, OrchestrationGetFullThreadDiffError, OrchestrationGetSnapshotError, @@ -294,6 +295,7 @@ const PROVIDER_STATUS_DEBOUNCE_MS = 200; // past this gap a single O(active-threads) snapshot is cheaper and bounded. // Matches the event store's default page size (DEFAULT_READ_FROM_SEQUENCE_LIMIT). const SHELL_RESUME_MAX_GAP = 1_000; +const THREAD_DELTA_SUBSCRIPTION_MAX_GAP = 1_000; const RPC_REQUIRED_SCOPE = new Map([ [ORCHESTRATION_WS_METHODS.dispatchCommand, AuthOrchestrationOperateScope], @@ -302,6 +304,7 @@ const RPC_REQUIRED_SCOPE = new Map([ [ORCHESTRATION_WS_METHODS.subscribeShell, AuthOrchestrationReadScope], [ORCHESTRATION_WS_METHODS.getArchivedShellSnapshot, AuthOrchestrationReadScope], [ORCHESTRATION_WS_METHODS.subscribeThread, AuthOrchestrationReadScope], + [ORCHESTRATION_WS_METHODS.subscribeThreadWithDelta, AuthOrchestrationReadScope], [WS_METHODS.serverProbe, AuthOrchestrationReadScope], [WS_METHODS.serverGetConfig, AuthOrchestrationReadScope], [WS_METHODS.serverRefreshProviders, AuthOrchestrationOperateScope], @@ -1026,6 +1029,122 @@ const makeWsRpcLayer = ( ); }; + const loadThreadSnapshot = Effect.fn("Ws.loadThreadSnapshot")(function* (threadId: ThreadId) { + const snapshot = yield* projectionSnapshotQuery.getThreadDetailSnapshot(threadId).pipe( + Effect.mapError( + (cause) => + new OrchestrationGetSnapshotError({ + message: `Failed to load thread ${threadId}`, + cause, + }), + ), + ); + + return Option.map(snapshot, projectThreadDetailSnapshot); + }); + + const makeThreadSubscription = Effect.fn("Ws.makeThreadSubscription")(function* (input: { + readonly request: OrchestrationSubscribeThreadInput; + readonly maxReplayGap: number | null; + }) { + const isThisThreadDetailEvent = (event: OrchestrationEvent) => + event.aggregateKind === "thread" && + event.aggregateId === input.request.threadId && + isThreadDetailEvent(event); + + const liveStream = orchestrationEngine.streamDomainEvents.pipe( + Stream.filter(isThisThreadDetailEvent), + Stream.map((event) => ({ + kind: "event" as const, + event: projectActivityEvent(event), + })), + ); + + // Attach live delivery before reading either replay or snapshot state. + // Otherwise an event published while the snapshot is loading is lost. + const liveBuffer = yield* Queue.unbounded(); + yield* Effect.forkScoped( + liveStream.pipe(Stream.runForEach((item) => Queue.offer(liveBuffer, item))), + ); + const bufferedLiveStream = Stream.fromQueue(liveBuffer); + const synchronizedThenLive = + input.request.requestCompletionMarker === true + ? Stream.concat( + Stream.fromEffect(Queue.offer(liveBuffer, { kind: "synchronized" as const })).pipe( + Stream.drain, + ), + bufferedLiveStream, + ) + : bufferedLiveStream; + + if (input.request.afterSequence !== undefined) { + const afterSequence = input.request.afterSequence; + if (input.maxReplayGap !== null) { + const headSequence = yield* orchestrationEngine.latestSequence; + const replayGap = headSequence - afterSequence; + if (replayGap < 0 || replayGap > input.maxReplayGap) { + const snapshot = yield* loadThreadSnapshot(input.request.threadId); + if (Option.isNone(snapshot)) { + return Stream.concat( + Stream.make({ kind: "not-found" as const }), + synchronizedThenLive, + ); + } + return Stream.concat( + Stream.make({ kind: "snapshot" as const, snapshot: snapshot.value }), + synchronizedThenLive, + ); + } + + const catchUpStream = orchestrationEngine.readEvents(afterSequence, replayGap).pipe( + Stream.filter(isThisThreadDetailEvent), + Stream.map((event) => ({ + kind: "event" as const, + event: projectActivityEvent(event), + })), + Stream.mapError( + (cause) => + new OrchestrationGetSnapshotError({ + message: `Failed to replay thread ${input.request.threadId} events`, + cause, + }), + ), + ); + return Stream.concat(catchUpStream, synchronizedThenLive); + } + + const catchUpStream = orchestrationEngine + .readEvents(afterSequence, Number.MAX_SAFE_INTEGER) + .pipe( + Stream.filter(isThisThreadDetailEvent), + Stream.map((event) => ({ + kind: "event" as const, + event: projectActivityEvent(event), + })), + Stream.mapError( + (cause) => + new OrchestrationGetSnapshotError({ + message: `Failed to replay thread ${input.request.threadId} events`, + cause, + }), + ), + ); + return Stream.concat(catchUpStream, synchronizedThenLive); + } + + const snapshot = yield* loadThreadSnapshot(input.request.threadId); + if (Option.isNone(snapshot)) { + return yield* new OrchestrationGetSnapshotError({ + message: `Thread ${input.request.threadId} was not found`, + cause: input.request.threadId, + }); + } + return Stream.concat( + Stream.make({ kind: "snapshot" as const, snapshot: snapshot.value }), + synchronizedThenLive, + ); + }); + const loadServerConfig = Effect.gen(function* () { const keybindingsConfig = yield* keybindings.loadConfigState; const providers = yield* providerRegistry.getProviders; @@ -1289,110 +1408,22 @@ const makeWsRpcLayer = ( [ORCHESTRATION_WS_METHODS.subscribeThread]: (input) => 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: projectActivityEvent(event), - })), - ); - - // Attach live delivery before reading either replay or snapshot state. - // Otherwise an event published while the snapshot is loading is lost. - const liveBuffer = yield* Queue.unbounded(); - yield* Effect.forkScoped( - liveStream.pipe(Stream.runForEach((item) => Queue.offer(liveBuffer, item))), - ); - const bufferedLiveStream = Stream.fromQueue(liveBuffer); - - // 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; - const catchUpStream = orchestrationEngine - .readEvents(afterSequence, Number.MAX_SAFE_INTEGER) - .pipe( - Stream.filter(isThisThreadDetailEvent), - Stream.map((event) => ({ - kind: "event" as const, - event: projectActivityEvent(event), - })), - Stream.mapError( - (cause) => - new OrchestrationGetSnapshotError({ - message: `Failed to replay thread ${input.threadId} events`, - cause, - }), - ), - ); - const afterCatchUp = - input.requestCompletionMarker === true - ? Stream.concat( - Stream.fromEffect( - Queue.offer(liveBuffer, { kind: "synchronized" as const }), - ).pipe(Stream.drain), - bufferedLiveStream, - ) - : bufferedLiveStream; - return Stream.concat(catchUpStream, afterCatchUp); - } - - const snapshot = yield* projectionSnapshotQuery - .getThreadDetailSnapshot(input.threadId) - .pipe( - Effect.mapError( - (cause) => - new OrchestrationGetSnapshotError({ - message: `Failed to load thread ${input.threadId}`, - cause, - }), - ), - ); - - if (Option.isNone(snapshot)) { - return yield* new OrchestrationGetSnapshotError({ - message: `Thread ${input.threadId} was not found`, - cause: input.threadId, - }); - } - - const afterSnapshot = - input.requestCompletionMarker === true - ? Stream.concat( - Stream.fromEffect( - Queue.offer(liveBuffer, { kind: "synchronized" as const }), - ).pipe(Stream.drain), - bufferedLiveStream, - ) - : bufferedLiveStream; - return Stream.concat( - Stream.make({ - kind: "snapshot" as const, - snapshot: projectThreadDetailSnapshot(snapshot.value), - }), - afterSnapshot, - ); + makeThreadSubscription({ request: input, maxReplayGap: null }).pipe( + Effect.map((stream) => + Stream.filter( + stream, + (item): item is OrchestrationThreadStreamItem => item.kind !== "not-found", + ), + ), + ), + { "rpc.aggregate": "orchestration" }, + ), + [ORCHESTRATION_WS_METHODS.subscribeThreadWithDelta]: (input) => + observeRpcStreamEffect( + ORCHESTRATION_WS_METHODS.subscribeThreadWithDelta, + makeThreadSubscription({ + request: input, + maxReplayGap: THREAD_DELTA_SUBSCRIPTION_MAX_GAP, }), { "rpc.aggregate": "orchestration" }, ), diff --git a/packages/client-runtime/src/rpc/client.ts b/packages/client-runtime/src/rpc/client.ts index c7c928b3c95..f06d8dd1cd8 100644 --- a/packages/client-runtime/src/rpc/client.ts +++ b/packages/client-runtime/src/rpc/client.ts @@ -42,6 +42,7 @@ type RpcMethod = WsRpcProtocolClient[TTag]; export type EnvironmentSubscriptionRpcTag = | typeof ORCHESTRATION_WS_METHODS.subscribeShell | typeof ORCHESTRATION_WS_METHODS.subscribeThread + | typeof ORCHESTRATION_WS_METHODS.subscribeThreadWithDelta | typeof WS_METHODS.subscribeAuthAccess | typeof WS_METHODS.subscribeServerConfig | typeof WS_METHODS.subscribeServerLifecycle @@ -156,9 +157,13 @@ interface SubscriptionOptions { readonly resubscribe?: Stream.Stream; } -export function subscribeDynamic( - tag: TTag, - makeInput: (session: RpcSession) => Effect.Effect>, +export interface EnvironmentSubscriptionRequest { + readonly tag: TTag; + readonly input: EnvironmentRpcInput; +} + +export function subscribeDynamicRequest( + makeRequest: (session: RpcSession) => Effect.Effect>, options?: SubscriptionOptions, ): Stream.Stream< EnvironmentRpcStreamValue, @@ -183,21 +188,21 @@ export function subscribeDynamic( Option.match({ onNone: () => Stream.empty, onSome: (session) => { - const method = session.client[tag] as ( - input: EnvironmentRpcInput, - ) => Stream.Stream< - EnvironmentRpcStreamValue, - EnvironmentRpcStreamFailure - >; const subscribeToSession = (): Stream.Stream< EnvironmentRpcStreamValue, EnvironmentRpcStreamFailure > => Stream.suspend(() => Stream.unwrap( - makeInput(session).pipe( - Effect.map((input) => - method(input).pipe( + makeRequest(session).pipe( + Effect.map((request) => { + const method = session.client[request.tag] as ( + input: EnvironmentRpcInput, + ) => Stream.Stream< + EnvironmentRpcStreamValue, + EnvironmentRpcStreamFailure + >; + return method(request.input).pipe( Stream.catchCause((cause) => { const hasOnlyExpectedFailures = cause.reasons.length > 0 && @@ -214,7 +219,7 @@ export function subscribeDynamic( "Durable RPC subscription lost its transport; waiting for the next session.", { cause: Cause.pretty(cause), - method: tag, + method: request.tag, environmentId: supervisor.target.environmentId, }, ), @@ -241,8 +246,11 @@ export function subscribeDynamic( } return Stream.failCause(cause); }), - ), - ), + Stream.withSpan("EnvironmentRpc.subscribe", { + attributes: { "rpc.method": request.tag }, + }), + ); + }), ), ), ); @@ -253,10 +261,27 @@ export function subscribeDynamic( ); }), ), - ).pipe( - Stream.withSpan("EnvironmentRpc.subscribe", { - attributes: { "rpc.method": tag }, - }), + ); +} + +export function subscribeDynamic( + tag: TTag, + makeInput: (session: RpcSession) => Effect.Effect>, + options?: SubscriptionOptions, +): Stream.Stream< + EnvironmentRpcStreamValue, + EnvironmentRpcStreamFailure, + EnvironmentSupervisor +> { + return subscribeDynamicRequest( + (session) => + makeInput(session).pipe( + Effect.map((input) => ({ + tag, + input, + })), + ), + options, ); } diff --git a/packages/client-runtime/src/state/threads.ts b/packages/client-runtime/src/state/threads.ts index 196229cc8b1..af2efb31fb4 100644 --- a/packages/client-runtime/src/state/threads.ts +++ b/packages/client-runtime/src/state/threads.ts @@ -2,8 +2,8 @@ import { ORCHESTRATION_WS_METHODS, type EnvironmentId as EnvironmentIdType, type OrchestrationThread, + type OrchestrationThreadDeltaStreamItem, type OrchestrationThreadDetailSnapshot, - type OrchestrationThreadStreamItem, type ThreadId as ThreadIdType, } from "@t3tools/contracts"; import * as Cause from "effect/Cause"; @@ -20,7 +20,7 @@ import { connectionProjectionPhase } from "../connection/model.ts"; import { EnvironmentSupervisor } from "../connection/supervisor.ts"; import * as ConnectionWakeups from "../connection/wakeups.ts"; import { EnvironmentCacheStore } from "../platform/persistence.ts"; -import { subscribeDynamic } from "../rpc/client.ts"; +import { subscribeDynamicRequest } from "../rpc/client.ts"; import { ThreadSnapshotLoader } from "./threadSnapshotHttp.ts"; import { parseThreadKey, threadKey } from "./entities.ts"; import { applyThreadDetailEvent } from "./threadReducer.ts"; @@ -180,8 +180,13 @@ export const makeEnvironmentThreadState = Effect.fn("EnvironmentThreadState.make }); const applyItem = Effect.fn("EnvironmentThreadState.applyItem")(function* ( - item: OrchestrationThreadStreamItem, + item: OrchestrationThreadDeltaStreamItem, ) { + if (item.kind === "not-found") { + yield* setDeleted(); + return; + } + if (item.kind === "synchronized") { yield* Ref.set(awaitingCompletion, false); yield* SubscriptionRef.update(state, (current) => @@ -241,13 +246,20 @@ export const makeEnvironmentThreadState = Effect.fn("EnvironmentThreadState.make yield* setSynchronizing; yield* Effect.forkScoped( - subscribeDynamic( - ORCHESTRATION_WS_METHODS.subscribeThread, - Effect.fn("EnvironmentThreadState.makeSubscribeInput")(function* (session) { - const supportsCompletionMarker = yield* session.initialConfig.pipe( - Effect.map((config) => config.threadResumeCompletionMarker === true), - Effect.orElseSucceed(() => false), + subscribeDynamicRequest( + Effect.fn("EnvironmentThreadState.makeSubscriptionRequest")(function* (session) { + const subscriptionCapabilities = yield* session.initialConfig.pipe( + Effect.map((config) => ({ + completionMarker: config.threadResumeCompletionMarker === true, + threadDeltaSubscription: + config.environment?.capabilities.threadDeltaSubscription === true, + })), + Effect.orElseSucceed(() => ({ + completionMarker: false, + threadDeltaSubscription: false, + })), ); + const supportsCompletionMarker = subscriptionCapabilities.completionMarker; yield* Ref.set(awaitingCompletion, supportsCompletionMarker); yield* setSynchronizing; @@ -285,9 +297,14 @@ export const makeEnvironmentThreadState = Effect.fn("EnvironmentThreadState.make } return { - threadId, - ...(canResume ? { afterSequence: sequence } : {}), - ...(supportsCompletionMarker ? { requestCompletionMarker: true as const } : {}), + tag: subscriptionCapabilities.threadDeltaSubscription + ? ORCHESTRATION_WS_METHODS.subscribeThreadWithDelta + : ORCHESTRATION_WS_METHODS.subscribeThread, + input: { + threadId, + ...(canResume ? { afterSequence: sequence } : {}), + ...(supportsCompletionMarker ? { requestCompletionMarker: true as const } : {}), + }, }; }), { diff --git a/packages/contracts/src/environment.ts b/packages/contracts/src/environment.ts index 7f4b6c16541..96744f85907 100644 --- a/packages/contracts/src/environment.ts +++ b/packages/contracts/src/environment.ts @@ -47,6 +47,9 @@ export const ExecutionEnvironmentCapabilities = Schema.Struct({ /** Server understands thread.snooze / thread.unsnooze commands. Same version-skew contract as threadSettlement. */ threadSnooze: Schema.optionalKey(Schema.Boolean), + /** Fork server exposes the bounded thread delta subscription. Missing on + upstream servers, so fork clients retain the upstream subscription. */ + threadDeltaSubscription: Schema.optionalKey(Schema.Boolean), /** The update path clients should offer for this server. Absent on servers that must be relaunched manually (dev checkouts, Windows foreground runs, pre-update servers). */ diff --git a/packages/contracts/src/orchestration.ts b/packages/contracts/src/orchestration.ts index 104bfd65786..413381a6bc5 100644 --- a/packages/contracts/src/orchestration.ts +++ b/packages/contracts/src/orchestration.ts @@ -29,6 +29,7 @@ export const ORCHESTRATION_WS_METHODS = { getArchivedShellSnapshot: "orchestration.getArchivedShellSnapshot", subscribeShell: "orchestration.subscribeShell", subscribeThread: "orchestration.subscribeThread", + subscribeThreadWithDelta: "orchestration.subscribeThread.withDelta", } as const; export const ProviderApprovalPolicy = Schema.Literals([ @@ -1378,6 +1379,14 @@ export const OrchestrationThreadStreamItem = Schema.Union([ ]); export type OrchestrationThreadStreamItem = typeof OrchestrationThreadStreamItem.Type; +export const OrchestrationThreadDeltaStreamItem = Schema.Union([ + OrchestrationThreadStreamItem, + Schema.Struct({ + kind: Schema.Literal("not-found"), + }), +]); +export type OrchestrationThreadDeltaStreamItem = typeof OrchestrationThreadDeltaStreamItem.Type; + export const OrchestrationCommandReceiptStatus = Schema.Literals(["accepted", "rejected"]); export type OrchestrationCommandReceiptStatus = typeof OrchestrationCommandReceiptStatus.Type; diff --git a/packages/contracts/src/rpc.ts b/packages/contracts/src/rpc.ts index 43fc2d1acb0..424f75808c9 100644 --- a/packages/contracts/src/rpc.ts +++ b/packages/contracts/src/rpc.ts @@ -55,6 +55,7 @@ import { OrchestrationGetTurnDiffError, OrchestrationGetTurnDiffInput, OrchestrationRpcSchemas, + OrchestrationThreadDeltaStreamItem, } from "./orchestration.ts"; import { ProviderInstanceId } from "./providerInstance.ts"; import { @@ -655,6 +656,16 @@ export const WsOrchestrationSubscribeThreadRpc = Rpc.make( }, ); +export const WsOrchestrationSubscribeThreadWithDeltaRpc = Rpc.make( + ORCHESTRATION_WS_METHODS.subscribeThreadWithDelta, + { + payload: OrchestrationRpcSchemas.subscribeThread.input, + success: OrchestrationThreadDeltaStreamItem, + error: Schema.Union([OrchestrationGetSnapshotError, EnvironmentAuthorizationError]), + stream: true, + }, +); + export const WsSubscribeTerminalEventsRpc = Rpc.make(WS_METHODS.subscribeTerminalEvents, { payload: Schema.Struct({}), success: TerminalEvent, @@ -760,4 +771,5 @@ export const WsRpcGroup = RpcGroup.make( WsOrchestrationGetArchivedShellSnapshotRpc, WsOrchestrationSubscribeShellRpc, WsOrchestrationSubscribeThreadRpc, + WsOrchestrationSubscribeThreadWithDeltaRpc, );