authorgravatar for me@t3.ggTheo Browne <me@t3.gg> 2026-09-01 02:12:35-07:00
committergravatar for noreply@github.comGitHub <noreply@github.com> 2026-09-01 02:12:35-07:00
log0bfb6df34b26dfe0162db6c09dca00bc8c5a5ec4
tree236fa71c14a1d3b7867896604851858ea05c506c
parenta9ffb8279614df6ae2f1f4b7f09a0dc42edef797
signature Signed by PGP key B5690EEEBB952194

perf(server): cut idle CPU use and stop provider event leaks (#8187)

19 files changed, 913 insertions(+), 98 deletions(-)

apps/server/src/orchestration/Layers/ProviderRuntimeIngestion.test.ts+23
......@@ -973,6 +973,29 @@ describe("ProviderRuntimeIngestion", () => {
973973 );
974974 });
975975
976 it("ignores provider content deltas that cannot change thread state", async () => {
977 const harness = await createHarness();
978 const initial = await harness.readModel();
979
980 for (const streamKind of ["reasoning_text", "command_output", "file_change_output"] as const) {
981 harness.emit({
982 type: "content.delta",
983 eventId: asEventId(`evt-ignored-${streamKind}`),
984 provider: ProviderDriverKind.make("codex"),
985 createdAt: "2026-01-01T00:00:00.000Z",
986 threadId: asThreadId("thread-1"),
987 turnId: asTurnId("turn-ignored"),
988 payload: {
989 streamKind,
990 delta: "ignored output",
991 },
992 });
993 }
994
995 await harness.drain();
996 expect(await harness.readModel()).toEqual(initial);
997 });
998
976999 it("maps canonical content delta/item completed into finalized assistant messages", async () => {
9771000 const harness = await createHarness();
9781001 const now = "2026-01-01T00:00:00.000Z";
apps/server/src/orchestration/Layers/ProviderRuntimeIngestion.ts+15-3
......@@ -1499,6 +1499,10 @@ const make = Effect.gen(function* () {
14991499
15001500 const processRuntimeEvent = (event: ProviderRuntimeEvent) =>
15011501 Effect.gen(function* () {
1502 if (event.type === "content.delta" && event.payload.streamKind !== "assistant_text") {
1503 return;
1504 }
1505
15021506 const thread = yield* resolveThreadShell(event.threadId);
15031507 if (!thread) return;
15041508
......@@ -1515,9 +1519,17 @@ const make = Effect.gen(function* () {
15151519 const now = event.createdAt;
15161520 const eventTurnId = toTurnId(event.turnId);
15171521 const activeTurnId = thread.session?.activeTurnId ?? null;
1518 const pendingTurnStart = yield* projectionTurnRepository.getPendingTurnStartByThreadId({
1519 threadId: thread.id,
1520 });
1522 const pendingTurnStart =
1523 event.type === "session.started" ||
1524 event.type === "session.state.changed" ||
1525 event.type === "session.exited" ||
1526 event.type === "thread.started" ||
1527 event.type === "turn.started" ||
1528 event.type === "turn.completed"
1529 ? yield* projectionTurnRepository.getPendingTurnStartByThreadId({
1530 threadId: thread.id,
1531 })
1532 : Option.none();
15211533 const hasPendingTurnStart =
15221534 Option.isSome(pendingTurnStart) && thread.session?.status === "starting";
15231535
apps/server/src/project/RepositoryIdentityResolver.test.ts+84
......@@ -5,6 +5,7 @@ import * as Effect from "effect/Effect";
55import * as FileSystem from "effect/FileSystem";
66import * as Layer from "effect/Layer";
77import * as Path from "effect/Path";
8import * as ChildProcessSpawner from "effect/unstable/process/ChildProcessSpawner";
89import { TestClock } from "effect/testing";
910
1011import * as ProcessRunner from "../processRunner.ts";
......@@ -35,6 +36,89 @@ const makeRepositoryIdentityResolverTestLayer = (options: {
3536 ).pipe(Layer.provide(ProcessRunner.layer));
3637
3738it.layer(NodeServices.layer)("RepositoryIdentityResolverLive", (it) => {
39 it.effect("reuses the cached Git root for repeated workspace lookups", () => {
40 const calls: Array<ReadonlyArray<string>> = [];
41 const processRunner = Layer.succeed(ProcessRunner.ProcessRunner, {
42 run: (input) =>
43 Effect.sync(() => {
44 calls.push(input.args);
45 return {
46 stdout: input.args.includes("rev-parse")
47 ? "/repo\n"
48 : "origin\tgit@github.com:T3Tools/t3code.git (fetch)\n",
49 stderr: "",
50 code: ChildProcessSpawner.ExitCode(0),
51 timedOut: false,
52 stdoutTruncated: false,
53 stderrTruncated: false,
54 stdoutInvalidUtf8: false,
55 stderrInvalidUtf8: false,
56 };
57 }),
58 });
59 const resolverLayer = Layer.effect(
60 RepositoryIdentityResolver.RepositoryIdentityResolver,
61 RepositoryIdentityResolver.make(),
62 ).pipe(Layer.provide(processRunner));
63
64 return Effect.gen(function* () {
65 const resolver = yield* RepositoryIdentityResolver.RepositoryIdentityResolver;
66 const first = yield* resolver.resolve("/repo/packages/web");
67 const second = yield* resolver.resolve("/repo/packages/web");
68
69 expect(first?.canonicalKey).toBe("github.com/t3tools/t3code");
70 expect(second).toEqual(first);
71 expect(calls).toEqual([
72 ["-C", "/repo/packages/web", "rev-parse", "--show-toplevel"],
73 ["-C", "/repo", "remote", "-v"],
74 ]);
75 }).pipe(Effect.provide(resolverLayer));
76 });
77
78 it.effect("retries Git root discovery after a failed lookup", () => {
79 const calls: Array<ReadonlyArray<string>> = [];
80 let rootAttempts = 0;
81 const processRunner = Layer.succeed(ProcessRunner.ProcessRunner, {
82 run: (input) =>
83 Effect.sync(() => {
84 calls.push(input.args);
85 const rootLookup = input.args.includes("rev-parse");
86 const failed = rootLookup && rootAttempts++ === 0;
87 return {
88 stdout: rootLookup
89 ? failed
90 ? ""
91 : "/repo\n"
92 : "origin\tgit@github.com:T3Tools/t3code.git (fetch)\n",
93 stderr: failed ? "temporary Git failure" : "",
94 code: ChildProcessSpawner.ExitCode(failed ? 1 : 0),
95 timedOut: false,
96 stdoutTruncated: false,
97 stderrTruncated: false,
98 stdoutInvalidUtf8: false,
99 stderrInvalidUtf8: false,
100 };
101 }),
102 });
103 const resolverLayer = Layer.effect(
104 RepositoryIdentityResolver.RepositoryIdentityResolver,
105 RepositoryIdentityResolver.make(),
106 ).pipe(Layer.provide(processRunner));
107
108 return Effect.gen(function* () {
109 const resolver = yield* RepositoryIdentityResolver.RepositoryIdentityResolver;
110 expect(yield* resolver.resolve("/repo/packages/web")).toBeNull();
111
112 const recovered = yield* resolver.resolve("/repo/packages/web");
113 expect(recovered?.rootPath).toBe("/repo");
114 expect(calls).toEqual([
115 ["-C", "/repo/packages/web", "rev-parse", "--show-toplevel"],
116 ["-C", "/repo/packages/web", "rev-parse", "--show-toplevel"],
117 ["-C", "/repo", "remote", "-v"],
118 ]);
119 }).pipe(Effect.provide(resolverLayer));
120 });
121
38122 it.effect("normalizes equivalent GitHub remotes into a stable repository identity", () =>
39123 Effect.gen(function* () {
40124 const fileSystem = yield* FileSystem.FileSystem;
apps/server/src/project/RepositoryIdentityResolver.ts+21-11
......@@ -90,7 +90,6 @@ function buildRepositoryIdentity(input: {
9090const resolveRepositoryIdentityCacheKey = Effect.fn("RepositoryIdentityResolver.resolveCacheKey")(
9191 function* (cwd: string) {
9292 const processRunner = yield* ProcessRunner.ProcessRunner;
93 let cacheKey = cwd;
9493
9594 // git is a real executable on every platform — no cmd.exe shell mode, which
9695 // would split paths containing spaces during cmd's re-tokenization.
......@@ -102,15 +101,11 @@ const resolveRepositoryIdentityCacheKey = Effect.fn("RepositoryIdentityResolver.
102101 })
103102 .pipe(Effect.option);
104103 if (topLevelResult._tag === "None" || topLevelResult.value.code !== 0) {
105 return cacheKey;
104 return null;
106105 }
107106
108107 const candidate = topLevelResult.value.stdout.trim();
109 if (candidate.length > 0) {
110 cacheKey = candidate;
111 }
112
113 return cacheKey;
108 return candidate.length > 0 ? candidate : null;
114109 },
115110);
116111
......@@ -139,6 +134,22 @@ export const make = Effect.fn("RepositoryIdentityResolver.make")(function* (
139134 options: RepositoryIdentityResolverOptions = {},
140135) {
141136 const processRunner = yield* ProcessRunner.ProcessRunner;
137 const cacheCapacity = options.cacheCapacity ?? DEFAULT_REPOSITORY_IDENTITY_CACHE_CAPACITY;
138
139 const repositoryRootCache = yield* Cache.makeWith<string, string | null>(
140 (cwd) =>
141 resolveRepositoryIdentityCacheKey(cwd).pipe(
142 Effect.provideService(ProcessRunner.ProcessRunner, processRunner),
143 ),
144 {
145 capacity: cacheCapacity,
146 timeToLive: Exit.match({
147 onSuccess: (value) =>
148 value === null ? Duration.zero : (options.positiveCacheTtl ?? DEFAULT_POSITIVE_CACHE_TTL),
149 onFailure: () => Duration.zero,
150 }),
151 },
152 );
142153
143154 const repositoryIdentityCache = yield* Cache.makeWith<string, RepositoryIdentity | null>(
144155 (cacheKey) =>
......@@ -146,7 +157,7 @@ export const make = Effect.fn("RepositoryIdentityResolver.make")(function* (
146157 Effect.provideService(ProcessRunner.ProcessRunner, processRunner),
147158 ),
148159 {
149 capacity: options.cacheCapacity ?? DEFAULT_REPOSITORY_IDENTITY_CACHE_CAPACITY,
160 capacity: cacheCapacity,
150161 timeToLive: Exit.match({
151162 onSuccess: (value) =>
152163 value === null
......@@ -160,9 +171,8 @@ export const make = Effect.fn("RepositoryIdentityResolver.make")(function* (
160171 const resolve: RepositoryIdentityResolver["Service"]["resolve"] = Effect.fn(
161172 "RepositoryIdentityResolver.resolve",
162173 )(function* (cwd) {
163 const cacheKey = yield* resolveRepositoryIdentityCacheKey(cwd).pipe(
164 Effect.provideService(ProcessRunner.ProcessRunner, processRunner),
165 );
174 const cacheKey = yield* Cache.get(repositoryRootCache, cwd);
175 if (cacheKey === null) return null;
166176 return yield* Cache.get(repositoryIdentityCache, cacheKey);
167177 });
168178
apps/server/src/provider/Layers/EventNdjsonLogger.test.ts+42-2
......@@ -286,7 +286,7 @@ describe("EventNdjsonLogger", () => {
286286 }),
287287 );
288288
289 it.effect("drops transient canonical events before serialization", () =>
289 it.effect("drops transient provider events before serialization", () =>
290290 Effect.gen(function* () {
291291 const tempDir = NodeFS.mkdtempSync(NodePath.join(NodeOS.tmpdir(), "t3-provider-log-"));
292292 const basePath = NodePath.join(tempDir, "events.log");
......@@ -302,6 +302,46 @@ describe("EventNdjsonLogger", () => {
302302 yield* canonical.write(circularDelta, threadId);
303303 yield* canonical.write({ type: "item.completed", id: "final" }, threadId);
304304 yield* native.write({ type: "content.delta", id: "native-delta" }, threadId);
305 yield* native.write(
306 { method: "item/agentMessage/delta", payload: circularDelta },
307 threadId,
308 );
309 yield* native.write(
310 { method: "thread/realtime/outputAudio/delta", payload: circularDelta },
311 threadId,
312 );
313 yield* native.write(
314 { method: "thread/realtime/transcript/delta", payload: circularDelta },
315 threadId,
316 );
317 yield* native.write(
318 {
319 event: {
320 method: "claude/stream_event/content_block_delta/text_delta",
321 payload: circularDelta,
322 },
323 },
324 threadId,
325 );
326 yield* native.write(
327 {
328 event: {
329 method: "session/update",
330 payload: { update: { sessionUpdate: "agent_message_chunk" } },
331 },
332 },
333 threadId,
334 );
335 yield* native.write(
336 {
337 event: {
338 type: "message.part.updated",
339 payload: { properties: { part: { type: "text" } } },
340 },
341 },
342 threadId,
343 );
344 yield* native.write({ type: "turn.completed", id: "native-final" }, threadId);
305345 yield* store.close();
306346
307347 const lines = NodeFS.readFileSync(ownedLogPath(basePath, "thread-filtered"), "utf8")
......@@ -313,7 +353,7 @@ describe("EventNdjsonLogger", () => {
313353 lines.map(({ stream, payload }) => ({ stream, payload })),
314354 [
315355 { stream: "CANON", payload: '{"type":"item.completed","id":"final"}' },
316 { stream: "NTIVE", payload: '{"type":"content.delta","id":"native-delta"}' },
356 { stream: "NTIVE", payload: '{"type":"turn.completed","id":"native-final"}' },
317357 ],
318358 );
319359 } finally {
apps/server/src/provider/Layers/EventNdjsonLogger.ts+54-7
......@@ -45,6 +45,17 @@ const transientCanonicalEventTypes = new Set([
4545 "tool.progress",
4646 "turn.proposed.delta",
4747]);
48const transientNativeMethods = new Set([
49 "item/agentMessage/delta",
50 "item/commandExecution/outputDelta",
51 "item/fileChange/outputDelta",
52 "item/plan/delta",
53 "item/reasoning/summaryTextDelta",
54 "item/reasoning/textDelta",
55 "thread/realtime/outputAudio/delta",
56 "thread/realtime/transcript/delta",
57]);
58const transientAcpUpdates = new Set(["agent_message_chunk", "agent_thought_chunk"]);
4859
4960export type EventNdjsonStream = "native" | "canonical" | "orchestration";
5061
......@@ -126,7 +137,7 @@ export interface PendingRecord {
126137}
127138
128139interface StoreState {
129 readonly pending: ReadonlyArray<PendingRecord>;
140 readonly pending: Array<PendingRecord>;
130141 readonly pendingBytes: number;
131142 readonly sinks: ReadonlyMap<string, RotatingFileSink>;
132143 readonly flushScheduled: boolean;
......@@ -178,12 +189,50 @@ function providerLogPath(directory: string, prefix: string, threadSegment: strin
178189}
179190
180191function shouldPersist(stream: EventNdjsonStream, event: unknown): boolean {
181 if (stream !== "canonical" || typeof event !== "object" || event === null) {
192 if (stream === "orchestration" || typeof event !== "object" || event === null) {
182193 return true;
183194 }
184195 try {
185196 const type = Reflect.get(event, "type");
186 return typeof type !== "string" || !transientCanonicalEventTypes.has(type);
197 if (typeof type === "string" && transientCanonicalEventTypes.has(type)) {
198 return false;
199 }
200 if (stream !== "native") return true;
201
202 const nested = Reflect.get(event, "event");
203 const nativeEvent = typeof nested === "object" && nested !== null ? nested : event;
204 const method = Reflect.get(nativeEvent, "method");
205 if (
206 typeof method === "string" &&
207 (transientNativeMethods.has(method) ||
208 method.startsWith("claude/stream_event/content_block_delta/"))
209 ) {
210 return false;
211 }
212
213 const nativeType = Reflect.get(nativeEvent, "type");
214 if (nativeType === "message.part.delta") return false;
215
216 const payload = Reflect.get(nativeEvent, "payload");
217 if (typeof payload !== "object" || payload === null) return true;
218
219 if (method === "session/update") {
220 const update = Reflect.get(payload, "update");
221 if (typeof update !== "object" || update === null) return true;
222 const updateType = Reflect.get(update, "sessionUpdate");
223 return typeof updateType !== "string" || !transientAcpUpdates.has(updateType);
224 }
225
226 if (nativeType === "message.part.updated") {
227 const properties = Reflect.get(payload, "properties");
228 if (typeof properties !== "object" || properties === null) return true;
229 const part = Reflect.get(properties, "part");
230 if (typeof part !== "object" || part === null) return true;
231 const partType = Reflect.get(part, "type");
232 return partType !== "text" && partType !== "reasoning";
233 }
234
235 return true;
187236 } catch {
188237 return true;
189238 }
......@@ -566,10 +615,8 @@ export const makeEventNdjsonLogStore = Effect.fnUntraced(function* (
566615 if (state.closed) {
567616 return Effect.succeed([{ flush: false }, state] as const);
568617 }
569 const pending = [
570 ...state.pending,
571 { stream, threadSegment: resolveThreadSegment(threadId), line, bytes },
572 ];
618 const pending = state.pending;
619 pending.push({ stream, threadSegment: resolveThreadSegment(threadId), line, bytes });
573620 const pendingBytes = state.pendingBytes + bytes;
574621 const flush =
575622 resolved.batchWindowMs === 0 ||
apps/server/src/provider/Layers/OpenCodeAdapter.test.ts+15
......@@ -4607,12 +4607,27 @@ it.layer(OpenCodeAdapterTestLayer)("OpenCodeAdapterLive", (it) => {
46074607 const firstUpdate = mergeOpenCodeAssistantText(undefined, "Hello");
46084608 const overlapDelta = appendOpenCodeAssistantTextDelta(firstUpdate.latestText, "lo world");
46094609 const secondUpdate = mergeOpenCodeAssistantText(overlapDelta.nextText, "Hellolo world");
4610 const appendedUpdate = mergeOpenCodeAssistantText("Hello", "Hello world");
4611 const changedUpdate = mergeOpenCodeAssistantText("Hello world", "Hello there");
4612 const staleUpdate = mergeOpenCodeAssistantText("Hello world", "Hello");
46104613
46114614 NodeAssert.deepEqual(
46124615 [firstUpdate.deltaToEmit, overlapDelta.deltaToEmit, secondUpdate.deltaToEmit],
46134616 ["Hello", "lo world", ""],
46144617 );
46154618 NodeAssert.equal(secondUpdate.latestText, "Hellolo world");
4619 NodeAssert.deepEqual(appendedUpdate, {
4620 latestText: "Hello world",
4621 deltaToEmit: " world",
4622 });
4623 NodeAssert.deepEqual(changedUpdate, {
4624 latestText: "Hello there",
4625 deltaToEmit: "there",
4626 });
4627 NodeAssert.deepEqual(staleUpdate, {
4628 latestText: "Hello world",
4629 deltaToEmit: "",
4630 });
46164631 }),
46174632 );
46184633
apps/server/src/provider/Layers/OpenCodeAdapter.ts+5-1
......@@ -565,9 +565,13 @@ export function mergeOpenCodeAssistantText(
565565 readonly deltaToEmit: string;
566566} {
567567 const latestText = resolveLatestAssistantText(previousText, nextText);
568 const previous = previousText ?? "";
569 const prefixLength = latestText.startsWith(previous)
570 ? previous.length
571 : commonPrefixLength(previous, latestText);
568572 return {
569573 latestText,
570 deltaToEmit: latestText.slice(commonPrefixLength(previousText ?? "", latestText)),
574 deltaToEmit: latestText.slice(prefixLength),
571575 };
572576}
573577
apps/server/src/provider/acp/AcpNativeLogging.test.ts+169
......@@ -28,6 +28,7 @@ nodeServicesIt("ACP native logging", (it) => {
2828 nativeEventLogger,
2929 provider: ProviderDriverKind.make("cursor"),
3030 threadId: ThreadId.make("thread-1"),
31 verboseProtocolLogging: true,
3132 });
3233 const secret = "secret-token-value";
3334 const requestLogger = logger.requestLogger;
......@@ -67,6 +68,174 @@ nodeServicesIt("ACP native logging", (it) => {
6768 }),
6869 );
6970
71 it.effect("keeps request diagnostics without enabling full protocol logging", () =>
72 Effect.gen(function* () {
73 const records: Array<unknown> = [];
74 const makeLogger = yield* makeAcpNativeLoggerFactory();
75 const logger = makeLogger({
76 nativeEventLogger: {
77 filePath: "/tmp/provider-native.ndjson",
78 write: (event) => Effect.sync(() => void records.push(event)),
79 close: () => Effect.void,
80 },
81 provider: ProviderDriverKind.make("grok"),
82 threadId: ThreadId.make("thread-1"),
83 });
84
85 assert.isUndefined(logger.protocolLogging);
86 const requestLogger = logger.requestLogger;
87 assert.exists(requestLogger);
88 if (!requestLogger) return;
89 yield* requestLogger({
90 method: "session/prompt",
91 payload: {},
92 status: "started",
93 });
94 assert.lengthOf(records, 1);
95 }),
96 );
97
98 it.effect("drops transient ACP chunks before formatting verbose protocol logs", () =>
99 Effect.gen(function* () {
100 const records: Array<unknown> = [];
101 const makeLogger = yield* makeAcpNativeLoggerFactory();
102 const logger = makeLogger({
103 nativeEventLogger: {
104 filePath: "/tmp/provider-native.ndjson",
105 write: (event) => Effect.sync(() => void records.push(event)),
106 close: () => Effect.void,
107 },
108 provider: ProviderDriverKind.make("cursor"),
109 threadId: ThreadId.make("thread-1"),
110 verboseProtocolLogging: true,
111 });
112 const protocolLogger = logger.protocolLogging?.logger;
113 assert.exists(protocolLogger);
114 if (!protocolLogger) return;
115
116 for (const updateType of ["agent_message_chunk", "agent_thought_chunk"] as const) {
117 yield* protocolLogger({
118 direction: "incoming",
119 stage: "raw",
120 payload: `${encodeUnknownJson({
121 method: "session/update",
122 params: { update: { sessionUpdate: updateType } },
123 })}\n`,
124 });
125 yield* protocolLogger({
126 direction: "incoming",
127 stage: "decoded",
128 payload: [
129 {
130 _tag: "Request",
131 tag: "session/update",
132 payload: { update: { sessionUpdate: updateType } },
133 },
134 ],
135 });
136 }
137
138 assert.lengthOf(records, 0);
139
140 yield* protocolLogger({
141 direction: "incoming",
142 stage: "decoded",
143 payload: [
144 {
145 _tag: "Request",
146 tag: "session/update",
147 payload: { update: { sessionUpdate: "tool_call" } },
148 },
149 ],
150 });
151 assert.lengthOf(records, 1);
152 }),
153 );
154
155 it.effect("keeps mixed and incomplete raw diagnostics", () =>
156 Effect.gen(function* () {
157 const records: Array<unknown> = [];
158 const makeLogger = yield* makeAcpNativeLoggerFactory();
159 const logger = makeLogger({
160 nativeEventLogger: {
161 filePath: "/tmp/provider-native.ndjson",
162 write: (event) => Effect.sync(() => void records.push(event)),
163 close: () => Effect.void,
164 },
165 provider: ProviderDriverKind.make("cursor"),
166 threadId: ThreadId.make("thread-1"),
167 verboseProtocolLogging: true,
168 });
169 const protocolLogger = logger.protocolLogging?.logger;
170 assert.exists(protocolLogger);
171 if (!protocolLogger) return;
172
173 const transient = encodeUnknownJson({
174 method: "session/update",
175 params: { update: { sessionUpdate: "agent_message_chunk" } },
176 });
177 const lifecycle = encodeUnknownJson({ method: "session/new", params: {} });
178
179 yield* protocolLogger({
180 direction: "incoming",
181 stage: "raw",
182 payload: `${transient}\n${lifecycle}\n`,
183 });
184 yield* protocolLogger({
185 direction: "incoming",
186 stage: "raw",
187 payload: transient,
188 });
189 yield* protocolLogger({
190 direction: "incoming",
191 stage: "raw",
192 payload: `${transient}\n{malformed}\n`,
193 });
194
195 assert.lengthOf(records, 3);
196 }),
197 );
198
199 it.effect("filters transient entries from mixed decoded batches", () =>
200 Effect.gen(function* () {
201 const records: Array<unknown> = [];
202 const makeLogger = yield* makeAcpNativeLoggerFactory();
203 const logger = makeLogger({
204 nativeEventLogger: {
205 filePath: "/tmp/provider-native.ndjson",
206 write: (event) => Effect.sync(() => void records.push(event)),
207 close: () => Effect.void,
208 },
209 provider: ProviderDriverKind.make("grok"),
210 threadId: ThreadId.make("thread-1"),
211 verboseProtocolLogging: true,
212 });
213 const protocolLogger = logger.protocolLogging?.logger;
214 assert.exists(protocolLogger);
215 if (!protocolLogger) return;
216
217 yield* protocolLogger({
218 direction: "incoming",
219 stage: "decoded",
220 payload: [
221 {
222 _tag: "Request",
223 tag: "session/update",
224 payload: { update: { sessionUpdate: "agent_thought_chunk" } },
225 },
226 {
227 _tag: "Request",
228 tag: "session/new",
229 payload: {},
230 },
231 ],
232 });
233
234 assert.lengthOf(records, 1);
235 assert.include(encodeUnknownJson(records), '"itemCount":1');
236 }),
237 );
238
70239 it.effect("logs a structural tag when the native writer defects", () => {
71240 const messages: Array<unknown> = [];
72241 const logCapture = Logger.make<unknown, void>(({ message }) => {
apps/server/src/provider/acp/AcpNativeLogging.ts+61-6
......@@ -9,6 +9,8 @@ import type * as EffectAcpProtocol from "effect-acp/protocol";
99import type { EventNdjsonLogger } from "../Layers/EventNdjsonLogger.ts";
1010import type * as AcpSessionRuntime from "./AcpSessionRuntime.ts";
1111
12const transientProtocolUpdates = new Set(["agent_message_chunk", "agent_thought_chunk"]);
13
1214function structuralMethod(value: string): string {
1315 return value.length <= 128 && /^[A-Za-z][A-Za-z0-9._:/-]*$/.test(value) ? value : "unknown";
1416}
......@@ -64,12 +66,61 @@ function formatProtocolLogPayload(event: EffectAcpProtocol.AcpProtocolLogEvent)
6466 };
6567}
6668
69function isTransientProtocolMessage(message: unknown): boolean {
70 if (typeof message !== "object" || message === null) return false;
71 const method = Reflect.get(message, "tag") ?? Reflect.get(message, "method");
72 if (method !== "session/update") return false;
73
74 const payload = Reflect.get(message, "payload") ?? Reflect.get(message, "params");
75 if (typeof payload !== "object" || payload === null) return false;
76 const update = Reflect.get(payload, "update");
77 if (typeof update !== "object" || update === null) return false;
78 const updateType = Reflect.get(update, "sessionUpdate");
79 return typeof updateType === "string" && transientProtocolUpdates.has(updateType);
80}
81
82function rawChunkContainsOnlyTransientMessages(payload: string): boolean {
83 const lines = payload.split("\n");
84 const remainder = lines.pop() ?? "";
85 if (remainder.trim().length > 0) return false;
86
87 const messages: Array<unknown> = [];
88 for (const line of lines) {
89 if (line.trim().length === 0) continue;
90 try {
91 messages.push(JSON.parse(line));
92 } catch {
93 return false;
94 }
95 }
96 return messages.length > 0 && messages.every(isTransientProtocolMessage);
97}
98
99function filterTransientProtocolLog(
100 event: EffectAcpProtocol.AcpProtocolLogEvent,
101): EffectAcpProtocol.AcpProtocolLogEvent | undefined {
102 if (event.direction !== "incoming") return event;
103
104 if (event.stage === "raw" && typeof event.payload === "string") {
105 return rawChunkContainsOnlyTransientMessages(event.payload) ? undefined : event;
106 }
107
108 if (event.stage !== "decoded") return event;
109 if (!Array.isArray(event.payload)) {
110 return isTransientProtocolMessage(event.payload) ? undefined : event;
111 }
112
113 const payload = event.payload.filter((message) => !isTransientProtocolMessage(message));
114 return payload.length === 0 ? undefined : { ...event, payload };
115}
116
67117export const makeAcpNativeLoggerFactory = Effect.fn("makeAcpNativeLoggerFactory")(function* () {
68118 const crypto = yield* Crypto.Crypto;
69119 return (input: {
70120 readonly nativeEventLogger: EventNdjsonLogger | undefined;
71121 readonly provider: ProviderDriverKind;
72122 readonly threadId: ThreadId;
123 readonly verboseProtocolLogging?: boolean;
73124 }): Pick<AcpSessionRuntime.AcpSessionRuntimeOptions, "requestLogger" | "protocolLogging"> => {
74125 const writeNativeAcpLog = (logInput: {
75126 readonly kind: "request" | "protocol";
......@@ -111,16 +162,20 @@ export const makeAcpNativeLoggerFactory = Effect.fn("makeAcpNativeLoggerFactory"
111162 kind: "request",
112163 payload: formatRequestLogPayload(event),
113164 }),
114 ...(input.nativeEventLogger
165 ...(input.nativeEventLogger && input.verboseProtocolLogging
115166 ? {
116167 protocolLogging: {
117168 logIncoming: true,
118169 logOutgoing: true,
119 logger: (event: EffectAcpProtocol.AcpProtocolLogEvent) =>
120 writeNativeAcpLog({
121 kind: "protocol",
122 payload: formatProtocolLogPayload(event),
123 }),
170 logger: (event: EffectAcpProtocol.AcpProtocolLogEvent) => {
171 const filtered = filterTransientProtocolLog(event);
172 return filtered
173 ? writeNativeAcpLog({
174 kind: "protocol",
175 payload: formatProtocolLogPayload(filtered),
176 })
177 : Effect.void;
178 },
124179 } satisfies NonNullable<AcpSessionRuntime.AcpSessionRuntimeOptions["protocolLogging"]>,
125180 }
126181 : {}),
apps/server/src/resourceTelemetry/NativeTelemetryClient.test.ts+3-2
......@@ -44,7 +44,7 @@ describe("resolveNativeSampleIntervalMs", () => {
4444 expect(resolveNativeSampleIntervalMs({ ...basePower, onBattery: "true" }, 1)).toBe(5_000);
4545 });
4646
47 it("keeps unknown background telemetry cheap but serves live diagnostics at 1Hz", () => {
47 it("slows background telemetry and serves live diagnostics at 1Hz", () => {
4848 const unknown: HostPowerSnapshot = {
4949 ...basePower,
5050 source: "unknown",
......@@ -58,7 +58,8 @@ describe("resolveNativeSampleIntervalMs", () => {
5858 0,
5959 ),
6060 ).toBe(5_000);
61 expect(resolveNativeSampleIntervalMs(basePower, 0)).toBe(1_000);
61 expect(resolveNativeSampleIntervalMs(basePower, 0)).toBe(5_000);
62 expect(resolveNativeSampleIntervalMs(basePower, 1)).toBe(1_000);
6263 });
6364});
6465
apps/server/src/resourceTelemetry/NativeTelemetryClient.ts+23-17
......@@ -268,7 +268,7 @@ export function resolveNativeSampleIntervalMs(
268268 return CONSTRAINED_SAMPLE_INTERVAL_MS;
269269 }
270270 if (snapshot.onBattery === "true") return BATTERY_SAMPLE_INTERVAL_MS;
271 return SAMPLE_INTERVAL_MS;
271 return liveSubscriberCount > 0 ? SAMPLE_INTERVAL_MS : UNKNOWN_BACKGROUND_SAMPLE_INTERVAL_MS;
272272}
273273
274274export function commitCollectionControlUpdate<E, R>(
......@@ -462,13 +462,16 @@ export const make = Effect.fn("resourceTelemetry.nativeTelemetryClient.make")(fu
462462 return Effect.gen(function* () {
463463 const nativeSnapshot = { generation, snapshot: event } satisfies NativeTelemetrySnapshot;
464464 const sampledAt = DateTime.makeUnsafe(event.sampledAtUnixMs);
465 yield* Ref.update(state, (current) => ({
466 ...current,
467 status: "healthy" as const,
468 lastSampleAt: Option.some(sampledAt),
469 lastError: Option.none(),
470 }));
471 yield* publishHealth;
465 const healthChanged = yield* Ref.modify(state, (current) => [
466 current.status !== "healthy" || Option.isSome(current.lastError),
467 {
468 ...current,
469 status: "healthy" as const,
470 lastSampleAt: Option.some(sampledAt),
471 lastError: Option.none(),
472 },
473 ]);
474 if (healthChanged) yield* publishHealth;
472475 yield* PubSub.publish(snapshots, nativeSnapshot);
473476 if (event.requestId) {
474477 const deferred = yield* Ref.modify(pendingSamples, (pending) => {
......@@ -485,15 +488,18 @@ export const make = Effect.fn("resourceTelemetry.nativeTelemetryClient.make")(fu
485488 case "historyChunk":
486489 return Effect.gen(function* () {
487490 const latestSnapshot = event.snapshots.at(-1);
488 yield* Ref.update(state, (current) => ({
489 ...current,
490 status: "healthy" as const,
491 lastSampleAt: latestSnapshot
492 ? Option.some(DateTime.makeUnsafe(latestSnapshot.sampledAtUnixMs))
493 : current.lastSampleAt,
494 lastError: Option.none(),
495 }));
496 yield* publishHealth;
491 const healthChanged = yield* Ref.modify(state, (current) => [
492 current.status !== "healthy" || Option.isSome(current.lastError),
493 {
494 ...current,
495 status: "healthy" as const,
496 lastSampleAt: latestSnapshot
497 ? Option.some(DateTime.makeUnsafe(latestSnapshot.sampledAtUnixMs))
498 : current.lastSampleAt,
499 lastError: Option.none(),
500 },
501 ]);
502 if (healthChanged) yield* publishHealth;
497503 const completed = yield* Ref.modify(pendingHistories, (pending) => {
498504 const request = pending.get(event.requestId);
499505 if (!request) return [Option.none(), pending] as const;
apps/server/src/vcs/GitVcsDriverCore.ts+2-1
......@@ -475,6 +475,7 @@ function trace2ChildKey(record: Record<string, unknown>): string | null {
475475}
476476
477477const Trace2Record = Schema.Record(Schema.String, Schema.Unknown);
478const decodeTrace2Record = decodeJsonResult(Trace2Record);
478479
479480const createTrace2Monitor = Effect.fn("createTrace2Monitor")(function* (
480481 input: Pick<GitVcsDriver.ExecuteGitInput, "operation" | "cwd" | "args">,
......@@ -509,7 +510,7 @@ const createTrace2Monitor = Effect.fn("createTrace2Monitor")(function* (
509510 return;
510511 }
511512
512 const traceRecord = decodeJsonResult(Trace2Record)(trimmedLine);
513 const traceRecord = decodeTrace2Record(trimmedLine);
513514 if (Result.isFailure(traceRecord)) {
514515 yield* Effect.logDebug(
515516 `GitVcsDriver.trace2: failed to parse trace line for ${input.operation} in ${input.cwd} (${input.args.length} arguments)`,
docs/internals/resource-telemetry.md+5-4
......@@ -103,9 +103,9 @@ power-adaptive interval selected by the server. It collects:
103103- resident and virtual memory;
104104- cumulative process I/O counters.
105105
106On Linux, task/thread enumeration is disabled. Command lines are loaded only
107when first needed. This avoids the expensive default behavior of walking every
108`/proc/<pid>/task/<tid>` directory on each refresh.
106On Linux, task/thread enumeration is disabled. Command lines are refreshed with
107each sample so process replacements remain visible. Disabling task enumeration
108avoids walking every `/proc/<pid>/task/<tid>` directory on each refresh.
109109
110110### Process-tree selection
111111
......@@ -142,7 +142,8 @@ The server adjusts native sampling without restarting the sidecar:
142142
143143- suspended, locked, low-power, or serious/critical thermal state: 15 seconds;
144144- battery: 5 seconds;
145- normal AC: 1 second;
145- normal AC: 5 seconds in the background and 1 second while live diagnostics is
146 open;
146147- unknown or stale power: 5 seconds in the background and 1 second while live
147148 diagnostics is open.
148149
native/resource-monitor/src/main.rs+22-13
......@@ -250,6 +250,10 @@ impl HistoryRecorder {
250250 max_retained_entries: usize,
251251 max_retained_bytes: usize,
252252 ) {
253 let clock_moved_backward = self
254 .snapshots
255 .back()
256 .is_some_and(|previous| previous.sampled_at_unix_ms > snapshot.sampled_at_unix_ms);
253257 let mut retained = snapshot.clone();
254258 retained.request_id = None;
255259 self.retained_entry_count = self
......@@ -264,6 +268,7 @@ impl HistoryRecorder {
264268 max_snapshots,
265269 max_retained_entries,
266270 max_retained_bytes,
271 clock_moved_backward,
267272 );
268273 }
269274
......@@ -273,20 +278,24 @@ impl HistoryRecorder {
273278 max_snapshots: usize,
274279 max_retained_entries: usize,
275280 max_retained_bytes: usize,
281 clock_moved_backward: bool,
276282 ) {
277 let mut future_entry_count = 0usize;
278 let mut future_bytes = 0usize;
279 self.snapshots.retain(|snapshot| {
280 let keep = snapshot.sampled_at_unix_ms <= now_ms;
281 if !keep {
282 future_entry_count =
283 future_entry_count.saturating_add(snapshot.retained_entry_count());
284 future_bytes = future_bytes.saturating_add(snapshot.estimated_history_bytes());
285 }
286 keep
287 });
288 self.retained_entry_count = self.retained_entry_count.saturating_sub(future_entry_count);
289 self.retained_bytes = self.retained_bytes.saturating_sub(future_bytes);
283 if clock_moved_backward {
284 let mut future_entry_count = 0usize;
285 let mut future_bytes = 0usize;
286 self.snapshots.retain(|snapshot| {
287 let keep = snapshot.sampled_at_unix_ms <= now_ms;
288 if !keep {
289 future_entry_count =
290 future_entry_count.saturating_add(snapshot.retained_entry_count());
291 future_bytes = future_bytes.saturating_add(snapshot.estimated_history_bytes());
292 }
293 keep
294 });
295 self.retained_entry_count =
296 self.retained_entry_count.saturating_sub(future_entry_count);
297 self.retained_bytes = self.retained_bytes.saturating_sub(future_bytes);
298 }
290299
291300 while self.snapshots.front().is_some_and(|snapshot| {
292301 snapshot.sampled_at_unix_ms < now_ms.saturating_sub(HISTORY_RETENTION_MS)
packages/effect-acp/src/protocol.test.ts+35
......@@ -135,6 +135,41 @@ it.layer(NodeServices.layer)("effect-acp protocol", (it) => {
135135 }),
136136 );
137137
138 it.effect("keeps only recent raw notifications after their callbacks run", () =>
139 Effect.gen(function* () {
140 const { stdio, input } = yield* makeInMemoryStdio();
141 const handled = yield* Deferred.make<void>();
142 let handledCount = 0;
143 const transport = yield* AcpProtocol.makeAcpPatchedProtocol({
144 stdio,
145 serverRequestMethods: new Set(),
146 onNotification: () =>
147 Effect.sync(() => ++handledCount).pipe(
148 Effect.flatMap((count) =>
149 count === 64 ? Deferred.succeed(handled, undefined).pipe(Effect.asVoid) : Effect.void,
150 ),
151 ),
152 });
153
154 const messages = Array.from({ length: 64 }, (_, index) =>
155 encodeUnknownJsonString({
156 jsonrpc: "2.0",
157 method: "x/performance",
158 params: { index },
159 }),
160 );
161 yield* Queue.offer(input, encoder.encode(`${messages.join("\n")}\n`));
162 yield* Deferred.await(handled);
163
164 const retained = yield* transport.incoming.pipe(Stream.take(32), Stream.runCollect);
165
166 assert.equal(handledCount, 64);
167 assert.equal(retained.length, 32);
168 assert.deepEqual(retained[0]?.params, { index: 32 });
169 assert.deepEqual(retained[31]?.params, { index: 63 });
170 }),
171 );
172
138173 it.effect("keeps invalid core notification values only in the schema cause", () =>
139174 Effect.gen(function* () {
140175 const secret = "acp-core-notification-secret-sentinel";
packages/effect-acp/src/protocol.ts+12-6
......@@ -76,6 +76,7 @@ const decodeElicitationComplete = Schema.decodeUnknownEffect(
7676 AcpSchema.ElicitationCompleteNotification,
7777);
7878const parserFactory = RpcSerialization.ndJsonRpc();
79const MAX_BUFFERED_RAW_NOTIFICATIONS = 32;
7980
8081export const makeAcpPatchedProtocol = Effect.fn("makeAcpPatchedProtocol")(function* (
8182 options: AcpPatchedProtocolOptions,
......@@ -83,7 +84,9 @@ export const makeAcpPatchedProtocol = Effect.fn("makeAcpPatchedProtocol")(functi
8384 const parser = parserFactory.makeUnsafe();
8485 const serverQueue = yield* Queue.unbounded<RpcMessage.FromClientEncoded>();
8586 const clientQueue = yield* Queue.unbounded<RpcMessage.FromServerEncoded>();
86 const notificationQueue = yield* Queue.unbounded<AcpIncomingNotification>();
87 const notificationQueue = yield* Queue.sliding<AcpIncomingNotification>(
88 MAX_BUFFERED_RAW_NOTIFICATIONS,
89 );
8790 const disconnects = yield* Queue.unbounded<number>();
8891 const outgoing = yield* Queue.unbounded<string | Uint8Array, Cause.Done<void>>();
8992 const nextRequestId = yield* Ref.make(1);
......@@ -408,11 +411,14 @@ export const makeAcpPatchedProtocol = Effect.fn("makeAcpPatchedProtocol")(functi
408411
409412 yield* options.stdio.stdin.pipe(
410413 Stream.runForEach((data) =>
411 logProtocol({
412 direction: "incoming",
413 stage: "raw",
414 payload: typeof data === "string" ? data : new TextDecoder().decode(data),
415 }).pipe(
414 (options.logIncoming
415 ? logProtocol({
416 direction: "incoming",
417 stage: "raw",
418 payload: typeof data === "string" ? data : new TextDecoder().decode(data),
419 })
420 : Effect.void
421 ).pipe(
416422 Effect.flatMap(() =>
417423 Effect.try({
418424 try: () =>
packages/effect-codex-app-server/src/protocol.test.ts+252
......@@ -3,6 +3,7 @@ import * as Effect from "effect/Effect";
33import * as Fiber from "effect/Fiber";
44import * as Queue from "effect/Queue";
55import * as Schema from "effect/Schema";
6import * as Stdio from "effect/Stdio";
67import * as Stream from "effect/Stream";
78
89import * as NodeServices from "@effect/platform-node/NodeServices";
......@@ -398,6 +399,257 @@ it.layer(NodeServices.layer)("effect-codex-app-server protocol", (it) => {
398399 }),
399400 );
400401
402 it.effect("keeps only recent raw notifications after their callbacks run", () =>
403 Effect.gen(function* () {
404 const { stdio, input } = yield* makeInMemoryStdio();
405 const handled = yield* Deferred.make<void>();
406 let handledCount = 0;
407 const transport = yield* CodexProtocol.makeCodexAppServerPatchedProtocol({
408 stdio,
409 onNotification: () =>
410 Effect.sync(() => ++handledCount).pipe(
411 Effect.flatMap((count) =>
412 count === 64 ? Deferred.succeed(handled, undefined).pipe(Effect.asVoid) : Effect.void,
413 ),
414 ),
415 });
416
417 const messages = Array.from({ length: 64 }, (_, index) =>
418 encodeUnknownJsonString({
419 method: "item/agentMessage/delta",
420 params: { index },
421 }),
422 );
423 yield* Queue.offer(input, encoder.encode(`${messages.join("\n")}\n`));
424 yield* Deferred.await(handled);
425
426 const retained = yield* transport.incomingNotifications.pipe(
427 Stream.take(32),
428 Stream.runCollect,
429 );
430
431 assert.equal(handledCount, 64);
432 assert.equal(retained.length, 32);
433 assert.deepEqual(retained[0]?.params, { index: 32 });
434 assert.deepEqual(retained[31]?.params, { index: 63 });
435 }),
436 );
437
438 it.effect("keeps processing protocol messages while an approval is pending", () =>
439 Effect.gen(function* () {
440 const { stdio, input, output } = yield* makeInMemoryStdio();
441 const approvalStarted = yield* Deferred.make<void>();
442 const approvalDecision = yield* Deferred.make<{ readonly decision: string }>();
443 const notificationReceived = yield* Deferred.make<void>();
444 const transport = yield* CodexProtocol.makeCodexAppServerPatchedProtocol({
445 stdio,
446 onRequest: () =>
447 Deferred.succeed(approvalStarted, undefined).pipe(
448 Effect.andThen(Deferred.await(approvalDecision)),
449 ),
450 onNotification: () => Deferred.succeed(notificationReceived, undefined).pipe(Effect.asVoid),
451 });
452
453 const pendingRequest = yield* transport.request("thread/read", {}).pipe(Effect.forkScoped);
454 yield* Queue.take(output);
455 yield* Queue.offer(
456 input,
457 encoder.encode(
458 `${[
459 encodeUnknownJsonString({ id: 7, method: "item/tool/requestUserInput", params: {} }),
460 encodeUnknownJsonString({ method: "item/agentMessage/delta", params: { delta: "ok" } }),
461 encodeUnknownJsonString({ id: 1, result: { threadId: "thread-1" } }),
462 ].join("\n")}\n`,
463 ),
464 );
465
466 yield* Deferred.await(approvalStarted);
467 yield* Deferred.await(notificationReceived);
468 assert.deepEqual(yield* Fiber.join(pendingRequest), { threadId: "thread-1" });
469
470 yield* Deferred.succeed(approvalDecision, { decision: "accept" });
471 assert.deepEqual(yield* decodeJson(yield* Queue.take(output)), {
472 id: 7,
473 result: { decision: "accept" },
474 });
475 }),
476 );
477
478 it.effect("rejects incoming requests after the active handler limit is reached", () =>
479 Effect.gen(function* () {
480 const { stdio, input, output } = yield* makeInMemoryStdio();
481 const handlersStarted = yield* Deferred.make<void>();
482 const releaseHandlers = yield* Deferred.make<void>();
483 let activeHandlers = 0;
484 yield* CodexProtocol.makeCodexAppServerPatchedProtocol({
485 stdio,
486 onRequest: () =>
487 Effect.sync(() => ++activeHandlers).pipe(
488 Effect.flatMap((count) =>
489 count === 32
490 ? Deferred.succeed(handlersStarted, undefined).pipe(Effect.asVoid)
491 : Effect.void,
492 ),
493 Effect.andThen(Deferred.await(releaseHandlers)),
494 Effect.as({ decision: "accept" }),
495 ),
496 });
497
498 const requests = Array.from({ length: 33 }, (_, index) =>
499 encodeUnknownJsonString({
500 id: index + 1,
501 method: "item/tool/requestUserInput",
502 params: {},
503 }),
504 );
505 yield* Queue.offer(input, encoder.encode(`${requests.join("\n")}\n`));
506 yield* Deferred.await(handlersStarted);
507
508 assert.deepEqual(yield* decodeJson(yield* Queue.take(output)), {
509 id: 33,
510 error: {
511 code: -32001,
512 message: "Too many Codex requests are already active.",
513 },
514 });
515 assert.equal(activeHandlers, 32);
516
517 yield* Deferred.succeed(releaseHandlers, undefined);
518 yield* Effect.forEach(Array.from({ length: 32 }), () => Queue.take(output), {
519 discard: true,
520 });
521 }),
522 );
523
524 it.effect("interrupts pending request handlers when the protocol terminates", () =>
525 Effect.gen(function* () {
526 const { stdio, input } = yield* makeInMemoryStdio();
527 const approvalStarted = yield* Deferred.make<void>();
528 const approvalInterrupted = yield* Deferred.make<void>();
529 const terminated = yield* Deferred.make<void>();
530 yield* CodexProtocol.makeCodexAppServerPatchedProtocol({
531 stdio,
532 onRequest: () =>
533 Deferred.succeed(approvalStarted, undefined).pipe(
534 Effect.andThen(Effect.never),
535 Effect.onInterrupt(() =>
536 Deferred.succeed(approvalInterrupted, undefined).pipe(Effect.asVoid),
537 ),
538 ),
539 onTermination: () => Deferred.succeed(terminated, undefined).pipe(Effect.asVoid),
540 });
541
542 yield* Queue.offer(
543 input,
544 encodeJsonl({ id: 7, method: "item/tool/requestUserInput", params: {} }),
545 );
546 yield* Deferred.await(approvalStarted);
547 yield* Queue.end(input);
548
549 yield* Deferred.await(approvalInterrupted);
550 yield* Deferred.await(terminated);
551 }),
552 );
553
554 it.effect("rejects outgoing messages after an approval response cannot be encoded", () =>
555 Effect.gen(function* () {
556 const { stdio: baseStdio, input } = yield* makeInMemoryStdio();
557 const terminated = yield* Deferred.make<CodexError.CodexAppServerError>();
558 const readerStopped = yield* Deferred.make<void>();
559 let notificationCount = 0;
560 let requestCount = 0;
561 const stdio = Stdio.make({
562 args: baseStdio.args,
563 stdin: baseStdio.stdin.pipe(
564 Stream.ensuring(Deferred.succeed(readerStopped, undefined).pipe(Effect.asVoid)),
565 ),
566 stdout: baseStdio.stdout,
567 stderr: baseStdio.stderr,
568 });
569 const transport = yield* CodexProtocol.makeCodexAppServerPatchedProtocol({
570 stdio,
571 onRequest: () =>
572 Effect.sync(() => ++requestCount).pipe(
573 Effect.map((count) => (count === 1 ? { invalid: 1n } : { ok: true })),
574 ),
575 onNotification: () => Effect.sync(() => notificationCount++).pipe(Effect.asVoid),
576 onTermination: (error) => Deferred.succeed(terminated, error).pipe(Effect.asVoid),
577 });
578
579 yield* Queue.offer(
580 input,
581 encodeJsonl({ id: 7, method: "item/tool/requestUserInput", params: {} }),
582 );
583
584 const failure = yield* Deferred.await(terminated);
585 assert.instanceOf(failure, CodexError.CodexAppServerProtocolParseError);
586 const requestFailure = yield* transport.request("thread/read", {}).pipe(
587 Effect.match({
588 onFailure: (error) => error,
589 onSuccess: () => assert.fail("Expected a terminated protocol request to fail"),
590 }),
591 );
592 const notificationFailure = yield* transport.notify("initialized").pipe(Effect.flip);
593 assert.strictEqual(requestFailure, failure);
594 assert.strictEqual(notificationFailure, failure);
595 yield* Deferred.await(readerStopped);
596
597 yield* Queue.offer(
598 input,
599 encoder.encode(
600 `${[
601 encodeUnknownJsonString({ method: "x/late-notification" }),
602 encodeUnknownJsonString({ id: 8, method: "x/late-request" }),
603 ].join("\n")}\n`,
604 ),
605 );
606
607 assert.equal(notificationCount, 0);
608 assert.equal(requestCount, 1);
609 assert.equal(yield* Queue.size(input), 1);
610 }),
611 );
612
613 it.effect("fails pending requests before interrupted handler cleanup completes", () =>
614 Effect.gen(function* () {
615 const { stdio, input, output } = yield* makeInMemoryStdio();
616 const handlerStarted = yield* Deferred.make<void>();
617 const finalizerStarted = yield* Deferred.make<void>();
618 const releaseFinalizer = yield* Deferred.make<void>();
619 const terminated = yield* Deferred.make<CodexError.CodexAppServerError>();
620 const transport = yield* CodexProtocol.makeCodexAppServerPatchedProtocol({
621 stdio,
622 onRequest: () =>
623 Deferred.succeed(handlerStarted, undefined).pipe(
624 Effect.andThen(Effect.never),
625 Effect.onInterrupt(() =>
626 Deferred.succeed(finalizerStarted, undefined).pipe(
627 Effect.andThen(Deferred.await(releaseFinalizer)),
628 ),
629 ),
630 ),
631 onTermination: (error) => Deferred.succeed(terminated, error).pipe(Effect.asVoid),
632 });
633 const pending = yield* transport.request("thread/read", {}).pipe(Effect.forkScoped);
634 yield* Queue.take(output);
635 yield* Queue.offer(input, encodeJsonl({ id: 7, method: "x/approval" }));
636 yield* Deferred.await(handlerStarted);
637 yield* Queue.end(input);
638
639 const failure = yield* Deferred.await(terminated);
640 yield* Deferred.await(finalizerStarted);
641 const pendingFailure = yield* Fiber.join(pending).pipe(
642 Effect.match({
643 onFailure: (error) => error,
644 onSuccess: () => assert.fail("Expected the pending request to fail"),
645 }),
646 );
647 assert.strictEqual(pendingFailure, failure);
648
649 yield* Deferred.succeed(releaseFinalizer, undefined);
650 }),
651 );
652
401653 it.effect("surfaces JSON encoding failures as protocol parse errors", () =>
402654 Effect.gen(function* () {
403655 const { stdio } = yield* makeInMemoryStdio();
packages/effect-codex-app-server/src/protocol.ts+70-25
......@@ -1,6 +1,8 @@
11import * as Cause from "effect/Cause";
22import * as Deferred from "effect/Deferred";
33import * as Effect from "effect/Effect";
4import * as Exit from "effect/Exit";
5import * as Option from "effect/Option";
46import * as Queue from "effect/Queue";
57import * as Ref from "effect/Ref";
68import * as Scope from "effect/Scope";
......@@ -13,6 +15,7 @@ import { JsonRpcId, JsonRpcResponseEnvelope } from "./_internal/shared.ts";
1315const isJsonRpcId = Schema.is(JsonRpcId);
1416const isJsonRpcResponseEnvelope = Schema.is(JsonRpcResponseEnvelope);
1517const isCodexAppServerError = Schema.is(CodexError.CodexAppServerError);
18const MAX_BUFFERED_RAW_MESSAGES = 32;
1619
1720export interface CodexAppServerProtocolLogEvent {
1821 readonly direction: "incoming" | "outgoing";
......@@ -152,13 +155,20 @@ export const makeCodexAppServerPatchedProtocol = Effect.fn("makeCodexAppServerPa
152155 function* (
153156 options: CodexAppServerPatchedProtocolOptions,
154157 ): Effect.fn.Return<CodexAppServerPatchedProtocol, never, Scope.Scope> {
158 const protocolScope = yield* Scope.Scope;
159 const requestHandlerScope = yield* Scope.fork(protocolScope, "parallel");
155160 const outgoing = yield* Queue.unbounded<string, Cause.Done<void>>();
156 const incomingNotifications = yield* Queue.unbounded<CodexAppServerIncomingNotification>();
157 const incomingRequests = yield* Queue.unbounded<CodexAppServerIncomingRequest>();
161 const incomingNotifications =
162 yield* Queue.sliding<CodexAppServerIncomingNotification>(MAX_BUFFERED_RAW_MESSAGES);
163 const incomingRequests =
164 yield* Queue.sliding<CodexAppServerIncomingRequest>(MAX_BUFFERED_RAW_MESSAGES);
158165 const pending = yield* Ref.make(new Map<string, CodexAppServerPendingRequest>());
159166 const nextRequestId = yield* Ref.make(1);
160167 const remainder: Array<string> = [];
161168 const terminationHandled = yield* Ref.make(false);
169 const terminationFailure = yield* Ref.make(Option.none<CodexError.CodexAppServerError>());
170 const terminationSignal = yield* Deferred.make<void>();
171 const activeRequestHandlers = yield* Ref.make(0);
162172
163173 const logProtocol = (event: CodexAppServerProtocolLogEvent) => {
164174 if (event.direction === "incoming" && !options.logIncoming) {
......@@ -191,8 +201,14 @@ export const makeCodexAppServerPatchedProtocol = Effect.fn("makeCodexAppServerPa
191201 return [
192202 Effect.gen(function* () {
193203 const error = yield* classify();
204 yield* Ref.set(terminationFailure, Option.some(error));
194205 yield* failAllPending(error);
195206 yield* Queue.end(outgoing);
207 yield* Deferred.succeed(terminationSignal, undefined);
208 yield* Scope.close(requestHandlerScope, Exit.void).pipe(
209 Effect.forkIn(protocolScope, { startImmediately: true }),
210 Effect.asVoid,
211 );
196212 if (options.onTermination) {
197213 yield* options.onTermination(error);
198214 }
......@@ -203,6 +219,9 @@ export const makeCodexAppServerPatchedProtocol = Effect.fn("makeCodexAppServerPa
203219
204220 const offerOutgoing = (message: Record<string, unknown>) =>
205221 Effect.gen(function* () {
222 const failure = yield* Ref.get(terminationFailure);
223 if (Option.isSome(failure)) return yield* failure.value;
224
206225 yield* logProtocol({
207226 direction: "outgoing",
208227 stage: "decoded",
......@@ -214,7 +233,14 @@ export const makeCodexAppServerPatchedProtocol = Effect.fn("makeCodexAppServerPa
214233 stage: "raw",
215234 payload: encoded,
216235 });
217 yield* Queue.offer(outgoing, encoded).pipe(Effect.asVoid);
236 const accepted = yield* Queue.offer(outgoing, encoded);
237 if (!accepted) {
238 const closed = yield* Ref.get(terminationFailure);
239 return yield* Option.getOrElse(
240 closed,
241 () => new CodexError.CodexAppServerInputStreamEndedError({}),
242 );
243 }
218244 });
219245
220246 const removePending = (requestId: string) =>
......@@ -271,9 +297,24 @@ export const makeCodexAppServerPatchedProtocol = Effect.fn("makeCodexAppServerPa
271297
272298 const handleRequest = (request: CodexAppServerIncomingRequest) =>
273299 Queue.offer(incomingRequests, request).pipe(
274 Effect.andThen(
275 options.onRequest
276 ? options.onRequest(request).pipe(
300 Effect.flatMap(() => {
301 const handler = options.onRequest;
302 if (!handler) return Effect.void;
303
304 return Ref.modify(activeRequestHandlers, (count) =>
305 count >= MAX_BUFFERED_RAW_MESSAGES ? [false, count] : [true, count + 1],
306 ).pipe(
307 Effect.flatMap((accepted) => {
308 if (!accepted) {
309 return respondError(
310 request.id,
311 CodexError.CodexAppServerRequestError.overloaded(
312 "Too many Codex requests are already active.",
313 ),
314 );
315 }
316
317 return handler(request).pipe(
277318 Effect.matchEffect({
278319 onFailure: (error) =>
279320 respondError(
......@@ -285,9 +326,21 @@ export const makeCodexAppServerPatchedProtocol = Effect.fn("makeCodexAppServerPa
285326 ),
286327 onSuccess: (result) => respond(request.id, result),
287328 }),
288 )
289 : Effect.void,
290 ),
329 Effect.ensuring(
330 Ref.update(activeRequestHandlers, (count) => Math.max(0, count - 1)),
331 ),
332 Effect.catch((error) =>
333 handleTermination(() => Effect.succeed(error)).pipe(
334 Effect.forkIn(protocolScope),
335 Effect.asVoid,
336 ),
337 ),
338 Effect.forkIn(requestHandlerScope, { startImmediately: true }),
339 Effect.asVoid,
340 );
341 }),
342 );
343 }),
291344 Effect.asVoid,
292345 );
293346
......@@ -297,22 +350,13 @@ export const makeCodexAppServerPatchedProtocol = Effect.fn("makeCodexAppServerPa
297350 Effect.asVoid,
298351 );
299352
300 const routeMessage = (
301 message: unknown,
302 ): Effect.Effect<void, CodexError.CodexAppServerError> => {
303 if (isIncomingRequest(message)) {
304 return handleRequest(message);
305 }
306 if (isIncomingNotification(message)) {
307 return handleNotification(message);
308 }
309 if (isIncomingResponse(message)) {
310 return handleResponse(message);
311 }
312 return Effect.fail(
313 CodexError.CodexAppServerProtocolParseError.fromUnroutableMessage(message),
314 );
315 };
353 const routeMessage = Effect.fnUntraced(function* (message: unknown) {
354 if (Option.isSome(yield* Ref.get(terminationFailure))) return;
355 if (isIncomingRequest(message)) return yield* handleRequest(message);
356 if (isIncomingNotification(message)) return yield* handleNotification(message);
357 if (isIncomingResponse(message)) return yield* handleResponse(message);
358 return yield* CodexError.CodexAppServerProtocolParseError.fromUnroutableMessage(message);
359 });
316360
317361 const handleLine = (line: string): Effect.Effect<void, CodexError.CodexAppServerError> => {
318362 if (line.trim().length === 0) {
......@@ -352,6 +396,7 @@ export const makeCodexAppServerPatchedProtocol = Effect.fn("makeCodexAppServerPa
352396 };
353397
354398 yield* options.stdio.stdin.pipe(
399 Stream.interruptWhen(Deferred.await(terminationSignal)),
355400 Stream.decodeText(),
356401 Stream.runForEach((chunk) =>
357402 Effect.sync(() => {