History

fix: reconcile main's round-18 features after the rebase

Adopts the round's main features into the v2 architecture: the #9023 media rework (streamed videos, media-file assets, protocol-relative links), #9098 shared live-activity row folded into the v2 working and thinking rows, the #9084/#9078 Claude model catalog for v2 consumers, a native #9005 OpenCode child-session abort in the v2 adapter, #9013's landed LegendList patch, and per-environment sidebar provider entries. For #8600 the server-side pieces land, but auto-settle evaluation stays client-side (reading the new server-owned settings) until the v2 orchestrator grows its own settlement reactor; main's v1-only reactor and coalescer additions are dropped with the rest of the v1 path.

Julius Marminge committed 2 months ago

Commit 7b13bca1 parent 786a25847 files · +1855 −2616Browse files

Showing the changes to apps/server/src (17 of 47 files). Show all

Unified · Split

17 files
  1. apps/server/src/claudeModelOptions.ts +14 −11
  2. apps/server/src/http.test.ts +177 −3
  3. apps/server/src/orchestration-v2/Adapters/ClaudeAdapterV2.ts +7 −2
  4. apps/server/src/orchestration-v2/Adapters/OpenCodeAdapterV2.ts +30 −0
  5. apps/server/src/orchestration/Layers/OrchestrationEngine.ts +0 −33
  6. apps/server/src/orchestration/Layers/ProjectionPipeline.ts +10 −23
  7. apps/server/src/orchestration/Layers/ProjectionSnapshotQuery.ts +70 −258
  8. apps/server/src/orchestration/Services/ProjectionSnapshotQuery.ts +0 −4
  9. apps/server/src/orchestration/ThreadLiveEventCoalescer.test.ts +0 −250
  10. apps/server/src/orchestration/ThreadLiveEventCoalescer.ts +0 −241
  11. apps/server/src/orchestration/ThreadSettlementReactor.test.ts +0 −858
  12. apps/server/src/orchestration/ThreadSettlementReactor.ts +0 −249
  13. apps/server/src/orchestration/decider.ts +1 −0
  14. apps/server/src/persistence/Layers/OrchestrationEventStore.ts +11 −10
  15. apps/server/src/persistence/Services/OrchestrationEventStore.ts +3 −2
  16. apps/server/src/provider/Drivers/ClaudeDriver.ts +24 −19
  17. apps/server/src/provider/acp/AcpNativeLogging.ts +44 −6
