History

feat(orchestration): bound thread history and resume payloads

- Add paginated thread history with bounded snapshots and replay limits - Trim oversized wire payloads and support progressive mobile history loading

Julius Marminge committed 3 months ago

Commit 7d42d311 parent e3883a943 files · +4239 −211Browse files

Showing the changes to packages/client-runtime/src/state/boundedThreadSnapshotHttp.ts (1 of 43 files). Show all

Unified · Split

packages/client-runtime/src/state/boundedThreadSnapshotHttp.ts+156−0
@@ -0,0 +1,156 @@
1+import type { ThreadId } from "@t3tools/contracts";
2+import * as Cause from "effect/Cause";
3+import * as Effect from "effect/Effect";
4+import * as Layer from "effect/Layer";
5+import * as Option from "effect/Option";
6+import { HttpClient } from "effect/unstable/http";
7+
8+import type { PreparedConnection } from "../connection/model.ts";
9+import { environmentEndpointUrl } from "../environment/endpoint.ts";
10+import { ManagedRelayDpopSigner } from "../relay/managedRelay.ts";
11+import { executeEnvironmentHttpRequest, makeEnvironmentHttpApiClient } from "../rpc/http.ts";
12+import { buildEnvironmentAuthHeaders, withEnvironmentCredentials } from "./environmentHttpAuth.ts";
13+import {
14+ fetchEnvironmentThreadSnapshot,
15+ ThreadSnapshotLoader,
16+ type ThreadSnapshotLoadResult,
17+} from "./threadSnapshotHttp.ts";
18+
19+// Same cold-open budget as the full snapshot path; bounded payloads should fit.
20+const DEFAULT_BOUNDED_THREAD_SNAPSHOT_TIMEOUT_MS = 6_000;
21+
22+/** Load a bounded recent-window thread snapshot over HTTP. */
23+export const fetchEnvironmentBoundedThreadSnapshot = Effect.fn(
24+ "clientRuntime.state.fetchEnvironmentBoundedThreadSnapshot",
25+)(function* (input: {
26+ readonly prepared: PreparedConnection;
27+ readonly threadId: ThreadId;
28+ readonly signer: Option.Option<ManagedRelayDpopSigner["Service"]>;
29+ readonly timeoutMs?: number;
30+}) {
31+ const requestUrl = environmentEndpointUrl(
32+ input.prepared.httpBaseUrl,
33+ `/api/orchestration/threads/${input.threadId}/bounded`,
34+ );
35+ const client = yield* makeEnvironmentHttpApiClient(input.prepared.httpBaseUrl);
36+ const headers = yield* buildEnvironmentAuthHeaders(
37+ input.prepared.httpAuthorization,
38+ "GET",
39+ requestUrl,
40+ input.signer,
41+ );
42+ return yield* executeEnvironmentHttpRequest(
43+ requestUrl,
44+ input.timeoutMs ?? DEFAULT_BOUNDED_THREAD_SNAPSHOT_TIMEOUT_MS,
45+ withEnvironmentCredentials(
46+ input.prepared.httpAuthorization,
47+ client.orchestration.threadBoundedSnapshot({
48+ params: { threadId: input.threadId },
49+ headers,
50+ }),
51+ ),
52+ );
53+});
54+
55+/**
56+ * Shared ThreadSnapshotLoader for clients that render progressive history.
57+ * Seeds from the bounded HTTP snapshot when available. Older servers that lack
58+ * the bounded route (plain/generic 404) fall back to the existing full HTTP
59+ * thread snapshot. Structured EnvironmentResourceNotFoundError from either
60+ * endpoint still means missing. Transient failures report `unavailable` so the
61+ * socket path remains a last resort for connectivity issues.
62+ */
63+export const boundedThreadSnapshotLoaderLayer: Layer.Layer<
64+ ThreadSnapshotLoader,
65+ never,
66+ HttpClient.HttpClient
67+> = Layer.effect(
68+ ThreadSnapshotLoader,
69+ Effect.gen(function* () {
70+ const httpClient = yield* HttpClient.HttpClient;
71+ const signer = yield* Effect.serviceOption(ManagedRelayDpopSigner);
72+ return ThreadSnapshotLoader.of({
73+ load: (prepared: PreparedConnection, threadId: ThreadId) => {
74+ const loadFullFallback = fetchEnvironmentThreadSnapshot({
75+ prepared,
76+ threadId,
77+ signer,
78+ }).pipe(
79+ Effect.map(
80+ (snapshot): ThreadSnapshotLoadResult => ({
81+ _tag: "present",
82+ snapshot,
83+ }),
84+ ),
85+ Effect.provideService(HttpClient.HttpClient, httpClient),
86+ Effect.catchTags({
87+ EnvironmentResourceNotFoundError: () =>
88+ Effect.logDebug(
89+ "Full thread snapshot not found over HTTP after bounded fallback; treating the thread as deleted.",
90+ ).pipe(
91+ Effect.annotateLogs({ threadId }),
92+ Effect.as({ _tag: "missing" } satisfies ThreadSnapshotLoadResult),
93+ ),
94+ }),
95+ Effect.catchCause((cause) =>
96+ Effect.logWarning(
97+ "Could not load the full thread snapshot over HTTP after bounded fallback; using the socket snapshot instead.",
98+ ).pipe(
99+ Effect.annotateLogs({ threadId, cause: Cause.pretty(cause) }),
100+ Effect.as({ _tag: "unavailable" } satisfies ThreadSnapshotLoadResult),
101+ ),
102+ ),
103+ );
104+
105+ return fetchEnvironmentBoundedThreadSnapshot({ prepared, threadId, signer }).pipe(
106+ Effect.map(
107+ (bounded): ThreadSnapshotLoadResult => ({
108+ _tag: "present",
109+ snapshot: {
110+ snapshotSequence: bounded.snapshotSequence,
111+ projection: bounded.projection,
112+ latestLocalTurnOrdinal: bounded.latestLocalTurnOrdinal,
113+ },
114+ history: {
115+ historyCursor: bounded.historyCursor,
116+ hasMoreHistory: bounded.hasMoreHistory,
117+ latestLocalTurnOrdinal: bounded.latestLocalTurnOrdinal,
118+ },
119+ }),
120+ ),
121+ Effect.provideService(HttpClient.HttpClient, httpClient),
122+ Effect.catchTags({
123+ EnvironmentResourceNotFoundError: () =>
124+ Effect.logDebug(
125+ "Bounded thread snapshot not found over HTTP; treating the thread as deleted.",
126+ ).pipe(
127+ Effect.annotateLogs({ threadId }),
128+ Effect.as({ _tag: "missing" } satisfies ThreadSnapshotLoadResult),
129+ ),
130+ RemoteEnvironmentAuthInvalidJsonError: (error) =>
131+ Effect.logDebug(
132+ "Bounded thread snapshot returned an invalid response; trying the full HTTP thread snapshot for an older server.",
133+ ).pipe(
134+ Effect.annotateLogs({ threadId, cause: error.message }),
135+ Effect.andThen(loadFullFallback),
136+ ),
137+ RemoteEnvironmentAuthUndeclaredStatusError: (error) =>
138+ error.status === 404
139+ ? Effect.logDebug(
140+ "Bounded thread snapshot route was not found; trying the full HTTP thread snapshot for an older server.",
141+ ).pipe(Effect.annotateLogs({ threadId }), Effect.andThen(loadFullFallback))
142+ : Effect.fail(error),
143+ }),
144+ Effect.catchCause((cause) =>
145+ Effect.logWarning(
146+ "Could not load the bounded thread snapshot over HTTP; using the socket snapshot instead.",
147+ ).pipe(
148+ Effect.annotateLogs({ threadId, cause: Cause.pretty(cause) }),
149+ Effect.as({ _tag: "unavailable" } satisfies ThreadSnapshotLoadResult),
150+ ),
151+ ),
152+ );
153+ },
154+ });
155+ }),
156+);