apps/server/src/claudeModelOptions.ts+14−11
@@ -7,12 +7,14 @@ import { · 6 unmodified lines
7} from "@t3tools/shared/model";7} from "@t3tools/shared/model";
8 8
9import {9import {
10 getClaudeModelCapabilities,10 BUNDLED_CLAUDE_MODEL_CATALOG,
11 isClaudeUltracodeEffort,11 getClaudeCatalogModelCapabilities,
12 normalizeClaudeCliEffort,12 isClaudeCatalogUltracodeEffort,
13 resolveClaudeApiModelId,13 normalizeClaudeCatalogEffort,
14 resolveClaudeEffort,14 resolveClaudeCatalogApiModelId,
15} from "./provider/Layers/ClaudeProvider.ts";15 resolveClaudeCatalogEffort,
16 type ClaudeModelCatalog,
17} from "./provider/ClaudeModelCatalog.ts";
16 18
17export interface CompiledClaudeModelSelection {19export interface CompiledClaudeModelSelection {
18 readonly apiModelId: string;20 readonly apiModelId: string;
@@ -25,14 +27,15 @@ export interface CompiledClaudeModelSelection { · 6 unmodified lines
25/** Compile every Claude model option at the provider boundary. */27/** Compile every Claude model option at the provider boundary. */
26export function compileClaudeModelSelection(28export function compileClaudeModelSelection(
27 selection: ModelSelection,29 selection: ModelSelection,
30 catalog: ClaudeModelCatalog = BUNDLED_CLAUDE_MODEL_CATALOG,
28): CompiledClaudeModelSelection {31): CompiledClaudeModelSelection {
29 const capabilities = getClaudeModelCapabilities(selection.model);32 const capabilities = getClaudeCatalogModelCapabilities(catalog, selection.model);
30 const descriptors = getProviderOptionDescriptors({ caps: capabilities });33 const descriptors = getProviderOptionDescriptors({ caps: capabilities });
31 const supportsBoolean = (id: string) =>34 const supportsBoolean = (id: string) =>
32 descriptors.some((descriptor) => descriptor.type === "boolean" && descriptor.id === id);35 descriptors.some((descriptor) => descriptor.type === "boolean" && descriptor.id === id);
33 const rawEffort = getModelSelectionStringOptionValue(selection, "effort");36 const rawEffort = getModelSelectionStringOptionValue(selection, "effort");
34 const resolvedEffort = resolveClaudeEffort(capabilities, rawEffort);37 const resolvedEffort = resolveClaudeCatalogEffort(catalog, selection.model, rawEffort);
35 const effort = normalizeClaudeCliEffort(resolvedEffort, selection.model);38 const effort = normalizeClaudeCatalogEffort(catalog, resolvedEffort, selection.model);
36 const fastMode = supportsBoolean("fastMode")39 const fastMode = supportsBoolean("fastMode")
37 ? getModelSelectionBooleanOptionValue(selection, "fastMode")40 ? getModelSelectionBooleanOptionValue(selection, "fastMode")
38 : undefined;41 : undefined;
@@ -42,9 +45,9 @@ export function compileClaudeModelSelection( · 3 unmodified lines
42 const settings = {45 const settings = {
43 ...(typeof thinking === "boolean" ? { alwaysThinkingEnabled: thinking } : {}),46 ...(typeof thinking === "boolean" ? { alwaysThinkingEnabled: thinking } : {}),
44 ...(typeof fastMode === "boolean" ? { fastMode } : {}),47 ...(typeof fastMode === "boolean" ? { fastMode } : {}),
45 ...(isClaudeUltracodeEffort(resolvedEffort) ? { ultracode: true } : {}),48 ...(isClaudeCatalogUltracodeEffort(resolvedEffort) ? { ultracode: true } : {}),
46 };49 };
47 const apiModelId = resolveClaudeApiModelId(selection);50 const apiModelId = resolveClaudeCatalogApiModelId(catalog, selection);
48 const promptEffort = resolvePromptInjectedEffort(capabilities, rawEffort) ?? undefined;51 const promptEffort = resolvePromptInjectedEffort(capabilities, rawEffort) ?? undefined;
49 return {52 return {
50 apiModelId,53 apiModelId,
apps/server/src/http.test.ts+177−3
@@ -1,12 +1,13 @@
1import { expect, it } from "@effect/vitest";1import { expect, it } from "@effect/vitest";
2import { describe, vi } from "vite-plus/test";
3import * as NodeHttpPlatform from "@effect/platform-node/NodeHttpPlatform";
4import * as NodeServices from "@effect/platform-node/NodeServices";
2import * as Effect from "effect/Effect";5import * as Effect from "effect/Effect";
3import * as FileSystem from "effect/FileSystem";6import * as FileSystem from "effect/FileSystem";
4import * as Layer from "effect/Layer";7import * as Layer from "effect/Layer";
5import * as Path from "effect/Path";8import * as Path from "effect/Path";
6import { HttpRouter, HttpServerResponse } from "effect/unstable/http";9import { HttpRouter, HttpServerResponse } from "effect/unstable/http";
7import { describe } from "vite-plus/test";10import { openMediaFile } from "./assets/MediaFile.ts";
8import * as NodeHttpPlatform from "@effect/platform-node/NodeHttpPlatform";
9import * as NodeServices from "@effect/platform-node/NodeServices";
10 11
11import { ORCHESTRATION_PROTOCOL_HEADER } from "@t3tools/contracts";12import { ORCHESTRATION_PROTOCOL_HEADER } from "@t3tools/contracts";
12 13
@@ -70,6 +71,179 @@ describe("browser API CORS", () => { · 57 unmodified lines
70const fileResponseLayer = Layer.mergeAll(NodeHttpPlatform.layer, NodeServices.layer);71const fileResponseLayer = Layer.mergeAll(NodeHttpPlatform.layer, NodeServices.layer);
71 72
72describe("video asset byte ranges", () => {73describe("video asset byte ranges", () => {
74 it.effect("uses current descriptor metadata after an in-place truncate or extension", () =>
75 Effect.gen(function* () {
76 const fs = yield* FileSystem.FileSystem;
77 const path = yield* Path.Path;
78 const directory = yield* fs.makeTempDirectoryScoped({ prefix: "t3-guarded-current-stat-" });
79 const filePath = path.join(directory, "clip.mp4");
80 for (const [contents, range, method, expected, status, contentRange] of [
81 ["1234", undefined, "GET", "1234", 200, null],
82 ["0123456789abcdef", undefined, "GET", "0123456789abcdef", 200, null],
83 ["1234", "bytes=4-", "GET", "", 416, "bytes */4"],
84 ["1234", "bytes=1-20", "GET", "234", 206, "bytes 1-3/4"],
85 ["0123456789abcdef", "bytes=10-", "GET", "abcdef", 206, "bytes 10-15/16"],
86 ["0123456789abcdef", undefined, "HEAD", "", 200, null],
87 ["", undefined, "GET", "", 200, null],
88 ["", "bytes=0-1", "GET", "", 416, "bytes */0"],
89 ] as const) {
90 yield* fs.writeFileString(filePath, "0123456789");
91 const canonicalPath = yield* fs.realPath(filePath);
92 const file = yield* openMediaFile(canonicalPath);
93 if (!file) throw new Error("Expected an opened media file");
94 yield* fs.writeFileString(filePath, contents);
95 const response = HttpServerResponse.toWeb(
96 yield* assetFileResponse(
97 { path: canonicalPath, file, mimeType: "video/mp4" },
98 range,
99 undefined,
100 method,
101 ),
102 );
103 expect(response.status).toBe(status);
104 expect(response.headers.get("content-range")).toBe(contentRange);
105 if (status !== 416) {
106 expect(response.headers.get("content-length")).toBe(
107 String(method === "HEAD" ? contents.length : expected.length),
108 );
109 }
110 expect(yield* Effect.promise(() => response.text())).toBe(expected);
111 }
112 }).pipe(Effect.provide(fileResponseLayer)),
113 );
114
115 it.effect(
116 "rejects unaddressable ranges before streaming and preserves small ranges on large files",
117 () =>
118 Effect.gen(function* () {
119 const fs = yield* FileSystem.FileSystem;
120 const path = yield* Path.Path;
121 const directory = yield* fs.makeTempDirectoryScoped({ prefix: "t3-guarded-offset-limit-" });
122 const filePath = path.join(directory, "clip.mp4");
123 yield* fs.writeFileString(filePath, "0123456789");
124 const canonicalPath = yield* fs.realPath(filePath);
125 const unsafeOffset = BigInt(Number.MAX_SAFE_INTEGER) + 1n;
126 const size = unsafeOffset + 32n;
127 for (const [range, status] of [
128 [`bytes=${unsafeOffset}-${unsafeOffset}`, 416],
129 [`bytes=0-${unsafeOffset}`, 416],
130 ["bytes=-1", 416],
131 [undefined, 413],
132 ["bytes=0-1", 206],
133 ] as const) {
134 const file = yield* openMediaFile(canonicalPath);
135 if (!file) throw new Error("Expected an opened media file");
136 // Model a sparse file beyond the native stream's numeric addressing limit.
137 const info = yield* Effect.promise(() => file.handle.stat({ bigint: true }));
138 info.size = size;
139 const statSpy = vi.spyOn(file.handle, "stat").mockResolvedValue(info);
140 yield* Effect.addFinalizer(() => Effect.sync(() => statSpy.mockRestore()));
141 const response = HttpServerResponse.toWeb(
142 yield* assetFileResponse({ path: canonicalPath, file, mimeType: "video/mp4" }, range),
143 );
144 expect(response.status).toBe(status);
145 if (status === 416) {
146 expect(response.headers.get("content-range")).toBe(`bytes */${size}`);
147 expect(yield* Effect.promise(() => response.text())).toBe("");
148 } else if (status === 206) {
149 expect(response.headers.get("content-range")).toBe(`bytes 0-1/${size}`);
150 expect(yield* Effect.promise(() => response.text())).toBe("01");
151 } else {
152 expect(yield* Effect.promise(() => response.text())).toBe(
153 "File is too large to preview.",
154 );
155 }
156 }
157 }).pipe(Effect.provide(fileResponseLayer)),
158 );
159
160 it.effect("streams guarded file ranges, including suffixes and conditional requests", () =>
161 Effect.gen(function* () {
162 const fs = yield* FileSystem.FileSystem;
163 const path = yield* Path.Path;
164 const directory = yield* fs.makeTempDirectoryScoped({ prefix: "t3-guarded-range-" });
165 const filePath = path.join(directory, "clip.mp4");
166 yield* fs.writeFileString(filePath, "0123456789");
167 const canonicalPath = yield* fs.realPath(filePath);
168 for (const [range, ifRange, expected, status, contentRange] of [
169 [undefined, undefined, "0123456789", 200, null],
170 ["bytes=0-1", undefined, "01", 206, "bytes 0-1/10"],
171 ["bytes=4-", undefined, "456789", 206, "bytes 4-9/10"],
172 ["bytes=-3", undefined, "789", 206, "bytes 7-9/10"],
173 ["bytes=-999999999999999999999999", undefined, "0123456789", 206, "bytes 0-9/10"],
174 ["bytes=10-", undefined, "", 416, "bytes */10"],
175 ["bytes=0-1", '"old-etag"', "0123456789", 200, null],
176 ["bytes=0-1", "", "0123456789", 200, null],
177 ] as const) {
178 const file = yield* openMediaFile(canonicalPath);
179 if (!file) throw new Error("Expected an opened media file");
180 const response = HttpServerResponse.toWeb(
181 yield* assetFileResponse(
182 { path: canonicalPath, file, mimeType: "video/mp4" },
183 range,
184 ifRange,
185 ),
186 );
187 expect(response.status).toBe(status);
188 expect(response.headers.get("accept-ranges")).toBe("bytes");
189 expect(response.headers.get("content-range")).toBe(contentRange);
190 expect(response.headers.get("cache-control")).toBe("private, no-store");
191 expect(response.headers.get("etag")).toBeNull();
192 expect(response.headers.get("last-modified")).toBeNull();
193 if (status !== 416)
194 expect(response.headers.get("content-length")).toBe(String(expected.length));
195 expect(yield* Effect.promise(() => response.text())).toBe(expected);
196 }
197 }).pipe(Effect.provide(fileResponseLayer)),
198 );
199
200 it.effect("closes guarded descriptors after full, HEAD, rejected, and cancelled responses", () =>
201 Effect.gen(function* () {
202 const fs = yield* FileSystem.FileSystem;
203 const path = yield* Path.Path;
204 const directory = yield* fs.makeTempDirectoryScoped({ prefix: "t3-guarded-cleanup-" });
205 const filePath = path.join(directory, "clip.mp4");
206 const bytes = new Uint8Array(1024 * 1024).fill(42);
207 yield* fs.writeFile(filePath, bytes);
208 const canonicalPath = yield* fs.realPath(filePath);
209 for (const mode of ["full", "HEAD", "rejected", "cancelled"] as const) {
210 const file = yield* Effect.scoped(
211 Effect.gen(function* () {
212 const file = yield* openMediaFile(canonicalPath);
213 if (!file) throw new Error("Expected an opened media file");
214 const response = HttpServerResponse.toWeb(
215 yield* assetFileResponse(
216 { path: canonicalPath, file, mimeType: "video/mp4" },
217 mode === "rejected" ? `bytes=${bytes.length}-` : "bytes=0-",
218 undefined,
219 mode === "HEAD" ? "HEAD" : "GET",
220 ),
221 );
222 if (mode === "HEAD") {
223 expect(response.status).toBe(200);
224 expect(response.headers.get("content-length")).toBe(String(bytes.length));
225 expect(response.headers.get("content-range")).toBeNull();
226 expect(yield* Effect.promise(() => response.text())).toBe("");
227 } else if (mode === "rejected") {
228 expect(response.status).toBe(416);
229 expect(yield* Effect.promise(() => response.text())).toBe("");
230 } else if (mode === "cancelled") {
231 const reader = response.body!.getReader();
232 const first = yield* Effect.promise(() => reader.read());
233 expect(first.done).toBe(false);
234 expect(first.value!.byteLength).toBeLessThan(bytes.length);
235 yield* Effect.promise(() => reader.cancel());
236 } else {
237 expect(yield* Effect.promise(() => response.arrayBuffer())).toEqual(bytes.buffer);
238 }
239 return file;
240 }),
241 );
242 expect(file.handle.fd).toBe(-1);
243 }
244 }).pipe(Effect.provide(fileResponseLayer)),
245 );
246
73 it.effect("streams exactly the requested bytes and leaves full downloads intact", () =>247 it.effect("streams exactly the requested bytes and leaves full downloads intact", () =>
74 Effect.gen(function* () {248 Effect.gen(function* () {
75 const fs = yield* FileSystem.FileSystem;249 const fs = yield* FileSystem.FileSystem;
apps/server/src/orchestration-v2/Adapters/ClaudeAdapterV2.ts+7−2
@@ -77,7 +77,10 @@ import { resolveAttachmentPath } from "../../attachmentStore.ts"; · 76 unmodified lines
77import { compileClaudeModelSelection } from "../../claudeModelOptions.ts";77import { compileClaudeModelSelection } from "../../claudeModelOptions.ts";
78import { ServerConfig } from "../../config.ts";78import { ServerConfig } from "../../config.ts";
79import { makeClaudeEnvironment } from "../../provider/Drivers/ClaudeHome.ts";79import { makeClaudeEnvironment } from "../../provider/Drivers/ClaudeHome.ts";
80import { resolveClaudeContextWindow } from "../../provider/Layers/ClaudeProvider.ts";80import {
81 BUNDLED_CLAUDE_MODEL_CATALOG,
82 resolveClaudeCatalogContextWindow,
83} from "../../provider/ClaudeModelCatalog.ts";
81import type { EventNdjsonLogger } from "../../provider/Layers/EventNdjsonLogger.ts";84import type { EventNdjsonLogger } from "../../provider/Layers/EventNdjsonLogger.ts";
82import { ProviderEventLoggers } from "../../provider/Layers/ProviderEventLoggers.ts";85import { ProviderEventLoggers } from "../../provider/Layers/ProviderEventLoggers.ts";
83import { mergeProviderInstanceEnvironment } from "../../provider/ProviderInstanceEnvironment.ts";86import { mergeProviderInstanceEnvironment } from "../../provider/ProviderInstanceEnvironment.ts";
@@ -134,7 +137,9 @@ function claudeContextWindow(modelSelection: ModelSelection): number | null { · 50 unmodified lines
134 if (modelSelection.model === "claude-opus-4-6" || modelSelection.model === "claude-opus-4-7") {137 if (modelSelection.model === "claude-opus-4-6" || modelSelection.model === "claude-opus-4-7") {
135 return 1_000_000;138 return 1_000_000;
136 }139 }
137 return resolveClaudeContextWindow(modelSelection) === "1m" ? 1_000_000 : 200_000;140 return resolveClaudeCatalogContextWindow(BUNDLED_CLAUDE_MODEL_CATALOG, modelSelection) === "1m"
141 ? 1_000_000
142 : 200_000;
138}143}
139 144
140export function claudeProviderTurnTokenUsage(145export function claudeProviderTurnTokenUsage(
apps/server/src/orchestration-v2/Adapters/OpenCodeAdapterV2.ts+30−0
@@ -3027,6 +3027,36 @@ export function makeOpenCodeAdapterV2(options: OpenCodeAdapterV2Options): Provid · 3026 unmodified lines
3027 }),3027 }),
3028 ),3028 ),
3029 );3029 );
3030 // Child agents run in their own sessions and outlive a root
3031 // abort (#9005). Best-effort: walk session.children and abort
3032 // each descendant; the root abort above already settled the
3033 // interrupt contract.
3034 yield* Effect.gen(function* () {
3035 const visited = new Set([sessionId]);
3036 const abortDescendants = (parentId: string): Effect.Effect<void> =>
3037 Effect.gen(function* () {
3038 const children = yield* sdkCall(
3039 "session.children",
3040 { sessionID: parentId },
3041 () => client.session.children({ sessionID: parentId }),
3042 ).pipe(Effect.option);
3043 const rows = Option.isSome(children) ? (children.value.data ?? []) : [];
3044 const fresh = rows.filter((child) => {
3045 if (visited.has(child.id)) return false;
3046 visited.add(child.id);
3047 return true;
3048 });
3049 yield* Effect.forEach(
3050 fresh,
3051 (child) =>
3052 sdkCall("session.abort", { sessionID: child.id }, () =>
3053 client.session.abort({ sessionID: child.id }),
3054 ).pipe(Effect.ignore, Effect.andThen(abortDescendants(child.id))),
3055 { concurrency: 8, discard: true },
3056 );
3057 });
3058 yield* abortDescendants(sessionId);
3059 }).pipe(Effect.timeout("15 seconds"), Effect.ignore);
3030 }).pipe(3060 }).pipe(
3031 Effect.mapError(3061 Effect.mapError(
3032 (cause) =>3062 (cause) =>
apps/server/src/orchestration/Layers/OrchestrationEngine.ts+0−33
@@ -166,42 +166,9 @@ const makeOrchestrationEngine = Effect.gen(function* () { · 165 unmodified lines
166 });166 });
167 }167 }
168 168
169 if (
170 envelope.command.type === "thread.auto-settle" &&
171 (yield* eventStore.hasEventAfter({
172 aggregateKind: "thread",
173 aggregateId: envelope.command.threadId,
174 sequenceExclusive: envelope.command.snapshotSequence,
175 }))
176 ) {
177 return yield* new OrchestrationCommandInvariantError({
178 commandType: envelope.command.type,
179 detail: `thread ${envelope.command.threadId} changed before automatic settlement`,
180 });
181 }
182
183 if (
184 envelope.command.type === "thread.auto-settle" &&
185 threadBackgroundLiveness.getThreadBackgroundLiveness(envelope.command.threadId) !== null
186 ) {
187 return yield* new OrchestrationCommandInvariantError({
188 commandType: envelope.command.type,
189 detail: `thread ${envelope.command.threadId} has live background work`,
190 });
191 }
192
193 // Command snapshots omit activities at startup and cap them while running.
194 // Read this request's durable state before deciding how to send the answer.
195 const userInputActivity =
196 envelope.command.type === "thread.user-input.respond"
197 ? yield* projectionSnapshotQuery.getUserInputActivity(envelope.command)
198 : Option.none();
199 const eventBase = yield* decideOrchestrationCommand({169 const eventBase = yield* decideOrchestrationCommand({
200 command: envelope.command,170 command: envelope.command,
201 readModel: commandReadModel,171 readModel: commandReadModel,
202 ...(Option.isSome(userInputActivity)
203 ? { userInputActivity: userInputActivity.value }
204 : {}),
205 }).pipe(172 }).pipe(
206 Effect.provideService(Crypto.Crypto, crypto),173 Effect.provideService(Crypto.Crypto, crypto),
207 Effect.mapError((cause) =>174 Effect.mapError((cause) =>
apps/server/src/orchestration/Layers/ProjectionPipeline.ts+10−23
@@ -1018,34 +1018,21 @@ const makeOrchestrationProjectionPipeline = Effect.fn("makeOrchestrationProjecti · 1017 unmodified lines
1018 return;1018 return;
1019 1019
1020 case "thread.message-sent": {1020 case "thread.message-sent": {
1021 if (event.payload.streaming) {
1022 const attachments =
1023 event.payload.attachments !== undefined
1024 ? yield* materializeAttachmentsForProjection({
1025 attachments: event.payload.attachments,
1026 })
1027 : undefined;
1028 yield* projectionThreadMessageRepository.appendStreaming({
1029 messageId: event.payload.messageId,
1030 threadId: event.payload.threadId,
1031 turnId: event.payload.turnId,
1032 role: event.payload.role,
1033 text: event.payload.text,
1034 ...(attachments !== undefined ? { attachments: [...attachments] } : {}),
1035 createdAt: event.payload.createdAt,
1036 updatedAt: event.payload.updatedAt,
1037 });
1038 return;
1039 }
1040
1041 const existingMessage = yield* projectionThreadMessageRepository.getByMessageId({1021 const existingMessage = yield* projectionThreadMessageRepository.getByMessageId({
1042 messageId: event.payload.messageId,1022 messageId: event.payload.messageId,
1043 });1023 });
1044 const previousMessage = Option.getOrUndefined(existingMessage);1024 const previousMessage = Option.getOrUndefined(existingMessage);
1045 const nextText = Option.match(existingMessage, {1025 const nextText = Option.match(existingMessage, {
1046 onNone: () => event.payload.text,1026 onNone: () => event.payload.text,
1047 onSome: (message) =>1027 onSome: (message) => {
1048 event.payload.text.length === 0 ? message.text : event.payload.text,1028 if (event.payload.streaming) {
1029 return `${message.text}${event.payload.text}`;
1030 }
1031 if (event.payload.text.length === 0) {
1032 return message.text;
1033 }
1034 return event.payload.text;
1035 },
1049 });1036 });
1050 const nextAttachments =1037 const nextAttachments =
1051 event.payload.attachments !== undefined1038 event.payload.attachments !== undefined
@@ -1060,7 +1047,7 @@ const makeOrchestrationProjectionPipeline = Effect.fn("makeOrchestrationProjecti · 8 unmodified lines
1060 role: event.payload.role,1047 role: event.payload.role,
1061 text: nextText,1048 text: nextText,
1062 ...(nextAttachments !== undefined ? { attachments: [...nextAttachments] } : {}),1049 ...(nextAttachments !== undefined ? { attachments: [...nextAttachments] } : {}),
1063 isStreaming: false,1050 isStreaming: event.payload.streaming,
1064 createdAt: previousMessage?.createdAt ?? event.payload.createdAt,1051 createdAt: previousMessage?.createdAt ?? event.payload.createdAt,
1065 updatedAt: event.payload.updatedAt,1052 updatedAt: event.payload.updatedAt,
1066 });1053 });
apps/server/src/orchestration/Layers/ProjectionSnapshotQuery.ts+70−258
@@ -77,9 +77,6 @@ const decodeThread = Schema.decodeUnknownEffect(OrchestrationThread); · 76 unmodified lines
77// activity window. Applying the limit in SQL avoids decoding an unbounded77// activity window. Applying the limit in SQL avoids decoding an unbounded
78// payload_json set before the projector can enforce that invariant.78// payload_json set before the projector can enforce that invariant.
79const THREAD_DETAIL_ACTIVITY_LIMIT = 500;79const THREAD_DETAIL_ACTIVITY_LIMIT = 500;
80// Snapshot payloads are decoded and projected in small sequential batches so
81// one client read does not retain the raw payloads for the full activity window.
82const THREAD_DETAIL_ACTIVITY_PAYLOAD_BATCH_SIZE = 25;
83const ProjectionProjectDbRowSchema = ProjectionProject.mapFields(80const ProjectionProjectDbRowSchema = ProjectionProject.mapFields(
84 Struct.assign({81 Struct.assign({
85 defaultModelSelection: Schema.NullOr(Schema.fromJsonString(ModelSelection)),82 defaultModelSelection: Schema.NullOr(Schema.fromJsonString(ModelSelection)),
@@ -104,9 +101,6 @@ const ProjectionThreadActivityDbRowSchema = ProjectionThreadActivity.mapFields( · 18 unmodified lines
104 sequence: Schema.NullOr(NonNegativeInt),101 sequence: Schema.NullOr(NonNegativeInt),
105 }),102 }),
106);103);
107const ProjectionThreadActivityIdRowSchema = Schema.Struct({
108 activityId: ProjectionThreadActivity.fields.activityId,
109});
110const ProjectionThreadSessionDbRowSchema = ProjectionThreadSession;104const ProjectionThreadSessionDbRowSchema = ProjectionThreadSession;
111const ProjectionThreadRuntimeContextDbRowSchema = Schema.Struct({105const ProjectionThreadRuntimeContextDbRowSchema = Schema.Struct({
112 id: ThreadId,106 id: ThreadId,
@@ -166,9 +160,6 @@ const ThreadActivityKindsLookupInput = Schema.Struct({ · 53 unmodified lines
166 threadId: ThreadId,160 threadId: ThreadId,
167 activityKinds: Schema.Array(Schema.String),161 activityKinds: Schema.Array(Schema.String),
168});162});
169const ThreadActivityIdsLookupInput = Schema.Struct({
170 activityIds: Schema.Array(ProjectionThreadActivity.fields.activityId),
171});
172// Windowed reads order turns by the stable keyset (anchor, turn key), where163// Windowed reads order turns by the stable keyset (anchor, turn key), where
173// anchor is requested_at and turn key is164// anchor is requested_at and turn key is
174// COALESCE(turn_id, ''). Both are event-derived, so cursors survive the165// COALESCE(turn_id, ''). Both are event-derived, so cursors survive the
@@ -372,21 +363,6 @@ function mapProposedPlanRow( · 197 unmodified lines
372 };363 };
373}364}
374 365
375function mapThreadActivityRow(
376 row: Schema.Schema.Type<typeof ProjectionThreadActivityDbRowSchema>,
377): OrchestrationThreadActivity {
378 return {
379 id: row.activityId,
380 tone: row.tone,
381 kind: row.kind,
382 summary: row.summary,
383 payload: row.payload,
384 turnId: row.turnId,
385 createdAt: row.createdAt,
386 ...(row.sequence !== null ? { sequence: row.sequence } : {}),
387 };
388}
389
390function toPersistenceSqlOrDecodeError(sqlOperation: string, decodeOperation: string) {366function toPersistenceSqlOrDecodeError(sqlOperation: string, decodeOperation: string) {
391 return (cause: unknown): ProjectionRepositoryError =>367 return (cause: unknown): ProjectionRepositoryError =>
392 Schema.isSchemaError(cause)368 Schema.isSchemaError(cause)
@@ -1187,44 +1163,6 @@ const makeProjectionSnapshotQuery = Effect.gen(function* () { · 794 unmodified lines
1187 `,1163 `,
1188 });1164 });
1189 1165
1190 const listThreadActivityIdsByThread = SqlSchema.findAll({
1191 Request: ThreadIdLookupInput,
1192 Result: ProjectionThreadActivityIdRowSchema,
1193 execute: ({ threadId }) =>
1194 sql`
1195 SELECT activity_id AS "activityId"
1196 FROM projection_thread_activities
1197 WHERE thread_id = ${threadId}
1198 ORDER BY
1199 sequence DESC,
1200 created_at DESC,
1201 activity_id DESC
1202 LIMIT ${THREAD_DETAIL_ACTIVITY_LIMIT}
1203 `,
1204 });
1205
1206 const listThreadActivityRowsByIds = SqlSchema.findAll({
1207 Request: ThreadActivityIdsLookupInput,
1208 Result: ProjectionThreadActivityDbRowSchema,
1209 execute: ({ activityIds }) =>
1210 sql`
1211 SELECT
1212 activity_id AS "activityId",
1213 thread_id AS "threadId",
1214 turn_id AS "turnId",
1215 tone,
1216 kind,
1217 summary,
1218 payload_json AS "payload",
1219 sequence,
1220 created_at AS "createdAt"
1221 FROM projection_thread_activities
1222 -- The selectors already scoped these globally unique ids to the
1223 -- thread inside this transaction. Keep this as a primary-key lookup.
1224 WHERE ${sql.in("activity_id", activityIds)}
1225 `,
1226 });
1227
1228 const listThreadActivityRowsByThreadAndKinds = SqlSchema.findAll({1166 const listThreadActivityRowsByThreadAndKinds = SqlSchema.findAll({
1229 Request: ThreadActivityKindsLookupInput,1167 Request: ThreadActivityKindsLookupInput,
1230 Result: ProjectionThreadActivityDbRowSchema,1168 Result: ProjectionThreadActivityDbRowSchema,
@@ -1475,8 +1413,15 @@ const makeProjectionSnapshotQuery = Effect.gen(function* () { · 244 unmodified lines
1475 `,1413 `,
1476 });1414 });
1477 1415
1478 const pinnedThreadActivityIdsCte = (threadId: string) => sql`1416 // Blocking request payloads must remain available even if they predate the
1479pending_approval_requests AS (1417 // recent activity window. Each CTE returns at most one unresolved row per
1418 // request, so the merge below stays bounded by actionable work.
1419 const listPinnedThreadActivityRowsByThread = SqlSchema.findAll({
1420 Request: ThreadIdLookupInput,
1421 Result: ProjectionThreadActivityDbRowSchema,
1422 execute: ({ threadId }) =>
1423 sql`
1424 WITH pending_approval_requests AS (
1480 SELECT request_id, thread_id1425 SELECT request_id, thread_id
1481 FROM projection_pending_approvals1426 FROM projection_pending_approvals
1482 WHERE thread_id = ${threadId}1427 WHERE thread_id = ${threadId}
@@ -1540,17 +1485,6 @@ pending_approval_requests AS ( · 57 unmodified lines
1540 WHERE request_order = 11485 WHERE request_order = 1
1541 AND kind = 'user-input.requested'1486 AND kind = 'user-input.requested'
1542 )1487 )
1543 `;
1544
1545 // Blocking request payloads must remain available even if they predate the
1546 // recent activity window. Each CTE returns at most one unresolved row per
1547 // request, so the merge below stays bounded by actionable work.
1548 const listPinnedThreadActivityRowsByThread = SqlSchema.findAll({
1549 Request: ThreadIdLookupInput,
1550 Result: ProjectionThreadActivityDbRowSchema,
1551 execute: ({ threadId }) =>
1552 sql`
1553 WITH ${pinnedThreadActivityIdsCte(threadId)}
1554 SELECT1488 SELECT
1555 activity.activity_id AS "activityId",1489 activity.activity_id AS "activityId",
1556 activity.thread_id AS "threadId",1490 activity.thread_id AS "threadId",
@@ -1568,17 +1502,6 @@ pending_approval_requests AS ( · 11 unmodified lines
1568 `,1502 `,
1569 });1503 });
1570 1504
1571 const listPinnedThreadActivityIdsByThread = SqlSchema.findAll({
1572 Request: ThreadIdLookupInput,
1573 Result: ProjectionThreadActivityIdRowSchema,
1574 execute: ({ threadId }) =>
1575 sql`
1576 WITH ${pinnedThreadActivityIdsCte(threadId)}
1577 SELECT activity_id AS "activityId"
1578 FROM pinned_activity_ids
1579 `,
1580 });
1581
1582 const listThreadActivityRowsByThreadWindow = SqlSchema.findAll({1505 const listThreadActivityRowsByThreadWindow = SqlSchema.findAll({
1583 Request: ThreadTurnRangeLookupInput,1506 Request: ThreadTurnRangeLookupInput,
1584 Result: ProjectionThreadActivityDbRowSchema,1507 Result: ProjectionThreadActivityDbRowSchema,
@@ -1646,48 +1569,6 @@ pending_approval_requests AS ( · 61 unmodified lines
1646 `,1569 `,
1647 });1570 });
1648 1571
1649 const listThreadActivityIdsByThreadWindow = SqlSchema.findAll({
1650 Request: ThreadTurnRangeLookupInput,
1651 Result: ProjectionThreadActivityIdRowSchema,
1652 execute: ({ threadId, minAnchorAt, minTurnKey, beforeAnchorAt, beforeTurnKey }) =>
1653 sql`
1654 SELECT activity_id AS "activityId"
1655 FROM projection_thread_activities
1656 WHERE thread_id = ${threadId}
1657 AND (
1658 turn_id IN (
1659 SELECT turn_id FROM projection_turns
1660 WHERE thread_id = ${threadId}
1661 AND turn_id IS NOT NULL
1662 AND (
1663 requested_at > ${minAnchorAt}
1664 OR (
1665 requested_at = ${minAnchorAt}
1666 AND turn_id >= ${minTurnKey}
1667 )
1668 )
1669 AND (
1670 requested_at < ${beforeAnchorAt}
1671 OR (
1672 requested_at = ${beforeAnchorAt}
1673 AND turn_id < ${beforeTurnKey}
1674 )
1675 )
1676 )
1677 OR (
1678 turn_id IS NULL
1679 AND created_at >= ${minAnchorAt}
1680 AND created_at < ${beforeAnchorAt}
1681 )
1682 )
1683 ORDER BY
1684 sequence DESC,
1685 created_at DESC,
1686 activity_id DESC
1687 LIMIT ${THREAD_DETAIL_ACTIVITY_LIMIT}
1688 `,
1689 });
1690
1691 const getFullThreadDiffContextRow = SqlSchema.findOneOption({1572 const getFullThreadDiffContextRow = SqlSchema.findOneOption({
1692 Request: FullThreadDiffContextLookupInput,1573 Request: FullThreadDiffContextLookupInput,
1693 Result: ProjectionFullThreadDiffContextRowSchema,1574 Result: ProjectionFullThreadDiffContextRowSchema,
@@ -2832,136 +2713,49 @@ pending_approval_requests AS ( · 1138 unmodified lines
2832 readonly beforeTurnKey: string;2713 readonly beforeTurnKey: string;
2833 }2714 }
2834 2715
2835 type ThreadDetailActivityRead =2716 const getThreadDetailByIdBounded = (
2836 | {2717 threadId: ThreadId,
2837 readonly mode: "raw";2718 bounds: ThreadDetailBounds | undefined,
2838 readonly query?: ProjectionThreadDetailQuery;2719 query: ProjectionThreadDetailQuery | undefined = undefined,
2839 }2720 ) =>
2840 | {2721 Effect.gen(function* () {
2841 readonly mode: "client";2722 const activityRowsEffect = (
2842 };2723 query?.activityKinds === undefined
2843 2724 ? bounds === undefined
2844 const listProjectedThreadActivities = Effect.fn(2725 ? listThreadActivityRowsByThread({ threadId })
2845 "ProjectionSnapshotQuery.listProjectedThreadActivities",2726 : listThreadActivityRowsByThreadWindow({ threadId, ...bounds })
2846 )(function* (threadId: ThreadId, bounds: ThreadDetailBounds | undefined) {2727 : query.activityKinds.length === 0
2847 const [activityIdRows, pinnedActivityIdRows] = yield* Effect.all([2728 ? Effect.succeed([])
2848 (bounds === undefined2729 : listThreadActivityRowsByThreadAndKinds({
2849 ? listThreadActivityIdsByThread({ threadId })2730 threadId,
2850 : listThreadActivityIdsByThreadWindow({ threadId, ...bounds })2731 activityKinds: query.activityKinds,
2732 })
2851 ).pipe(2733 ).pipe(
2852 Effect.mapError(2734 Effect.mapError(
2853 toPersistenceSqlOrDecodeError(2735 toPersistenceSqlOrDecodeError(
2854 "ProjectionSnapshotQuery.getThreadDetailById:listActivityIds:query",2736 "ProjectionSnapshotQuery.getThreadDetailById:listActivities:query",
2855 "ProjectionSnapshotQuery.getThreadDetailById:listActivityIds:decodeRows",2737 "ProjectionSnapshotQuery.getThreadDetailById:listActivities:decodeRows",
2856 ),2738 ),
2857 ),2739 ),
2858 ),
2859 listPinnedThreadActivityIdsByThread({ threadId }).pipe(
2860 Effect.mapError(
2861 toPersistenceSqlOrDecodeError(
2862 "ProjectionSnapshotQuery.getThreadDetailById:listPinnedActivityIds:query",
2863 "ProjectionSnapshotQuery.getThreadDetailById:listPinnedActivityIds:decodeRows",
2864 ),
2865 ),
2866 ),
2867 ]);
2868 const activityIds = [
2869 ...new Set([...activityIdRows, ...pinnedActivityIdRows].map(({ activityId }) => activityId)),
2870 ];
2871 const activities: OrchestrationThreadActivity[] = [];
2872
2873 for (
2874 let offset = 0;
2875 offset < activityIds.length;
2876 offset += THREAD_DETAIL_ACTIVITY_PAYLOAD_BATCH_SIZE
2877 ) {
2878 const batchIds = activityIds.slice(
2879 offset,
2880 offset + THREAD_DETAIL_ACTIVITY_PAYLOAD_BATCH_SIZE,
2881 );2740 );
2882 const batchRows = yield* listThreadActivityRowsByIds({ activityIds: batchIds }).pipe(2741 const pinnedActivityRowsEffect =
2883 Effect.mapError(2742 query?.activityKinds === undefined
2884 toPersistenceSqlOrDecodeError(2743 ? listPinnedThreadActivityRowsByThread({ threadId }).pipe(
2885 "ProjectionSnapshotQuery.getThreadDetailById:listActivityPayloadBatch:query",2744 Effect.mapError(
2886 "ProjectionSnapshotQuery.getThreadDetailById:listActivityPayloadBatch:decodeRows",2745 toPersistenceSqlOrDecodeError(
2887 ),2746 "ProjectionSnapshotQuery.getThreadDetailById:listPinnedActivities:query",
2888 ),2747 "ProjectionSnapshotQuery.getThreadDetailById:listPinnedActivities:decodeRows",
2889 );
2890 for (const row of batchRows) {
2891 activities.push(projectActivityPayload(mapThreadActivityRow(row)));
2892 }
2893 }
2894
2895 return activities.toSorted(
2896 (left, right) =>
2897 (left.sequence ?? -1) - (right.sequence ?? -1) ||
2898 left.createdAt.localeCompare(right.createdAt) ||
2899 left.id.localeCompare(right.id),
2900 );
2901 });
2902
2903 const getThreadDetailByIdBounded = (
2904 threadId: ThreadId,
2905 bounds: ThreadDetailBounds | undefined,
2906 activityRead: ThreadDetailActivityRead = { mode: "raw" },
2907 ) =>
2908 Effect.gen(function* () {
2909 const activitiesEffect =
2910 activityRead.mode === "client"
2911 ? listProjectedThreadActivities(threadId, bounds)
2912 : Effect.all([
2913 (activityRead.query?.activityKinds === undefined
2914 ? bounds === undefined
2915 ? listThreadActivityRowsByThread({ threadId })
2916 : listThreadActivityRowsByThreadWindow({ threadId, ...bounds })
2917 : activityRead.query.activityKinds.length === 0
2918 ? Effect.succeed([])
2919 : listThreadActivityRowsByThreadAndKinds({
2920 threadId,
2921 activityKinds: activityRead.query.activityKinds,
2922 })
2923 ).pipe(
2924 Effect.mapError(
2925 toPersistenceSqlOrDecodeError(
2926 "ProjectionSnapshotQuery.getThreadDetailById:listActivities:query",
2927 "ProjectionSnapshotQuery.getThreadDetailById:listActivities:decodeRows",
2928 ),
2929 ),2748 ),
2930 ),2749 ),
2931 activityRead.query?.activityKinds === undefined2750 )
2932 ? listPinnedThreadActivityRowsByThread({ threadId }).pipe(2751 : Effect.succeed([]);
2933 Effect.mapError(
2934 toPersistenceSqlOrDecodeError(
2935 "ProjectionSnapshotQuery.getThreadDetailById:listPinnedActivities:query",
2936 "ProjectionSnapshotQuery.getThreadDetailById:listPinnedActivities:decodeRows",
2937 ),
2938 ),
2939 )
2940 : Effect.succeed([]),
2941 ]).pipe(
2942 Effect.map(([activityRows, pinnedActivityRows]) =>
2943 [
2944 ...new Map(
2945 [...activityRows, ...pinnedActivityRows].map(
2946 (row) => [row.activityId, row] as const,
2947 ),
2948 ).values(),
2949 ]
2950 .toSorted(
2951 (left, right) =>
2952 (left.sequence ?? -1) - (right.sequence ?? -1) ||
2953 left.createdAt.localeCompare(right.createdAt) ||
2954 left.activityId.localeCompare(right.activityId),
2955 )
2956 .map(mapThreadActivityRow),
2957 ),
2958 );
2959 2752
2960 const [2753 const [
2961 threadRow,2754 threadRow,
2962 messageRows,2755 messageRows,
2963 proposedPlanRows,2756 proposedPlanRows,
2964 activities,2757 activityRows,
2758 pinnedActivityRows,
2965 checkpointRows,2759 checkpointRows,
2966 latestTurnRow,2760 latestTurnRow,
2967 sessionRow,2761 sessionRow,
@@ -2993,7 +2787,8 @@ pending_approval_requests AS ( · 25 unmodified lines
2993 ),2787 ),
2994 ),2788 ),
2995 ),2789 ),
2996 activitiesEffect,2790 activityRowsEffect,
2791 pinnedActivityRowsEffect,
2997 listCheckpointRowsByThread({ threadId }).pipe(2792 listCheckpointRowsByThread({ threadId }).pipe(
2998 Effect.mapError(2793 Effect.mapError(
2999 toPersistenceSqlOrDecodeError(2794 toPersistenceSqlOrDecodeError(
@@ -3024,6 +2819,17 @@ pending_approval_requests AS ( · 24 unmodified lines
3024 return Option.none<OrchestrationThread>();2819 return Option.none<OrchestrationThread>();
3025 }2820 }
3026 2821
2822 const selectedActivityRows = [
2823 ...new Map(
2824 [...activityRows, ...pinnedActivityRows].map((row) => [row.activityId, row] as const),
2825 ).values(),
2826 ].toSorted(
2827 (left, right) =>
2828 (left.sequence ?? -1) - (right.sequence ?? -1) ||
2829 left.createdAt.localeCompare(right.createdAt) ||
2830 left.activityId.localeCompare(right.activityId),
2831 );
2832
3027 const thread = {2833 const thread = {
3028 id: threadRow.value.threadId,2834 id: threadRow.value.threadId,
3029 projectId: threadRow.value.projectId,2835 projectId: threadRow.value.projectId,
@@ -3061,7 +2867,21 @@ pending_approval_requests AS ( · 31 unmodified lines
3061 return message;2867 return message;
3062 }),2868 }),
3063 proposedPlans: proposedPlanRows.map(mapProposedPlanRow),2869 proposedPlans: proposedPlanRows.map(mapProposedPlanRow),
3064 activities,2870 activities: selectedActivityRows.map((row) => {
2871 const activity = {
2872 id: row.activityId,
2873 tone: row.tone,
2874 kind: row.kind,
2875 summary: row.summary,
2876 payload: row.payload,
2877 turnId: row.turnId,
2878 createdAt: row.createdAt,
2879 };
2880 if (row.sequence !== null) {
2881 return Object.assign(activity, { sequence: row.sequence });
2882 }
2883 return activity;
2884 }),
3065 checkpoints: checkpointRows.map((row) => ({2885 checkpoints: checkpointRows.map((row) => ({
3066 turnId: row.turnId,2886 turnId: row.turnId,
3067 checkpointTurnCount: row.checkpointTurnCount,2887 checkpointTurnCount: row.checkpointTurnCount,
@@ -3086,11 +2906,7 @@ pending_approval_requests AS ( · 18 unmodified lines
3086 const getThreadDetailById: ProjectionSnapshotQueryShape["getThreadDetailById"] = (2906 const getThreadDetailById: ProjectionSnapshotQueryShape["getThreadDetailById"] = (
3087 threadId,2907 threadId,
3088 query,2908 query,
3089 ) =>2909 ) => getThreadDetailByIdBounded(threadId, undefined, query);
3090 getThreadDetailByIdBounded(threadId, undefined, {
3091 mode: "raw",
3092 ...(query === undefined ? {} : { query }),
3093 });
3094 2910
3095 // Bounds pathological fan-out: one user turn that spawned hundreds of2911 // Bounds pathological fan-out: one user turn that spawned hundreds of
3096 // subagent turns still pages in bounded chunks, at the cost of splitting the2912 // subagent turns still pages in bounded chunks, at the cost of splitting the
@@ -3114,9 +2930,7 @@ pending_approval_requests AS ( · 17 unmodified lines
3114 .withTransaction(2930 .withTransaction(
3115 Effect.gen(function* () {2931 Effect.gen(function* () {
3116 if (window?.turnLimit === undefined) {2932 if (window?.turnLimit === undefined) {
3117 const thread = yield* getThreadDetailByIdBounded(threadId, undefined, {2933 const thread = yield* getThreadDetailById(threadId);
3118 mode: "client",
3119 });
3120 if (Option.isNone(thread)) {2934 if (Option.isNone(thread)) {
3121 return Option.none<OrchestrationThreadDetailSnapshot>();2935 return Option.none<OrchestrationThreadDetailSnapshot>();
3122 }2936 }
@@ -3169,9 +2983,7 @@ pending_approval_requests AS ( · 46 unmodified lines
3169 ? { minAnchorAt: "", minTurnKey: "", beforeAnchorAt: "", beforeTurnKey: "" }2983 ? { minAnchorAt: "", minTurnKey: "", beforeAnchorAt: "", beforeTurnKey: "" }
3170 : undefined;2984 : undefined;
3171 2985
3172 const thread = yield* getThreadDetailByIdBounded(threadId, emptyBounds ?? bounds, {2986 const thread = yield* getThreadDetailByIdBounded(threadId, emptyBounds ?? bounds);
3173 mode: "client",
3174 });
3175 if (Option.isNone(thread)) {2987 if (Option.isNone(thread)) {
3176 return Option.none<OrchestrationThreadDetailSnapshot>();2988 return Option.none<OrchestrationThreadDetailSnapshot>();
3177 }2989 }
apps/server/src/orchestration/Services/ProjectionSnapshotQuery.ts+0−4
@@ -227,10 +227,6 @@ export interface ProjectionSnapshotQueryShape { · 226 unmodified lines
227 * response carries `page` metadata (see `OrchestrationThreadDetailWindow`).227 * response carries `page` metadata (see `OrchestrationThreadDetailWindow`).
228 * Without a window the full thread is returned with no `page` field —228 * Without a window the full thread is returned with no `page` field —
229 * pagination is strictly opt-in.229 * pagination is strictly opt-in.
230 *
231 * Activity payloads are projected for clients as they are read in small
232 * sequential batches. Callers still apply the full snapshot projector for
233 * collection-level activity pruning.
234 */230 */
235 readonly getThreadDetailSnapshot: (231 readonly getThreadDetailSnapshot: (
236 threadId: ThreadId,232 threadId: ThreadId,
apps/server/src/orchestration/ThreadLiveEventCoalescer.test.ts+0−250
@@ -1,250 +0,0 @@
1import {
2 EventId,
3 MessageId,
4 ThreadId,
5 TurnId,
6 OrchestrationEvent,
7 type OrchestrationThreadActivity,
8} from "@t3tools/contracts";
9import { it } from "@effect/vitest";
10import * as Clock from "effect/Clock";
11import * as Effect from "effect/Effect";
12import * as Schema from "effect/Schema";
13import * as Stream from "effect/Stream";
14import * as TestClock from "effect/testing/TestClock";
15import { describe, expect } from "vite-plus/test";
16
17import {
18 coalesceLiveToolUpdatedEvents,
19 makeThreadLiveEventCoalescer,
20} from "./ThreadLiveEventCoalescer.ts";
21
22const threadId = ThreadId.make("thread-coalescer-test");
23const turnId = TurnId.make("turn-coalescer-test");
24const encodeEvent = Schema.encodeSync(Schema.fromJsonString(OrchestrationEvent));
25
26function makeToolActivity(
27 sequence: number,
28 options: {
29 readonly kind?: "tool.updated" | "tool.completed";
30 readonly toolCallId?: string;
31 readonly turnId?: TurnId;
32 } = {},
33): OrchestrationEvent {
34 const {
35 kind = "tool.updated",
36 toolCallId = "call-edit",
37 turnId: activityTurnId = turnId,
38 } = options;
39 const activity: OrchestrationThreadActivity = {
40 id: EventId.make(`activity-${sequence}`),
41 tone: "tool",
42 kind,
43 summary: "Editing app.ts",
44 payload: {
45 itemType: "file_change",
46 title: "Editing app.ts",
47 data: toolCallId ? { toolCallId } : {},
48 },
49 turnId: activityTurnId,
50 createdAt: "2026-01-01T00:00:01.000Z",
51 };
52 return {
53 sequence,
54 eventId: EventId.make(`event-${sequence}`),
55 aggregateKind: "thread",
56 aggregateId: threadId,
57 occurredAt: "2026-01-01T00:00:01.000Z",
58 commandId: null,
59 causationEventId: null,
60 correlationId: null,
61 metadata: {},
62 type: "thread.activity-appended",
63 payload: { threadId, activity },
64 };
65}
66
67function makeMessage(sequence: number, text = "Still working"): OrchestrationEvent {
68 return {
69 sequence,
70 eventId: EventId.make(`event-${sequence}`),
71 aggregateKind: "thread",
72 aggregateId: threadId,
73 occurredAt: "2026-01-01T00:00:02.000Z",
74 commandId: null,
75 causationEventId: null,
76 correlationId: null,
77 metadata: {},
78 type: "thread.message-sent",
79 payload: {
80 threadId,
81 messageId: MessageId.make(`message-${sequence}`),
82 role: "assistant",
83 text,
84 turnId,
85 streaming: false,
86 createdAt: "2026-01-01T00:00:02.000Z",
87 updatedAt: "2026-01-01T00:00:02.000Z",
88 },
89 };
90}
91
92describe("ThreadLiveEventCoalescer", () => {
93 it("coalesces only calls with a stable toolCallId", () => {
94 const events = [
95 makeToolActivity(1, { toolCallId: "call-a" }),
96 makeToolActivity(2, { toolCallId: "call-b" }),
97 makeToolActivity(3, { toolCallId: "call-a" }),
98 ];
99
100 expect(coalesceLiveToolUpdatedEvents(events).map((event) => event.sequence)).toEqual([2, 3]);
101 });
102
103 it("preserves parallel same-label calls without a stable toolCallId", () => {
104 const events = [
105 makeToolActivity(1, { toolCallId: "" }),
106 makeToolActivity(2, { toolCallId: "" }),
107 makeToolActivity(3, { kind: "tool.completed", toolCallId: "" }),
108 ];
109
110 expect(coalesceLiveToolUpdatedEvents(events).map((event) => event.sequence)).toEqual([1, 2, 3]);
111 });
112
113 it("does not coalesce stable tool calls across turns", () => {
114 const events = [
115 makeToolActivity(1, { turnId: TurnId.make("turn-old") }),
116 makeToolActivity(2, { turnId: TurnId.make("turn-new") }),
117 ];
118
119 expect(coalesceLiveToolUpdatedEvents(events).map((event) => event.sequence)).toEqual([1, 2]);
120 });
121
122 it("flushes a stable update run before a completion boundary", () => {
123 const events = [
124 makeToolActivity(1),
125 makeToolActivity(2),
126 makeToolActivity(3, { kind: "tool.completed" }),
127 makeToolActivity(4),
128 ];
129
130 expect(coalesceLiveToolUpdatedEvents(events).map((event) => event.sequence)).toEqual([2, 3, 4]);
131 });
132
133 it.effect("flushes pending tool updates as soon as an unrelated event arrives", () =>
134 Effect.scoped(
135 Effect.gen(function* () {
136 const coalescer = yield* makeThreadLiveEventCoalescer({ coalesceWindow: "500 millis" });
137 const startedAt = yield* Clock.currentTimeMillis;
138 yield* Effect.forEach(
139 Array.from({ length: 10 }, (_, index) => index + 2),
140 (sequence) => coalescer.offer({ kind: "event", event: makeToolActivity(sequence) }),
141 { discard: true },
142 );
143 yield* coalescer.offer({ kind: "event", event: makeMessage(12) });
144
145 expect(yield* Clock.currentTimeMillis).toBe(startedAt);
146 expect(
147 (yield* coalescer.stream.pipe(Stream.take(2), Stream.runCollect)).map((item) =>
148 item.kind === "event" ? item.event.sequence : item.kind,
149 ),
150 ).toEqual([11, 12]);
151 }),
152 ).pipe(Effect.provide(TestClock.layer())),
153 );
154
155 it.effect("flushes pending tool updates as soon as a synchronization marker arrives", () =>
156 Effect.scoped(
157 Effect.gen(function* () {
158 const coalescer = yield* makeThreadLiveEventCoalescer({ coalesceWindow: "500 millis" });
159 const startedAt = yield* Clock.currentTimeMillis;
160 yield* coalescer.offer({ kind: "event", event: makeToolActivity(2) });
161 yield* coalescer.offer({ kind: "event", event: makeToolActivity(3) });
162 yield* coalescer.offer({ kind: "synchronized" });
163
164 expect(yield* Clock.currentTimeMillis).toBe(startedAt);
165 expect(
166 (yield* coalescer.stream.pipe(Stream.take(2), Stream.runCollect)).map((item) =>
167 item.kind === "event" ? item.event.sequence : item.kind,
168 ),
169 ).toEqual([3, "synchronized"]);
170 }),
171 ).pipe(Effect.provide(TestClock.layer())),
172 );
173
174 it.effect("fails and clears pending updates when their serialized payload fills the budget", () =>
175 Effect.scoped(
176 Effect.gen(function* () {
177 const first = makeToolActivity(1);
178 const coalescer = yield* makeThreadLiveEventCoalescer({
179 coalesceWindow: "500 millis",
180 maxSerializedBytes: Buffer.byteLength(encodeEvent(first)),
181 });
182 yield* coalescer.offer({ kind: "event", event: first });
183 expect(yield* coalescer.usage).toEqual({
184 retainedItems: 1,
185 retainedSerializedBytes: Buffer.byteLength(encodeEvent(first)),
186 });
187
188 const overflow = yield* coalescer
189 .offer({ kind: "event", event: makeToolActivity(2) })
190 .pipe(Effect.result);
191 expect(overflow._tag).toBe("Failure");
192 yield* coalescer.closed;
193 expect(yield* coalescer.usage).toEqual({ retainedItems: 0, retainedSerializedBytes: 0 });
194 const marker = yield* coalescer.offer({ kind: "synchronized" }).pipe(Effect.result);
195 expect(marker._tag).toBe("Failure");
196 const delivered = yield* coalescer.stream.pipe(Stream.runCollect, Effect.result);
197 expect(delivered._tag).toBe("Failure");
198 }),
199 ),
200 );
201
202 it.effect("keeps the flush timer alive when an offer's shorter scope closes", () =>
203 Effect.scoped(
204 Effect.gen(function* () {
205 const coalescer = yield* makeThreadLiveEventCoalescer({ coalesceWindow: "50 millis" });
206 yield* Effect.scoped(coalescer.offer({ kind: "event", event: makeToolActivity(1) }));
207 yield* TestClock.adjust("50 millis");
208 const items = yield* coalescer.stream.pipe(Stream.take(1), Stream.runCollect);
209 expect(
210 items.map((item) => (item.kind === "event" ? item.event.sequence : item.kind)),
211 ).toEqual([1]);
212 }),
213 ),
214 );
215
216 it.effect("keeps an unacknowledged batch charged and clears later events on overflow", () =>
217 Effect.scoped(
218 Effect.gen(function* () {
219 const coalescer = yield* makeThreadLiveEventCoalescer({ maxItems: 3 });
220 const first = makeMessage(1, "é".repeat(1_024));
221 yield* coalescer.offer({ kind: "event", event: first });
222
223 yield* Effect.scoped(
224 Effect.gen(function* () {
225 const pull = yield* Stream.toPull(coalescer.stream);
226 const batch = yield* pull;
227 expect(
228 batch.map((item) => (item.kind === "event" ? item.event.sequence : null)),
229 ).toEqual([1]);
230 yield* coalescer.offer({ kind: "event", event: makeMessage(2) });
231 yield* coalescer.offer({ kind: "event", event: makeToolActivity(3) });
232 expect((yield* coalescer.usage).retainedItems).toBe(3);
233
234 const overflow = yield* coalescer
235 .offer({ kind: "event", event: makeToolActivity(4, { kind: "tool.completed" }) })
236 .pipe(Effect.result);
237 expect(overflow._tag).toBe("Failure");
238 // Do not pull or acknowledge the batch. Cleanup must still finish.
239 yield* coalescer.closed;
240 expect(yield* coalescer.usage).toEqual({
241 retainedItems: 1,
242 retainedSerializedBytes: Buffer.byteLength(encodeEvent(first)),
243 });
244 }),
245 );
246 expect(yield* coalescer.usage).toEqual({ retainedItems: 0, retainedSerializedBytes: 0 });
247 }),
248 ),
249 );
250});
apps/server/src/orchestration/ThreadLiveEventCoalescer.ts+0−241
@@ -1,241 +0,0 @@
1import type {
2 OrchestrationEvent,
3 OrchestrationGetSnapshotError,
4 OrchestrationThreadStreamItem,
5} from "@t3tools/contracts";
6import * as Deferred from "effect/Deferred";
7import * as Duration from "effect/Duration";
8import * as Effect from "effect/Effect";
9import * as Fiber from "effect/Fiber";
10import * as Predicate from "effect/Predicate";
11import * as Queue from "effect/Queue";
12import * as Semaphore from "effect/Semaphore";
13import * as Stream from "effect/Stream";
14
15import { projectActivityEvent } from "./ActivityPayloadProjection.ts";
16import { makeLiveStreamBudget, type RetainedLiveItem } from "./LiveStreamBudget.ts";
17
18const COALESCE_WINDOW = Duration.millis(50);
19const MAX_PENDING_UPDATES = 512;
20
21export type ThreadLiveInput =
22 | { readonly kind: "event"; readonly event: OrchestrationEvent }
23 | { readonly kind: "synchronized" };
24
25function isToolUpdated(event: OrchestrationEvent): boolean {
26 return (
27 event.type === "thread.activity-appended" && event.payload.activity.kind === "tool.updated"
28 );
29}
30
31function asTrimmedString(value: unknown): string | null {
32 if (!Predicate.isString(value)) {
33 return null;
34 }
35 const trimmed = value.trim();
36 return trimmed.length > 0 ? trimmed : null;
37}
38
39function stableToolCallIdentity(event: OrchestrationEvent): string | null {
40 if (event.type !== "thread.activity-appended") {
41 return null;
42 }
43 const payload = event.payload.activity.payload;
44 if (!Predicate.isObject(payload)) {
45 return null;
46 }
47 const data = Predicate.isObject(payload.data) ? payload.data : null;
48 return asTrimmedString(payload.toolCallId) ?? asTrimmedString(data?.toolCallId);
49}
50
51/**
52 * Retain only the latest in-flight update for each stable tool-call id in a
53 * live run. Anonymous calls pass through because labels are not unique when
54 * tools execute in parallel. Survivors remain in sequence order.
55 */
56export function coalesceLiveToolUpdatedEvents(
57 events: ReadonlyArray<OrchestrationEvent>,
58): ReadonlyArray<OrchestrationEvent> {
59 const survivors: Array<OrchestrationEvent> = [];
60 let pendingUpdates: Array<OrchestrationEvent> = [];
61
62 const flushUpdates = () => {
63 const seen = new Set<string>();
64 const latestUpdates: Array<OrchestrationEvent> = [];
65 for (let index = pendingUpdates.length - 1; index >= 0; index -= 1) {
66 const event = pendingUpdates[index]!;
67 const identity = stableToolCallIdentity(event);
68 const activity =
69 event.type === "thread.activity-appended" ? event.payload.activity : undefined;
70 const key = identity ? `${activity?.turnId ?? ""}\u0000${identity}` : null;
71 if (key && seen.has(key)) {
72 continue;
73 }
74 if (key) {
75 seen.add(key);
76 }
77 latestUpdates.push(event);
78 }
79 latestUpdates.reverse();
80 survivors.push(...latestUpdates);
81 pendingUpdates = [];
82 };
83
84 for (const event of events) {
85 if (isToolUpdated(event)) {
86 pendingUpdates.push(event);
87 continue;
88 }
89 flushUpdates();
90 survivors.push(event);
91 }
92 flushUpdates();
93 return survivors;
94}
95
96export const makeThreadLiveEventCoalescer = Effect.fn("makeThreadLiveEventCoalescer")(
97 function* (options?: {
98 readonly coalesceWindow?: Duration.Input;
99 readonly maxItems?: number;
100 readonly maxSerializedBytes?: number;
101 }) {
102 const coalescerScope = yield* Effect.scope;
103 const budget = yield* makeLiveStreamBudget(options);
104 const cleanupComplete = yield* Deferred.make<void>();
105 const output = yield* Queue.unbounded<
106 RetainedLiveItem<OrchestrationThreadStreamItem>,
107 OrchestrationGetSnapshotError
108 >();
109 const mutex = yield* Semaphore.make(1);
110 const coalesceWindow = options?.coalesceWindow ?? COALESCE_WINDOW;
111 let pendingUpdates: Array<RetainedLiveItem<OrchestrationEvent>> = [];
112 let windowGeneration = 0;
113 let windowFiber: Fiber.Fiber<void, never> | null = null;
114 let closed = false;
115
116 const cancelWindow = Effect.fn("ThreadLiveEventCoalescer.cancelWindow")(function* () {
117 const fiber = windowFiber;
118 if (!fiber) {
119 return;
120 }
121 windowFiber = null;
122 yield* Fiber.interrupt(fiber);
123 });
124
125 const flushPending = Effect.fn("ThreadLiveEventCoalescer.flushPending")(function* () {
126 if (pendingUpdates.length === 0) {
127 return;
128 }
129 const items = yield* budget.replace(
130 pendingUpdates,
131 coalesceLiveToolUpdatedEvents(pendingUpdates.map((item) => item.value)).map((event) => ({
132 kind: "event" as const,
133 event,
134 })),
135 (item) => item.event,
136 );
137 pendingUpdates = [];
138 yield* Queue.offerAll(output, items);
139 }, Effect.uninterruptible);
140
141 const flushWindow = (generation: number) =>
142 Effect.sleep(coalesceWindow).pipe(
143 Effect.andThen(
144 mutex.withPermits(1)(
145 Effect.suspend(() => (generation === windowGeneration ? flushPending() : Effect.void)),
146 ),
147 ),
148 Effect.ensuring(
149 Effect.sync(() => {
150 if (generation === windowGeneration) {
151 windowFiber = null;
152 }
153 }),
154 ),
155 Effect.catchTags({ OrchestrationGetSnapshotError: () => Effect.void }),
156 );
157
158 // Keep each source batch together so a synchronization marker cannot pass
159 // events already pulled from PubSub but still being coalesced.
160 const offerAll = Effect.fn("ThreadLiveEventCoalescer.offerAll")(function* (
161 inputs: ReadonlyArray<ThreadLiveInput>,
162 ) {
163 yield* mutex.withPermits(1)(
164 Effect.forEach(
165 inputs,
166 (input) =>
167 Effect.gen(function* () {
168 yield* budget.check;
169 if (input.kind === "event") {
170 // Retain only the client payload, not full persisted tool output.
171 yield* budget.retain(projectActivityEvent(input.event)).pipe(
172 Effect.tap((item) => Effect.sync(() => pendingUpdates.push(item))),
173 Effect.uninterruptible,
174 );
175 }
176 if (input.kind === "event" && isToolUpdated(input.event)) {
177 if (pendingUpdates.length === 1) {
178 const generation = ++windowGeneration;
179 windowFiber = yield* Effect.forkIn(flushWindow(generation), coalescerScope);
180 }
181 if (pendingUpdates.length >= MAX_PENDING_UPDATES) {
182 yield* cancelWindow();
183 windowGeneration += 1;
184 yield* flushPending();
185 }
186 return;
187 }
188
189 yield* cancelWindow();
190 windowGeneration += 1;
191 // A non-update event closes the run immediately. The coalescer keeps
192 // that boundary after the final update from the run.
193 yield* flushPending();
194 if (input.kind === "synchronized") {
195 yield* budget.retain({ kind: "synchronized" as const }).pipe(
196 Effect.flatMap((marker) => Queue.offer(output, marker)),
197 Effect.uninterruptible,
198 );
199 }
200 }),
201 { discard: true },
202 ),
203 );
204 });
205
206 const close = (error?: OrchestrationGetSnapshotError) =>
207 mutex.withPermits(1)(
208 Effect.gen(function* () {
209 if (closed) {
210 return;
211 }
212 closed = true;
213 windowGeneration += 1;
214 yield* cancelWindow();
215 budget.release(pendingUpdates);
216 pendingUpdates = [];
217 budget.release(yield* Queue.clear(output).pipe(Effect.orDie));
218 if (error) {
219 yield* Queue.fail(output, error);
220 }
221 yield* Queue.shutdown(output);
222 yield* Deferred.succeed(cleanupComplete, undefined);
223 }),
224 );
225
226 yield* Effect.addFinalizer(() => close());
227 yield* budget.failed.pipe(
228 Effect.catchTags({ OrchestrationGetSnapshotError: close }),
229 Effect.forkScoped,
230 );
231
232 return {
233 offer: (input: ThreadLiveInput) => offerAll([input]),
234 offerAll,
235 stream: budget.deliver(Stream.fromQueue(output)),
236 failed: budget.failed,
237 closed: Deferred.await(cleanupComplete),
238 usage: budget.usage,
239 } as const;
240 },
241);
apps/server/src/orchestration/ThreadSettlementReactor.test.ts+0−858
@@ -1,858 +0,0 @@
1import {
2 DEFAULT_SERVER_SETTINGS,
3 ProjectId,
4 ProviderInstanceId,
5 PullRequestOperationError,
6 ThreadId,
7 type OrchestrationCommand,
8 type OrchestrationProjectShell,
9 type OrchestrationShellSnapshot,
10 type OrchestrationThreadShell,
11 type PullRequestSummary,
12 type ServerSettings,
13 type ServerSettingsPatch,
14} from "@t3tools/contracts";
15import { applyServerSettingsPatch } from "@t3tools/shared/serverSettings";
16import { assert, describe, it } from "@effect/vitest";
17import * as Crypto from "effect/Crypto";
18import * as Deferred from "effect/Deferred";
19import * as Effect from "effect/Effect";
20import * as FileSystem from "effect/FileSystem";
21import * as Layer from "effect/Layer";
22import * as PubSub from "effect/PubSub";
23import * as Queue from "effect/Queue";
24import * as Ref from "effect/Ref";
25import * as Stream from "effect/Stream";
26import { TestClock } from "effect/testing";
27
28import { GitManager } from "../git/GitManager.ts";
29import {
30 PullRequestService,
31 type PullRequestMergeEvent,
32} from "../pullRequest/PullRequestService.ts";
33import { ServerActivation } from "../serverActivation.ts";
34import { ServerSettingsService } from "../serverSettings.ts";
35import { OrchestrationCommandInvariantError } from "./Errors.ts";
36import {
37 OrchestrationEngineService,
38 type OrchestrationEngineShape,
39} from "./Services/OrchestrationEngine.ts";
40import { ProjectionSnapshotQuery } from "./Services/ProjectionSnapshotQuery.ts";
41import * as ThreadSettlementReactor from "./ThreadSettlementReactor.ts";
42
43const NOW = "2026-08-28T12:00:00.000Z";
44const PROJECT_ID = ProjectId.make("settlement-project");
45const LINKED_PROJECT_ID = ProjectId.make("linked-settlement-project");
46
47type AutoSettleCommand = Extract<OrchestrationCommand, { readonly type: "thread.auto-settle" }>;
48
49const testCrypto = Crypto.make({
50 randomBytes: (size) => new Uint8Array(size).fill(1),
51 digest: (_algorithm, data) => Effect.succeed(data),
52});
53
54function makeProject(
55 id: ProjectId = PROJECT_ID,
56 workspaceRoot = "/workspace/project",
57): OrchestrationProjectShell {
58 return {
59 id,
60 title: `Project ${id}`,
61 workspaceRoot,
62 defaultModelSelection: null,
63 scripts: [],
64 createdAt: "2026-08-01T00:00:00.000Z",
65 updatedAt: NOW,
66 };
67}
68
69function makeThread(
70 id: string,
71 overrides: Partial<OrchestrationThreadShell> = {},
72): OrchestrationThreadShell {
73 return {
74 id: ThreadId.make(id),
75 projectId: PROJECT_ID,
76 title: id,
77 modelSelection: {
78 instanceId: ProviderInstanceId.make("codex"),
79 model: "gpt-5",
80 },
81 runtimeMode: "full-access",
82 interactionMode: "default",
83 branch: null,
84 worktreePath: null,
85 latestTurn: null,
86 createdAt: "2026-08-01T00:00:00.000Z",
87 updatedAt: "2026-08-20T00:00:00.000Z",
88 archivedAt: null,
89 settledOverride: null,
90 settledAt: null,
91 session: null,
92 latestUserMessageAt: "2026-08-20T00:00:00.000Z",
93 hasPendingApprovals: false,
94 hasPendingUserInput: false,
95 hasActionableProposedPlan: false,
96 ...overrides,
97 };
98}
99
100function makeSnapshot(
101 threads: ReadonlyArray<OrchestrationThreadShell>,
102 projects: ReadonlyArray<OrchestrationProjectShell> = [makeProject()],
103): OrchestrationShellSnapshot {
104 return {
105 snapshotSequence: 1,
106 projects,
107 threads,
108 updatedAt: NOW,
109 };
110}
111
112function makePullRequestSummary(input: {
113 readonly projectId: ProjectId;
114 readonly repository: string;
115 readonly number: number;
116 readonly state: "open" | "closed" | "merged";
117 readonly updatedAt?: string;
118}): PullRequestSummary {
119 return {
120 provider: "github",
121 projectId: input.projectId,
122 repository: input.repository,
123 number: input.number,
124 title: "Pull request",
125 url: `https://example.test/${input.repository}/pull/${input.number}`,
126 state: input.state,
127 headBranch: "feature",
128 baseBranch: "main",
129 updatedAt: input.updatedAt ?? NOW,
130 };
131}
132
133interface HarnessOptions {
134 readonly snapshot: OrchestrationShellSnapshot;
135 readonly settings?: ServerSettings;
136 readonly branchPullRequest?: GitManager["Service"]["branchPullRequest"];
137 readonly pullRequestSummary?: PullRequestService["Service"]["summary"];
138 readonly existingWorktreePaths?: ReadonlyArray<string>;
139 readonly onDispatch?: (
140 command: AutoSettleCommand,
141 ) => Effect.Effect<void, OrchestrationCommandInvariantError>;
142}
143
144const makeHarness = Effect.fn("makeThreadSettlementHarness")(function* (options: HarnessOptions) {
145 const activation = yield* Deferred.make<void>();
146 const snapshots = yield* Ref.make(options.snapshot);
147 const snapshotReadCount = yield* Ref.make(0);
148 const snapshotReads = yield* Queue.unbounded<number>();
149 const settings = yield* Ref.make(options.settings ?? DEFAULT_SERVER_SETTINGS);
150 const settingsChanges = yield* PubSub.unbounded<ServerSettings>();
151 const mergedPullRequests = yield* PubSub.unbounded<PullRequestMergeEvent>();
152 const commands = yield* Ref.make<ReadonlyArray<AutoSettleCommand>>([]);
153 const branchCalls = yield* Ref.make<
154 ReadonlyArray<{ readonly cwd: string; readonly branch: string }>
155 >([]);
156 const summaryCalls = yield* Ref.make<
157 ReadonlyArray<{
158 readonly projectId: ProjectId;
159 readonly repository: string;
160 readonly number: number;
161 }>
162 >([]);
163 const summaryRecovery = yield* Ref.make<ReadonlyArray<boolean | undefined>>([]);
164 const invalidatedCwds = yield* Ref.make<ReadonlyArray<string>>([]);
165
166 const updateSettings = (patch: ServerSettingsPatch) =>
167 Effect.gen(function* () {
168 const next = applyServerSettingsPatch(yield* Ref.get(settings), patch);
169 yield* Ref.set(settings, next);
170 yield* PubSub.publish(settingsChanges, next);
171 return next;
172 });
173
174 const branchPullRequest: GitManager["Service"]["branchPullRequest"] = (input) =>
175 Ref.update(branchCalls, (calls) => [...calls, input]).pipe(
176 Effect.andThen(options.branchPullRequest?.(input) ?? Effect.succeed(null)),
177 );
178 const pullRequestSummary: PullRequestService["Service"]["summary"] = (input, readOptions) =>
179 Effect.gen(function* () {
180 yield* Ref.update(summaryCalls, (calls) => [...calls, input]);
181 yield* Ref.update(summaryRecovery, (values) => [
182 ...values,
183 readOptions?.recoverTransientFailure,
184 ]);
185 return yield* (
186 options.pullRequestSummary?.(input, readOptions) ??
187 Effect.succeed(
188 makePullRequestSummary({
189 ...input,
190 state: "open",
191 }),
192 )
193 );
194 });
195
196 const dispatch: OrchestrationEngineShape["dispatch"] = (command) => {
197 if (command.type !== "thread.auto-settle") {
198 return Effect.die(new Error(`Unexpected command: ${command.type}`));
199 }
200 return Ref.update(commands, (recorded) => [...recorded, command]).pipe(
201 Effect.andThen(options.onDispatch?.(command) ?? Effect.void),
202 Effect.as({ sequence: 1 }),
203 );
204 };
205
206 const serverSettings = ServerSettingsService.of({
207 start: Effect.void,
208 ready: Effect.void,
209 getSettings: Ref.get(settings),
210 updateSettings,
211 streamChanges: Stream.fromPubSub(settingsChanges),
212 subscribeChanges: PubSub.subscribe(settingsChanges).pipe(
213 Effect.map((subscription) => Stream.fromSubscription(subscription)),
214 ),
215 });
216
217 const dependencies = Layer.mergeAll(
218 Layer.mock(ProjectionSnapshotQuery)({
219 getShellSnapshot: () =>
220 Ref.updateAndGet(snapshotReadCount, (count) => count + 1).pipe(
221 Effect.tap((count) => Queue.offer(snapshotReads, count)),
222 Effect.andThen(Ref.get(snapshots)),
223 ),
224 }),
225 Layer.mock(GitManager)({
226 branchPullRequest,
227 invalidateStatus: (cwd) => Ref.update(invalidatedCwds, (cwds) => [...cwds, cwd]),
228 }),
229 Layer.mock(PullRequestService)({
230 summary: pullRequestSummary,
231 subscribeMerges: PubSub.subscribe(mergedPullRequests).pipe(
232 Effect.map((subscription) => Stream.fromSubscription(subscription)),
233 ),
234 }),
235 Layer.mock(OrchestrationEngineService)({
236 readEvents: () => Stream.empty,
237 dispatch,
238 streamDomainEvents: Stream.empty,
239 latestSequence: Effect.succeed(0),
240 }),
241 Layer.succeed(ServerSettingsService, serverSettings),
242 Layer.succeed(ServerActivation, Deferred.await(activation)),
243 Layer.succeed(Crypto.Crypto, testCrypto),
244 FileSystem.layerNoop({
245 exists: (path) => Effect.succeed(options.existingWorktreePaths?.includes(path) ?? false),
246 }),
247 );
248
249 return {
250 activation,
251 snapshots,
252 snapshotReadCount,
253 snapshotReads,
254 commands,
255 branchCalls,
256 summaryCalls,
257 summaryRecovery,
258 invalidatedCwds,
259 updateSettings,
260 publishMerge: PubSub.publish(mergedPullRequests, {
261 projectId: PROJECT_ID,
262 repository: "owner/repository",
263 number: 42,
264 mergedAt: NOW,
265 }),
266 layer: ThreadSettlementReactor.layer.pipe(Layer.provide(dependencies)),
267 };
268});
269
270const startHarness = Effect.fn("startThreadSettlementHarness")(function* (
271 reactor: ThreadSettlementReactor.ThreadSettlementReactor["Service"],
272 activation: Deferred.Deferred<void>,
273 snapshotReads: Queue.Queue<number>,
274) {
275 yield* reactor.start();
276 yield* Deferred.succeed(activation, undefined);
277 yield* Queue.take(snapshotReads);
278 yield* reactor.drain;
279});
280
281describe("ThreadSettlementReactor", () => {
282 it.effect("starts without clients and skips protected threads before pull request lookup", () =>
283 Effect.scoped(
284 Effect.gen(function* () {
285 yield* TestClock.setTime(Date.parse(NOW));
286 const linkedPullRequest = {
287 projectId: LINKED_PROJECT_ID,
288 repository: "owner/repository",
289 number: 42,
290 url: "https://example.test/owner/repository/pull/42",
291 } as const;
292 const skipped = [
293 makeThread("pending-approval", {
294 branch: "skip-approval",
295 hasPendingApprovals: true,
296 }),
297 makeThread("snoozed", {
298 branch: "skip-snoozed",
299 snoozedUntil: "2026-08-29T00:00:00.000Z",
300 }),
301 ];
302 const fixture = yield* makeHarness({
303 snapshot: makeSnapshot(
304 [
305 makeThread("inactive", { branch: "inactive-feature" }),
306 makeThread("closed-pr", { linkedPullRequest }),
307 ...skipped,
308 ],
309 [makeProject(), makeProject(LINKED_PROJECT_ID, "/workspace/linked")],
310 ),
311 branchPullRequest: () => Effect.succeed(null),
312 pullRequestSummary: (input) =>
313 Effect.succeed(makePullRequestSummary({ ...input, state: "closed" })),
314 });
315
316 yield* Effect.gen(function* () {
317 const reactor = yield* ThreadSettlementReactor.ThreadSettlementReactor;
318 yield* reactor.start();
319 assert.strictEqual(yield* Ref.get(fixture.snapshotReadCount), 0);
320
321 yield* Deferred.succeed(fixture.activation, undefined);
322 yield* Queue.take(fixture.snapshotReads);
323 yield* reactor.drain;
324
325 const commands = yield* Ref.get(fixture.commands);
326 assert.deepStrictEqual(
327 commands
328 .map(({ threadId, snapshotSequence, settledAt }) => ({
329 threadId,
330 snapshotSequence,
331 settledAt,
332 }))
333 .sort((left, right) => left.threadId.localeCompare(right.threadId)),
334 [
335 {
336 threadId: ThreadId.make("closed-pr"),
337 snapshotSequence: 1,
338 settledAt: "2026-08-20T00:00:00.000Z",
339 },
340 {
341 threadId: ThreadId.make("inactive"),
342 snapshotSequence: 1,
343 settledAt: "2026-08-20T00:00:00.000Z",
344 },
345 ],
346 );
347 assert.deepStrictEqual(yield* Ref.get(fixture.branchCalls), [
348 { cwd: "/workspace/project", branch: "inactive-feature" },
349 ]);
350 assert.deepStrictEqual(yield* Ref.get(fixture.summaryCalls), [
351 { projectId: LINKED_PROJECT_ID, repository: "owner/repository", number: 42 },
352 ]);
353 assert.deepStrictEqual(yield* Ref.get(fixture.summaryRecovery), [false]);
354 }).pipe(Effect.provide(fixture.layer));
355 }),
356 ),
357 );
358
359 it.effect("reevaluates inactivity and pull request state once per minute", () =>
360 Effect.scoped(
361 Effect.gen(function* () {
362 yield* TestClock.setTime(Date.parse(NOW));
363 const pullRequest = yield* Ref.make<"open" | "merged">("open");
364 const fixture = yield* makeHarness({
365 snapshot: makeSnapshot([
366 makeThread("at-boundary", {
367 latestUserMessageAt: "2026-08-25T12:00:00.000Z",
368 }),
369 makeThread("open-pr", {
370 branch: "saved-feature",
371 latestUserMessageAt: "2026-08-27T00:00:00.000Z",
372 }),
373 ]),
374 branchPullRequest: () =>
375 Ref.get(pullRequest).pipe(Effect.map((state) => ({ state, updatedAt: NOW }))),
376 });
377
378 yield* Effect.gen(function* () {
379 const reactor = yield* ThreadSettlementReactor.ThreadSettlementReactor;
380 yield* startHarness(reactor, fixture.activation, fixture.snapshotReads);
381 assert.deepStrictEqual(yield* Ref.get(fixture.commands), []);
382
383 yield* Ref.set(pullRequest, "merged");
384 yield* TestClock.adjust("1 minute");
385 yield* Queue.take(fixture.snapshotReads);
386 yield* reactor.drain;
387
388 assert.deepStrictEqual(
389 (yield* Ref.get(fixture.commands))
390 .map((command) => command.threadId)
391 .sort((left, right) => left.localeCompare(right)),
392 [ThreadId.make("at-boundary"), ThreadId.make("open-pr")],
393 );
394 assert.strictEqual((yield* Ref.get(fixture.branchCalls)).length, 2);
395 }).pipe(Effect.provide(fixture.layer));
396 }),
397 ),
398 );
399
400 it.effect("reevaluates immediately after a pull request merge", () =>
401 Effect.scoped(
402 Effect.gen(function* () {
403 yield* TestClock.setTime(Date.parse(NOW));
404 const periodicLookupStarted = yield* Deferred.make<void>();
405 const releasePeriodicLookup = yield* Deferred.make<void>();
406 const mergedThreadSettled = yield* Deferred.make<void>();
407 const branchLookupCount = yield* Ref.make(0);
408 const fixture = yield* makeHarness({
409 snapshot: makeSnapshot([
410 makeThread("merged-in-app", {
411 latestUserMessageAt: "2026-08-27T00:00:00.000Z",
412 linkedPullRequest: {
413 projectId: PROJECT_ID,
414 repository: "owner/repository",
415 number: 42,
416 url: "https://example.test/owner/repository/pull/42",
417 },
418 }),
419 makeThread("slow-periodic-lookup", {
420 branch: "another-feature",
421 latestUserMessageAt: "2026-08-27T00:00:00.000Z",
422 }),
423 ]),
424 branchPullRequest: () =>
425 Ref.updateAndGet(branchLookupCount, (count) => count + 1).pipe(
426 Effect.flatMap((count) =>
427 count === 1
428 ? Effect.succeed({ state: "open" as const, updatedAt: NOW })
429 : Deferred.succeed(periodicLookupStarted, undefined).pipe(
430 Effect.andThen(Deferred.await(releasePeriodicLookup)),
431 Effect.as({ state: "open" as const, updatedAt: NOW }),
432 ),
433 ),
434 ),
435 onDispatch: () => Deferred.succeed(mergedThreadSettled, undefined),
436 });
437
438 yield* Effect.gen(function* () {
439 const reactor = yield* ThreadSettlementReactor.ThreadSettlementReactor;
440 yield* startHarness(reactor, fixture.activation, fixture.snapshotReads);
441 yield* fixture.updateSettings({ sidebarAutoSettleAfterDays: 4 });
442 yield* Deferred.await(periodicLookupStarted);
443
444 yield* fixture.publishMerge;
445 yield* Deferred.await(mergedThreadSettled);
446
447 assert.deepStrictEqual(
448 (yield* Ref.get(fixture.commands)).map((command) => command.threadId),
449 [ThreadId.make("merged-in-app")],
450 );
451 yield* Deferred.succeed(releasePeriodicLookup, undefined);
452 yield* reactor.drain;
453 }).pipe(Effect.provide(fixture.layer));
454 }),
455 ),
456 );
457
458 it.effect(
459 "settles branch threads on a pull request merge without waiting for the next sweep",
460 () =>
461 Effect.scoped(
462 Effect.gen(function* () {
463 yield* TestClock.setTime(Date.parse(NOW));
464 const state = yield* Ref.make<"open" | "merged">("open");
465 const mergedThreadSettled = yield* Deferred.make<void>();
466 const fixture = yield* makeHarness({
467 snapshot: makeSnapshot([
468 makeThread("branch-thread", {
469 branch: "saved-feature",
470 latestUserMessageAt: "2026-08-27T00:00:00.000Z",
471 }),
472 ]),
473 branchPullRequest: () =>
474 Ref.get(state).pipe(
475 Effect.map((pullRequestState) => ({ state: pullRequestState, updatedAt: NOW })),
476 ),
477 onDispatch: () => Deferred.succeed(mergedThreadSettled, undefined),
478 });
479
480 yield* Effect.gen(function* () {
481 const reactor = yield* ThreadSettlementReactor.ThreadSettlementReactor;
482 yield* startHarness(reactor, fixture.activation, fixture.snapshotReads);
483 assert.deepStrictEqual(yield* Ref.get(fixture.commands), []);
484 assert.deepStrictEqual(yield* Ref.get(fixture.invalidatedCwds), []);
485
486 yield* Ref.set(state, "merged");
487 yield* fixture.publishMerge;
488 yield* Deferred.await(mergedThreadSettled);
489
490 assert.deepStrictEqual(
491 (yield* Ref.get(fixture.commands)).map((command) => command.threadId),
492 [ThreadId.make("branch-thread")],
493 );
494 assert.deepStrictEqual(yield* Ref.get(fixture.invalidatedCwds), ["/workspace/project"]);
495 yield* reactor.drain;
496 }).pipe(Effect.provide(fixture.layer));
497 }),
498 ),
499 );
500
501 it.effect("a merge does not settle threads linked to an unrelated pull request", () =>
502 Effect.scoped(
503 Effect.gen(function* () {
504 yield* TestClock.setTime(Date.parse(NOW));
505 const mergedThreadSettled = yield* Deferred.make<void>();
506 const mergeLookupStarted = yield* Deferred.make<void>();
507 const releaseMergeLookup = yield* Deferred.make<void>();
508 const lookupCount = yield* Ref.make(0);
509 const fixture = yield* makeHarness({
510 snapshot: makeSnapshot([
511 makeThread("merged-in-app", {
512 latestUserMessageAt: "2026-08-27T00:00:00.000Z",
513 linkedPullRequest: {
514 projectId: PROJECT_ID,
515 repository: "owner/repository",
516 number: 42,
517 url: "https://example.test/owner/repository/pull/42",
518 },
519 }),
520 makeThread("unrelated-linked", {
521 latestUserMessageAt: "2026-08-27T00:00:00.000Z",
522 linkedPullRequest: {
523 projectId: PROJECT_ID,
524 repository: "owner/repository",
525 number: 99,
526 url: "https://example.test/owner/repository/pull/99",
527 },
528 }),
529 ]),
530 pullRequestSummary: (input) =>
531 Ref.updateAndGet(lookupCount, (count) => count + 1).pipe(
532 // The initial sweep looks up both linked threads; the merge
533 // sweep only looks up the unrelated one, since the merged
534 // thread settles from the event itself.
535 Effect.tap((count) =>
536 count === 3 ? Deferred.succeed(mergeLookupStarted, undefined) : Effect.void,
537 ),
538 Effect.tap((count) =>
539 count === 3 ? Deferred.await(releaseMergeLookup) : Effect.void,
540 ),
541 Effect.map(() => makePullRequestSummary({ ...input, state: "open" })),
542 ),
543 onDispatch: () => Deferred.succeed(mergedThreadSettled, undefined),
544 });
545
546 yield* Effect.gen(function* () {
547 const reactor = yield* ThreadSettlementReactor.ThreadSettlementReactor;
548 yield* startHarness(reactor, fixture.activation, fixture.snapshotReads);
549 assert.deepStrictEqual(yield* Ref.get(fixture.commands), []);
550
551 yield* fixture.publishMerge;
552 yield* Deferred.await(mergeLookupStarted);
553 yield* Deferred.await(mergedThreadSettled);
554 yield* Deferred.succeed(releaseMergeLookup, undefined);
555 yield* reactor.drain;
556
557 assert.deepStrictEqual(
558 (yield* Ref.get(fixture.commands)).map((command) => command.threadId),
559 [ThreadId.make("merged-in-app")],
560 );
561 assert.deepStrictEqual(
562 (yield* Ref.get(fixture.summaryCalls))
563 .map((call) => call.number)
564 .toSorted((left, right) => left - right),
565 [42, 99, 99],
566 );
567 }).pipe(Effect.provide(fixture.layer));
568 }),
569 ),
570 );
571
572 it.effect("uses fresh settlement settings after lookup and ignores unrelated changes", () =>
573 Effect.scoped(
574 Effect.gen(function* () {
575 yield* TestClock.setTime(Date.parse(NOW));
576 const state = yield* Ref.make<"merged" | "closed">("merged");
577 const firstLookupStarted = yield* Deferred.make<void>();
578 const releaseFirstLookup = yield* Deferred.make<void>();
579 const laterLookupStarted = yield* Deferred.make<void>();
580 const releaseLaterLookup = yield* Deferred.make<void>();
581 const lookupCount = yield* Ref.make(0);
582 const fixture = yield* makeHarness({
583 snapshot: makeSnapshot([makeThread("settings-thread", { branch: "saved-feature" })]),
584 settings: {
585 ...DEFAULT_SERVER_SETTINGS,
586 sidebarAutoSettleAfterDays: null,
587 sidebarAutoSettleOnMerge: true,
588 },
589 branchPullRequest: () =>
590 Ref.updateAndGet(lookupCount, (count) => count + 1).pipe(
591 Effect.tap((count) =>
592 count === 1
593 ? Deferred.succeed(firstLookupStarted, undefined)
594 : count === 3
595 ? Deferred.succeed(laterLookupStarted, undefined)
596 : Effect.void,
597 ),
598 Effect.tap((count) =>
599 count === 1
600 ? Deferred.await(releaseFirstLookup)
601 : count === 3
602 ? Deferred.await(releaseLaterLookup)
603 : Effect.void,
604 ),
605 Effect.andThen(Ref.get(state)),
606 Effect.map((pullRequestState) => ({ state: pullRequestState, updatedAt: NOW })),
607 ),
608 });
609
610 yield* Effect.gen(function* () {
611 const reactor = yield* ThreadSettlementReactor.ThreadSettlementReactor;
612 yield* reactor.start();
613 yield* Deferred.succeed(fixture.activation, undefined);
614 yield* Queue.take(fixture.snapshotReads);
615 yield* Deferred.await(firstLookupStarted);
616
617 yield* fixture.updateSettings({ sidebarAutoSettleOnMerge: false });
618 yield* Deferred.succeed(releaseFirstLookup, undefined);
619 yield* Queue.take(fixture.snapshotReads);
620 yield* reactor.drain;
621 assert.deepStrictEqual(yield* Ref.get(fixture.commands), []);
622 assert.strictEqual(yield* Ref.get(fixture.snapshotReadCount), 2);
623
624 yield* Ref.set(state, "closed");
625 yield* fixture.updateSettings({ enableAgentBrowserAccess: false });
626 yield* fixture.updateSettings({ sidebarAutoSettleAfterDays: 1 });
627 yield* Deferred.await(laterLookupStarted);
628 yield* Deferred.succeed(releaseLaterLookup, undefined);
629 yield* reactor.drain;
630
631 assert.strictEqual(yield* Ref.get(fixture.snapshotReadCount), 3);
632 assert.strictEqual(yield* Ref.get(lookupCount), 3);
633 assert.deepStrictEqual(
634 (yield* Ref.get(fixture.commands)).map((command) => command.threadId),
635 [ThreadId.make("settings-thread")],
636 );
637 }).pipe(Effect.provide(fixture.layer));
638 }),
639 ),
640 );
641
642 it.effect("keeps an unknown pull request active and continues with other candidates", () =>
643 Effect.scoped(
644 Effect.gen(function* () {
645 yield* TestClock.setTime(Date.parse(NOW));
646 const fixture = yield* makeHarness({
647 snapshot: makeSnapshot(
648 [
649 makeThread("lookup-failed", {
650 linkedPullRequest: {
651 projectId: LINKED_PROJECT_ID,
652 repository: "owner/repository",
653 number: 9,
654 url: "https://example.test/owner/repository/pull/9",
655 },
656 }),
657 makeThread("inactive-without-pr"),
658 ],
659 [makeProject(), makeProject(LINKED_PROJECT_ID, "/workspace/linked")],
660 ),
661 pullRequestSummary: () =>
662 Effect.fail(
663 new PullRequestOperationError({
664 operation: "summary",
665 detail: "host unavailable",
666 }),
667 ),
668 });
669
670 yield* Effect.gen(function* () {
671 const reactor = yield* ThreadSettlementReactor.ThreadSettlementReactor;
672 yield* startHarness(reactor, fixture.activation, fixture.snapshotReads);
673
674 assert.deepStrictEqual(
675 (yield* Ref.get(fixture.commands)).map((command) => command.threadId),
676 [ThreadId.make("inactive-without-pr")],
677 );
678 assert.strictEqual((yield* Ref.get(fixture.summaryCalls)).length, 1);
679 }).pipe(Effect.provide(fixture.layer));
680 }),
681 ),
682 );
683
684 it.effect("keeps threads active when their pull request project is unavailable", () =>
685 Effect.scoped(
686 Effect.gen(function* () {
687 yield* TestClock.setTime(Date.parse(NOW));
688 const linkedPullRequest = {
689 projectId: LINKED_PROJECT_ID,
690 repository: "owner/repository",
691 number: 10,
692 url: "https://example.test/owner/repository/pull/10",
693 } as const;
694 const fixture = yield* makeHarness({
695 snapshot: makeSnapshot(
696 [
697 makeThread("missing-own-project", {
698 latestUserMessageAt: "2026-08-27T00:00:00.000Z",
699 linkedPullRequest,
700 }),
701 makeThread("missing-branch-project", { branch: "saved-feature" }),
702 ],
703 [makeProject(LINKED_PROJECT_ID, "/workspace/linked")],
704 ),
705 pullRequestSummary: (input) =>
706 Effect.succeed(makePullRequestSummary({ ...input, state: "open" })),
707 });
708
709 yield* Effect.gen(function* () {
710 const reactor = yield* ThreadSettlementReactor.ThreadSettlementReactor;
711 yield* startHarness(reactor, fixture.activation, fixture.snapshotReads);
712
713 assert.deepStrictEqual(yield* Ref.get(fixture.commands), []);
714 assert.deepStrictEqual(yield* Ref.get(fixture.summaryCalls), [
715 { projectId: LINKED_PROJECT_ID, repository: "owner/repository", number: 10 },
716 ]);
717 assert.deepStrictEqual(yield* Ref.get(fixture.branchCalls), []);
718 }).pipe(Effect.provide(fixture.layer));
719 }),
720 ),
721 );
722
723 it.effect("deduplicates saved-branch and linked pull request lookups within a sweep", () =>
724 Effect.scoped(
725 Effect.gen(function* () {
726 yield* TestClock.setTime(Date.parse(NOW));
727 const linkedPullRequest = {
728 projectId: LINKED_PROJECT_ID,
729 repository: "owner/repository",
730 number: 77,
731 url: "https://example.test/owner/repository/pull/77",
732 } as const;
733 const fixture = yield* makeHarness({
734 snapshot: makeSnapshot(
735 [
736 makeThread("branch-one", {
737 branch: "saved-feature",
738 worktreePath: "/deleted/worktree-one",
739 }),
740 makeThread("branch-two", {
741 branch: "saved-feature",
742 worktreePath: "/deleted/worktree-two",
743 }),
744 makeThread("linked-one", { linkedPullRequest }),
745 makeThread("linked-two", { linkedPullRequest }),
746 ],
747 [
748 makeProject(PROJECT_ID, "/workspace/project-root"),
749 makeProject(LINKED_PROJECT_ID, "/workspace/linked-root"),
750 ],
751 ),
752 branchPullRequest: () => Effect.succeed({ state: "closed", updatedAt: NOW }),
753 pullRequestSummary: (input) =>
754 Effect.succeed(makePullRequestSummary({ ...input, state: "merged" })),
755 });
756
757 yield* Effect.gen(function* () {
758 const reactor = yield* ThreadSettlementReactor.ThreadSettlementReactor;
759 yield* startHarness(reactor, fixture.activation, fixture.snapshotReads);
760
761 assert.deepStrictEqual(yield* Ref.get(fixture.branchCalls), [
762 { cwd: "/workspace/project-root", branch: "saved-feature" },
763 ]);
764 assert.deepStrictEqual(yield* Ref.get(fixture.summaryCalls), [
765 { projectId: LINKED_PROJECT_ID, repository: "owner/repository", number: 77 },
766 ]);
767 assert.deepStrictEqual(
768 new Set((yield* Ref.get(fixture.commands)).map((command) => command.threadId)),
769 new Set([
770 ThreadId.make("branch-one"),
771 ThreadId.make("branch-two"),
772 ThreadId.make("linked-one"),
773 ThreadId.make("linked-two"),
774 ]),
775 );
776 }).pipe(Effect.provide(fixture.layer));
777 }),
778 ),
779 );
780
781 it.effect("looks up the branch pull request from a thread's live worktree", () =>
782 Effect.scoped(
783 Effect.gen(function* () {
784 yield* TestClock.setTime(Date.parse(NOW));
785 const fixture = yield* makeHarness({
786 snapshot: makeSnapshot(
787 [
788 makeThread("live-worktree", {
789 branch: "feature/live",
790 worktreePath: "/workspace/project-root/.worktrees/live",
791 }),
792 makeThread("deleted-worktree", {
793 branch: "feature/deleted",
794 worktreePath: "/workspace/project-root/.worktrees/deleted",
795 }),
796 ],
797 [makeProject(PROJECT_ID, "/workspace/project-root")],
798 ),
799 existingWorktreePaths: ["/workspace/project-root/.worktrees/live"],
800 });
801
802 yield* Effect.gen(function* () {
803 const reactor = yield* ThreadSettlementReactor.ThreadSettlementReactor;
804 yield* startHarness(reactor, fixture.activation, fixture.snapshotReads);
805
806 assert.deepStrictEqual(
807 new Set(yield* Ref.get(fixture.branchCalls)),
808 new Set([
809 { cwd: "/workspace/project-root/.worktrees/live", branch: "feature/live" },
810 { cwd: "/workspace/project-root", branch: "feature/deleted" },
811 ]),
812 );
813 }).pipe(Effect.provide(fixture.layer));
814 }),
815 ),
816 );
817
818 it.effect("carries the snapshot guard and survives a stale dispatch rejection", () =>
819 Effect.scoped(
820 Effect.gen(function* () {
821 yield* TestClock.setTime(Date.parse(NOW));
822 const fixture = yield* makeHarness({
823 snapshot: makeSnapshot([makeThread("stale"), makeThread("next-candidate")]),
824 onDispatch: (command) =>
825 command.threadId === ThreadId.make("stale")
826 ? Effect.fail(
827 new OrchestrationCommandInvariantError({
828 commandType: command.type,
829 detail: "thread changed after settlement evaluation",
830 }),
831 )
832 : Effect.void,
833 });
834
835 yield* Effect.gen(function* () {
836 const reactor = yield* ThreadSettlementReactor.ThreadSettlementReactor;
837 yield* startHarness(reactor, fixture.activation, fixture.snapshotReads);
838
839 const firstSweep = yield* Ref.get(fixture.commands);
840 assert.strictEqual(
841 firstSweep.find((command) => command.threadId === ThreadId.make("stale"))
842 ?.snapshotSequence,
843 1,
844 );
845 assert.strictEqual(
846 firstSweep.some((command) => command.threadId === ThreadId.make("next-candidate")),
847 true,
848 );
849
850 yield* fixture.updateSettings({ sidebarAutoSettleAfterDays: 4 });
851 yield* Queue.take(fixture.snapshotReads);
852 yield* reactor.drain;
853 assert.strictEqual((yield* Ref.get(fixture.commands)).length, 4);
854 }).pipe(Effect.provide(fixture.layer));
855 }),
856 ),
857 );
858});
apps/server/src/orchestration/ThreadSettlementReactor.ts+0−249
@@ -1,249 +0,0 @@
1import { CommandId } from "@t3tools/contracts";
2import { makeDrainableWorker } from "@t3tools/shared/DrainableWorker";
3import * as Cause from "effect/Cause";
4import * as Context from "effect/Context";
5import * as Crypto from "effect/Crypto";
6import * as DateTime from "effect/DateTime";
7import * as Effect from "effect/Effect";
8import * as FileSystem from "effect/FileSystem";
9import * as Layer from "effect/Layer";
10import * as Schedule from "effect/Schedule";
11import type * as Scope from "effect/Scope";
12import * as Stream from "effect/Stream";
13
14import * as GitManager from "../git/GitManager.ts";
15import * as PullRequestService from "../pullRequest/PullRequestService.ts";
16import * as ServerSettings from "../serverSettings.ts";
17import { forkParked } from "../serverActivation.ts";
18import * as OrchestrationEngine from "./Services/OrchestrationEngine.ts";
19import * as ProjectionSnapshotQuery from "./Services/ProjectionSnapshotQuery.ts";
20import {
21 isAutoSettlementCandidate,
22 resolveAutoSettlementAt,
23 type SettlementPullRequest,
24} from "./ThreadSettlementPolicy.ts";
25
26export class ThreadSettlementReactor extends Context.Service<
27 ThreadSettlementReactor,
28 {
29 readonly start: () => Effect.Effect<void, never, Scope.Scope>;
30 readonly drain: Effect.Effect<void>;
31 }
32>()("t3/orchestration/ThreadSettlementReactor") {}
33
34export const make = Effect.gen(function* () {
35 const engine = yield* OrchestrationEngine.OrchestrationEngineService;
36 const snapshots = yield* ProjectionSnapshotQuery.ProjectionSnapshotQuery;
37 const settingsService = yield* ServerSettings.ServerSettingsService;
38 const git = yield* GitManager.GitManager;
39 const pullRequests = yield* PullRequestService.PullRequestService;
40 const crypto = yield* Crypto.Crypto;
41 const fileSystem = yield* FileSystem.FileSystem;
42
43 const sweep = Effect.fn("ThreadSettlementReactor.sweep")(function* (
44 mergedPullRequest: PullRequestService.PullRequestMergeEvent | null,
45 ) {
46 const snapshot = yield* snapshots.getShellSnapshot();
47 const now = DateTime.formatIso(yield* DateTime.now);
48 const projects = new Map(snapshot.projects.map((project) => [project.id, project]));
49 // A merge event re-sweeps every candidate, not just the threads linked to
50 // the merged pull request: most threads carry no link and settle from
51 // their branch lookup, which would otherwise wait for the next minute's
52 // sweep on a possibly stale cached answer.
53 const candidates = snapshot.threads.filter((thread) => isAutoSettlementCandidate(thread, now));
54 // Use the same cwd as the sidebar so both paths share GitManager's PR cache.
55 const lookupCwdByThreadId = new Map<string, string>();
56 yield* Effect.forEach(
57 candidates,
58 (thread) =>
59 Effect.gen(function* () {
60 const project = projects.get(thread.projectId);
61 if (project === undefined || thread.linkedPullRequest != null) return;
62 const worktreeExists =
63 thread.worktreePath !== null &&
64 (yield* fileSystem.exists(thread.worktreePath).pipe(Effect.orElseSucceed(() => false)));
65 lookupCwdByThreadId.set(
66 thread.id,
67 worktreeExists && thread.worktreePath !== null
68 ? thread.worktreePath
69 : project.workspaceRoot,
70 );
71 }),
72 { concurrency: 8, discard: true },
73 );
74 if (mergedPullRequest !== null) {
75 // The merge just confirmed a terminal state the lookup caches can still
76 // call open (branch answers live two minutes, the sweep runs every
77 // minute). Drop the swept checkouts' cached answers so the merge settles
78 // its branch threads now instead of on a later sweep. Threads linked to
79 // the merged pull request settle from the event itself below and need no
80 // lookup, so they are absent from this map by construction.
81 const cwds = [...new Set(lookupCwdByThreadId.values())];
82 yield* Effect.forEach(cwds, (cwd) => git.invalidateStatus(cwd), {
83 concurrency: 8,
84 discard: true,
85 });
86 }
87 const lookupKey = (thread: (typeof candidates)[number]) => {
88 if (thread.linkedPullRequest != null) {
89 return JSON.stringify([
90 "linked",
91 thread.linkedPullRequest.projectId,
92 thread.linkedPullRequest.repository,
93 thread.linkedPullRequest.number,
94 ]);
95 }
96 if (thread.branch === null) return JSON.stringify(["none", thread.id]);
97 const cwd = lookupCwdByThreadId.get(thread.id);
98 return JSON.stringify(
99 cwd === undefined ? ["missing-project", thread.id] : ["branch", cwd, thread.branch],
100 );
101 };
102 const groups = Map.groupBy(candidates, lookupKey);
103
104 const pullRequestFor = Effect.fn("ThreadSettlementReactor.pullRequestFor")(function* (
105 thread: (typeof candidates)[number],
106 ) {
107 if (thread.linkedPullRequest != null) {
108 // The event carries the merged state, so only the threads linked to
109 // that exact pull request settle from it. Every other linked thread
110 // falls through to a fresh summary lookup below: the merge sweep
111 // covers all candidates, and an unrelated merge must never settle
112 // them.
113 if (
114 mergedPullRequest !== null &&
115 thread.linkedPullRequest.projectId === mergedPullRequest.projectId &&
116 thread.linkedPullRequest.repository.toLowerCase() ===
117 mergedPullRequest.repository.toLowerCase() &&
118 thread.linkedPullRequest.number === mergedPullRequest.number
119 ) {
120 return {
121 state: "merged",
122 updatedAt: mergedPullRequest.mergedAt,
123 } satisfies SettlementPullRequest;
124 }
125 if (!projects.has(thread.linkedPullRequest.projectId)) {
126 return yield* Effect.die(new Error("linked pull request project not found"));
127 }
128 const summary = yield* pullRequests.summary(
129 {
130 projectId: thread.linkedPullRequest.projectId,
131 repository: thread.linkedPullRequest.repository,
132 number: thread.linkedPullRequest.number,
133 },
134 { recoverTransientFailure: false },
135 );
136 return {
137 state: summary.state,
138 updatedAt: summary.updatedAt,
139 } satisfies SettlementPullRequest;
140 }
141 if (thread.branch === null) return null;
142 const cwd = lookupCwdByThreadId.get(thread.id);
143 if (cwd === undefined) {
144 return yield* Effect.die(new Error("thread project not found"));
145 }
146 return yield* git.branchPullRequest({ cwd, branch: thread.branch });
147 });
148
149 yield* Effect.forEach(
150 groups.values(),
151 (group) =>
152 Effect.gen(function* () {
153 const pullRequest = yield* pullRequestFor(group[0]!);
154 yield* Effect.forEach(
155 group,
156 (thread) =>
157 Effect.gen(function* () {
158 const settings = yield* settingsService.getSettings;
159 const decisionNow = DateTime.formatIso(yield* DateTime.now);
160 const settledAt = resolveAutoSettlementAt({
161 thread,
162 pullRequest,
163 now: decisionNow,
164 autoSettleAfterDays: settings.sidebarAutoSettleAfterDays,
165 autoSettleOnMerge: settings.sidebarAutoSettleOnMerge,
166 });
167 if (settledAt === null) {
168 return;
169 }
170 const uuid = yield* crypto.randomUUIDv4;
171 yield* engine.dispatch({
172 type: "thread.auto-settle",
173 commandId: CommandId.make(`server:auto-settle:${thread.id}:${uuid}`),
174 threadId: thread.id,
175 snapshotSequence: snapshot.snapshotSequence,
176 settledAt,
177 });
178 }).pipe(
179 Effect.catchCause((cause) =>
180 Cause.hasInterruptsOnly(cause)
181 ? Effect.failCause(cause)
182 : Effect.logWarning("automatic thread settlement skipped", {
183 threadId: thread.id,
184 cause: Cause.pretty(cause),
185 }),
186 ),
187 ),
188 { discard: true },
189 );
190 }).pipe(
191 Effect.catchCause((cause) =>
192 Cause.hasInterruptsOnly(cause)
193 ? Effect.failCause(cause)
194 : Effect.logWarning("automatic thread settlement skipped", {
195 threadIds: group.map((thread) => thread.id),
196 cause: Cause.pretty(cause),
197 }),
198 ),
199 ),
200 { concurrency: 8, discard: true },
201 );
202 });
203
204 const runSweep = (mergedPullRequest: PullRequestService.PullRequestMergeEvent | null) =>
205 sweep(mergedPullRequest).pipe(
206 Effect.catchCause((cause) =>
207 Cause.hasInterruptsOnly(cause)
208 ? Effect.failCause(cause)
209 : Effect.logWarning("automatic thread settlement sweep failed", {
210 cause: Cause.pretty(cause),
211 }),
212 ),
213 );
214 const worker = yield* makeDrainableWorker(() => runSweep(null));
215
216 const start: ThreadSettlementReactor["Service"]["start"] = Effect.fn(
217 "ThreadSettlementReactor.start",
218 )(function* () {
219 const settingsChanges = yield* settingsService.subscribeChanges;
220 const mergedPullRequests = yield* pullRequests.subscribeMerges;
221 const initialSettings = yield* settingsService.getSettings.pipe(Effect.orDie);
222 let lastAfterDays = initialSettings.sidebarAutoSettleAfterDays;
223 let lastOnMerge = initialSettings.sidebarAutoSettleOnMerge;
224 yield* forkParked(
225 Effect.gen(function* () {
226 yield* worker.enqueue(undefined);
227 yield* worker.drain;
228 }).pipe(Effect.repeat(Schedule.spaced("1 minute")), Effect.asVoid),
229 );
230 yield* forkParked(
231 Stream.runForEach(settingsChanges, (settings) => {
232 if (
233 settings.sidebarAutoSettleAfterDays === lastAfterDays &&
234 settings.sidebarAutoSettleOnMerge === lastOnMerge
235 ) {
236 return Effect.void;
237 }
238 lastAfterDays = settings.sidebarAutoSettleAfterDays;
239 lastOnMerge = settings.sidebarAutoSettleOnMerge;
240 return worker.enqueue(undefined);
241 }),
242 );
243 yield* forkParked(Stream.runForEach(mergedPullRequests, runSweep));
244 });
245
246 return { start, drain: worker.drain } satisfies ThreadSettlementReactor["Service"];
247});
248
249export const layer = Layer.effect(ThreadSettlementReactor, make);
apps/server/src/orchestration/decider.ts+1−0
@@ -3,6 +3,7 @@ import { · 2 unmodified lines
3 type OrchestrationCommand,3 type OrchestrationCommand,
4 type OrchestrationEvent,4 type OrchestrationEvent,
5 type OrchestrationReadModel,5 type OrchestrationReadModel,
6 type OrchestrationThread,
6} from "@t3tools/contracts/legacy-orchestration";7} from "@t3tools/contracts/legacy-orchestration";
7import * as DateTime from "effect/DateTime";8import * as DateTime from "effect/DateTime";
8import * as Crypto from "effect/Crypto";9import * as Crypto from "effect/Crypto";
apps/server/src/persistence/Layers/OrchestrationEventStore.ts+11−10
@@ -647,16 +647,17 @@ const makeEventStore = Effect.gen(function* () { · 646 unmodified lines
647 const findEventAfter = SqlSchema.findOneOption({647 const findEventAfter = SqlSchema.findOneOption({
648 Request: HasEventAfterRequestSchema,648 Request: HasEventAfterRequestSchema,
649 Result: Schema.Struct({ sequence: Schema.Number }),649 Result: Schema.Struct({ sequence: Schema.Number }),
650 execute: (request) =>650 execute: (request) => sql`
651 sql`651 SELECT sequence
652 SELECT sequence652 FROM orchestration_events
653 FROM orchestration_events653 WHERE aggregate_kind = ${request.aggregateKind}
654 WHERE aggregate_kind = ${request.aggregateKind}654 AND stream_id = ${request.aggregateId}
655 AND stream_id = ${request.aggregateId}655 AND ${sql.and([
656 AND event_type = ${request.type}656 sql`sequence > ${request.sequenceExclusive}`,
657 AND sequence > ${request.sequenceExclusive}657 ...(request.type === undefined ? [] : [sql`event_type = ${request.type}`]),
658 LIMIT 1658 ])}
659 `,659 LIMIT 1
660 `,
660 });661 });
661 662
662 const hasEventAfter: OrchestrationEventStoreShape["hasEventAfter"] = (input) =>663 const hasEventAfter: OrchestrationEventStoreShape["hasEventAfter"] = (input) =>
apps/server/src/persistence/Services/OrchestrationEventStore.ts+3−2
@@ -123,7 +123,8 @@ export interface OrchestrationEventStoreShape { · 122 unmodified lines
123 readonly afterSequence?: number;123 readonly afterSequence?: number;
124 }) => Stream.Stream<ApplicationStoredEvent, OrchestrationEventStoreError>;124 }) => Stream.Stream<ApplicationStoredEvent, OrchestrationEventStoreError>;
125 /**125 /**
126 * Check whether an aggregate has an event of the given type after a sequence.126 * Check whether an aggregate has an event after a sequence, optionally
127 * restricted to one event type.
127 *128 *
128 * Used during replay to tell whether a later event supersedes the one being129 * Used during replay to tell whether a later event supersedes the one being
129 * applied, without streaming the rest of the log.130 * applied, without streaming the rest of the log.
@@ -131,7 +132,7 @@ export interface OrchestrationEventStoreShape { · 1 unmodified line
131 readonly hasEventAfter: (input: {132 readonly hasEventAfter: (input: {
132 readonly aggregateKind: OrchestrationEvent["aggregateKind"];133 readonly aggregateKind: OrchestrationEvent["aggregateKind"];
133 readonly aggregateId: string;134 readonly aggregateId: string;
134 readonly type: OrchestrationEvent["type"];135 readonly type?: OrchestrationEvent["type"];
135 readonly sequenceExclusive: number;136 readonly sequenceExclusive: number;
136 }) => Effect.Effect<boolean, OrchestrationEventStoreError>;137 }) => Effect.Effect<boolean, OrchestrationEventStoreError>;
137}138}
apps/server/src/provider/Drivers/ClaudeDriver.ts+24−19
@@ -40,6 +40,7 @@ import { · 39 unmodified lines
40} from "../Layers/ClaudeProvider.ts";40} from "../Layers/ClaudeProvider.ts";
41import { makeManagedServerProvider } from "../makeManagedServerProvider.ts";41import { makeManagedServerProvider } from "../makeManagedServerProvider.ts";
42import * as ModelManifest from "../ModelManifest.ts";42import * as ModelManifest from "../ModelManifest.ts";
43import { resolveClaudeModelCatalog } from "../ClaudeModelCatalog.ts";
43import {44import {
44 defaultProviderContinuationIdentity,45 defaultProviderContinuationIdentity,
45 type ProviderDriver,46 type ProviderDriver,
@@ -114,6 +115,7 @@ export const ClaudeDriver: ProviderDriver<ClaudeSettings, ClaudeDriverEnv> = { · 68 unmodified lines
114 const serverSettings = yield* ServerSettingsService;115 const serverSettings = yield* ServerSettingsService;
115 const eventLoggers = yield* ProviderEventLoggers;116 const eventLoggers = yield* ProviderEventLoggers;
116 const modelManifest = yield* ModelManifest.ModelManifest;117 const modelManifest = yield* ModelManifest.ModelManifest;
118 const modelCatalog = modelManifest.current.pipe(Effect.map(resolveClaudeModelCatalog));
117 const processEnv = mergeProviderInstanceEnvironment(environment);119 const processEnv = mergeProviderInstanceEnvironment(environment);
118 const fallbackContinuationIdentity = defaultProviderContinuationIdentity({120 const fallbackContinuationIdentity = defaultProviderContinuationIdentity({
119 driverKind: DRIVER_KIND,121 driverKind: DRIVER_KIND,
@@ -157,7 +159,11 @@ export const ClaudeDriver: ProviderDriver<ClaudeSettings, ClaudeDriverEnv> = { · 37 unmodified lines
157 }),159 }),
158 ),160 ),
159 );161 );
160 const textGeneration = yield* makeClaudeTextGeneration(effectiveConfig, processEnv);162 const textGeneration = yield* makeClaudeTextGeneration(
163 effectiveConfig,
164 processEnv,
165 modelCatalog,
166 );
161 167
162 // Per-instance capabilities cache: keyed on binary + resolved HOME so168 // Per-instance capabilities cache: keyed on binary + resolved HOME so
163 // account-specific probes never share auth metadata across instances.169 // account-specific probes never share auth metadata across instances.
@@ -171,22 +177,21 @@ export const ClaudeDriver: ProviderDriver<ClaudeSettings, ClaudeDriverEnv> = { · 7 unmodified lines
171 });177 });
172 const capabilitiesCacheKey = yield* makeClaudeCapabilitiesCacheKey(effectiveConfig, cwd);178 const capabilitiesCacheKey = yield* makeClaudeCapabilitiesCacheKey(effectiveConfig, cwd);
173 179
174 // Kick the TTL-gated manifest refresh in the background and classify180 // Start the TTL-gated refresh without delaying provider readiness. The
175 // with the in-memory manifest, so a slow or hung fetch never delays the181 // next check observes a remote manifest after the background fetch lands.
176 // provider check. A refresh that lands mid-probe applies on the next one.
177 const checkProvider = modelManifest.refreshInBackground.pipe(182 const checkProvider = modelManifest.refreshInBackground.pipe(
178 Effect.andThen(183 Effect.andThen(
179 Effect.zipWith(184 modelManifest.current.pipe(
180 checkClaudeProviderStatus(185 Effect.flatMap((manifest) =>
181 effectiveConfig,186 checkClaudeProviderStatus(
182 () => Cache.get(capabilitiesProbeCache, capabilitiesCacheKey),187 effectiveConfig,
183 processEnv,188 () => Cache.get(capabilitiesProbeCache, capabilitiesCacheKey),
184 cwd,189 processEnv,
190 cwd,
191 resolveClaudeModelCatalog(manifest),
192 ),
185 ),193 ),
186 modelManifest.current,194 Effect.map(stampIdentity),
187 (draft, manifest) =>
188 stampIdentity(ModelManifest.applyModelManifest(draft, manifest, DRIVER_KIND)),
189 { concurrent: true },
190 ),195 ),
191 ),196 ),
192 Effect.provideService(ChildProcessSpawner.ChildProcessSpawner, spawner),197 Effect.provideService(ChildProcessSpawner.ChildProcessSpawner, spawner),
@@ -201,11 +206,11 @@ export const ClaudeDriver: ProviderDriver<ClaudeSettings, ClaudeDriverEnv> = { · 8 unmodified lines
201 streamSettings: snapshotSettings.streamSettings,206 streamSettings: snapshotSettings.streamSettings,
202 haveSettingsChanged: haveProviderSnapshotSettingsChanged,207 haveSettingsChanged: haveProviderSnapshotSettingsChanged,
203 initialSnapshot: (settings) =>208 initialSnapshot: (settings) =>
204 Effect.zipWith(209 modelManifest.current.pipe(
205 makePendingClaudeProvider(settings.provider),210 Effect.flatMap((manifest) =>
206 modelManifest.current,211 makePendingClaudeProvider(settings.provider, resolveClaudeModelCatalog(manifest)),
207 (draft, manifest) =>212 ),
208 stampIdentity(ModelManifest.applyModelManifest(draft, manifest, DRIVER_KIND)),213 Effect.map(stampIdentity),
209 ),214 ),
210 checkProvider,215 checkProvider,
211 enrichSnapshot: ({ settings, snapshot, publishSnapshot }) =>216 enrichSnapshot: ({ settings, snapshot, publishSnapshot }) =>
apps/server/src/provider/acp/AcpNativeLogging.ts+44−6
@@ -1,5 +1,5 @@
1import type { ProviderDriverKind, ThreadId } from "@t3tools/contracts";1import type { ProviderDriverKind, ThreadId } from "@t3tools/contracts";
2import { causeErrorTag } from "@t3tools/shared/observability";2import { causeErrorTag, errorTag } from "@t3tools/shared/observability";
3import * as Cause from "effect/Cause";3import * as Cause from "effect/Cause";
4import * as Crypto from "effect/Crypto";4import * as Crypto from "effect/Crypto";
5import * as DateTime from "effect/DateTime";5import * as DateTime from "effect/DateTime";
@@ -13,6 +13,40 @@ import { · 7 unmodified lines
13} from "../NativeProtocolLogging.ts";13} from "../NativeProtocolLogging.ts";
14import type * as AcpSessionRuntime from "./AcpSessionRuntime.ts";14import type * as AcpSessionRuntime from "./AcpSessionRuntime.ts";
15 15
16const transientProtocolUpdates = new Set(["agent_message_chunk", "agent_thought_chunk"]);
17
18function structuralMethod(value: string): string {
19 return value.length <= 128 && /^[A-Za-z][A-Za-z0-9._:/-]*$/.test(value) ? value : "unknown";
20}
21
22function summarizePayload(payload: unknown): Readonly<Record<string, unknown>> {
23 if (payload === null) return { valueType: "null" };
24 if (typeof payload === "string") {
25 return { valueType: "string", byteLength: new TextEncoder().encode(payload).byteLength };
26 }
27 if (payload instanceof Uint8Array) {
28 return { valueType: "bytes", byteLength: payload.byteLength };
29 }
30 if (Array.isArray(payload)) {
31 return { valueType: "array", itemCount: payload.length };
32 }
33 if (typeof payload !== "object") {
34 return { valueType: typeof payload };
35 }
36
37 try {
38 const record = payload as Record<string, unknown>;
39 return {
40 valueType: "object",
41 fieldCount: Object.keys(record).length,
42 ...(typeof record._tag === "string" ? { messageTag: errorTag(record) } : {}),
43 ...(typeof record.tag === "string" ? { method: structuralMethod(record.tag) } : {}),
44 };
45 } catch {
46 return { valueType: "object" };
47 }
48}
49
16function formatRequestLogPayload(event: AcpSessionRuntime.AcpSessionRequestLogEvent) {50function formatRequestLogPayload(event: AcpSessionRuntime.AcpSessionRequestLogEvent) {
17 return {51 return {
18 method: structuralProtocolMethod(event.method),52 method: structuralProtocolMethod(event.method),
@@ -137,11 +171,15 @@ export const makeAcpNativeLoggerFactory = Effect.fn("makeAcpNativeLoggerFactory" · 118 unmodified lines
137 protocolLogging: {171 protocolLogging: {
138 logIncoming: true,172 logIncoming: true,
139 logOutgoing: true,173 logOutgoing: true,
140 logger: (event: EffectAcpProtocol.AcpProtocolLogEvent) =>174 logger: (event: EffectAcpProtocol.AcpProtocolLogEvent) => {
141 writeNativeAcpLog({175 const filtered = filterTransientProtocolLog(event);
142 kind: "protocol",176 return filtered
143 payload: formatAcpProtocolLogPayload(event),177 ? writeNativeAcpLog({
144 }),178 kind: "protocol",
179 payload: formatAcpProtocolLogPayload(filtered),
180 })
181 : Effect.void;
182 },
145 } satisfies NonNullable<AcpSessionRuntime.AcpSessionRuntimeOptions["protocolLogging"]>,183 } satisfies NonNullable<AcpSessionRuntime.AcpSessionRuntimeOptions["protocolLogging"]>,
146 }184 }
147 : {}),185 : {}),