authorgravatar for julius0216@outlook.comJulius Marminge <julius0216@outlook.com> 2026-04-04 23:22:55-07:00
committergravatar for noreply@github.comGitHub <noreply@github.com> 2026-04-04 23:22:55-07:00
log740d7a321602963c4fa9da1b25f05c992333620d
tree82f767fe266c0aa06b4fef9d1255177cbd05eb30
parentcd7980b4574a611f9f721d0463755684653b65f2
signature Signed by PGP key B5690EEEBB952194

Use lazy stream accessors for provider runtime events (#1746)

Co-authored-by: codex <codex@users.noreply.github.com>

14 files changed, 176 insertions(+), 55 deletions(-)

apps/server/integration/OrchestrationEngineHarness.integration.ts+3-3
......@@ -48,7 +48,7 @@ import { CheckpointReactorLive } from "../src/orchestration/Layers/CheckpointRea
4848import { OrchestrationEngineLive } from "../src/orchestration/Layers/OrchestrationEngine.ts";
4949import { OrchestrationProjectionPipelineLive } from "../src/orchestration/Layers/ProjectionPipeline.ts";
5050import { OrchestrationProjectionSnapshotQueryLive } from "../src/orchestration/Layers/ProjectionSnapshotQuery.ts";
51import { RuntimeReceiptBusLive } from "../src/orchestration/Layers/RuntimeReceiptBus.ts";
51import { RuntimeReceiptBusTest } from "../src/orchestration/Layers/RuntimeReceiptBus.ts";
5252import { OrchestrationReactorLive } from "../src/orchestration/Layers/OrchestrationReactor.ts";
5353import { ProviderCommandReactorLive } from "../src/orchestration/Layers/ProviderCommandReactor.ts";
5454import { ProviderRuntimeIngestionLive } from "../src/orchestration/Layers/ProviderRuntimeIngestion.ts";
......@@ -297,7 +297,7 @@ export const makeOrchestrationIntegrationHarness = (
297297 ProjectionPendingApprovalRepositoryLive,
298298 checkpointStoreLayer,
299299 providerLayer,
300 RuntimeReceiptBusLive,
300 RuntimeReceiptBusTest,
301301 );
302302 const serverSettingsLayer = ServerSettingsService.layerTest();
303303 const runtimeIngestionLayer = ProviderRuntimeIngestionLive.pipe(
......@@ -376,7 +376,7 @@ export const makeOrchestrationIntegrationHarness = (
376376 runtime.runPromise(reactor.start().pipe(Scope.provide(scope))),
377377 ).pipe(Effect.orDie);
378378 const receiptHistory = yield* Ref.make<ReadonlyArray<OrchestrationRuntimeReceipt>>([]);
379 yield* Stream.runForEach(runtimeReceiptBus.stream, (receipt) =>
379 yield* Stream.runForEach(runtimeReceiptBus.streamEventsForTest, (receipt) =>
380380 Ref.update(receiptHistory, (history) => [...history, receipt]).pipe(Effect.asVoid),
381381 ).pipe(Effect.forkIn(scope));
382382 yield* Effect.sleep(10);
apps/server/integration/providerService.integration.test.ts+5-5
......@@ -86,6 +86,7 @@ const collectEventsDuring = <A, E, R>(
8686 Effect.forkScoped,
8787 );
8888
89 yield* Effect.sleep("50 millis");
8990 yield* action;
9091
9192 return yield* Effect.forEach(
......@@ -104,7 +105,6 @@ const runTurn = (input: {
104105}) =>
105106 Effect.gen(function* () {
106107 yield* input.harness.queueTurnResponse(input.threadId, input.response);
107
108108 return yield* collectEventsDuring(
109109 input.provider.streamEvents,
110110 input.response.events.length,
......@@ -116,7 +116,7 @@ const runTurn = (input: {
116116 );
117117 });
118118
119it.effect("replays typed runtime fixture events", () =>
119it.live("replays typed runtime fixture events", () =>
120120 Effect.gen(function* () {
121121 const fixture = yield* makeIntegrationFixture;
122122
......@@ -149,7 +149,7 @@ it.effect("replays typed runtime fixture events", () =>
149149 }).pipe(Effect.provide(NodeServices.layer)),
150150);
151151
152it.effect("replays file-changing fixture turn events", () =>
152it.live("replays file-changing fixture turn events", () =>
153153 Effect.gen(function* () {
154154 const fixture = yield* makeIntegrationFixture;
155155 const { join } = yield* Path.Path;
......@@ -188,7 +188,7 @@ it.effect("replays file-changing fixture turn events", () =>
188188 }).pipe(Effect.provide(NodeServices.layer)),
189189);
190190
191it.effect("runs multi-turn tool/approval flow", () =>
191it.live("runs multi-turn tool/approval flow", () =>
192192 Effect.gen(function* () {
193193 const fixture = yield* makeIntegrationFixture;
194194 const { join } = yield* Path.Path;
......@@ -242,7 +242,7 @@ it.effect("runs multi-turn tool/approval flow", () =>
242242 }).pipe(Effect.provide(NodeServices.layer)),
243243);
244244
245it.effect("rolls back provider conversation state only", () =>
245it.live("rolls back provider conversation state only", () =>
246246 Effect.gen(function* () {
247247 const fixture = yield* makeIntegrationFixture;
248248 const { join } = yield* Path.Path;
apps/server/src/orchestration/Layers/CheckpointReactor.test.ts+3-1
......@@ -96,7 +96,9 @@ function createProviderServiceHarness(
9696 listSessions,
9797 getCapabilities: () => Effect.succeed({ sessionModelSwitch: "in-session" }),
9898 rollbackConversation,
99 streamEvents: Stream.fromPubSub(runtimeEventPubSub),
99 get streamEvents() {
100 return Stream.fromPubSub(runtimeEventPubSub);
101 },
100102 };
101103
102104 const emit = (event: LegacyProviderRuntimeEvent): void => {
apps/server/src/orchestration/Layers/ProviderCommandReactor.test.ts+3-1
......@@ -208,7 +208,9 @@ describe("ProviderCommandReactor", () => {
208208 sessionModelSwitch: input?.sessionModelSwitch ?? "in-session",
209209 }),
210210 rollbackConversation: () => unsupported(),
211 streamEvents: Stream.fromPubSub(runtimeEventPubSub),
211 get streamEvents() {
212 return Stream.fromPubSub(runtimeEventPubSub);
213 },
212214 };
213215
214216 const orchestrationLayer = OrchestrationEngineLive.pipe(
apps/server/src/orchestration/Layers/ProviderRuntimeIngestion.test.ts+3-1
......@@ -98,7 +98,9 @@ function createProviderServiceHarness() {
9898 listSessions: () => Effect.succeed([...runtimeSessions]),
9999 getCapabilities: () => Effect.succeed({ sessionModelSwitch: "in-session" }),
100100 rollbackConversation: () => unsupported(),
101 streamEvents: Stream.fromPubSub(runtimeEventPubSub),
101 get streamEvents() {
102 return Stream.fromPubSub(runtimeEventPubSub);
103 },
102104 };
103105
104106 const setSession = (session: ProviderSession): void => {
apps/server/src/orchestration/Layers/RuntimeReceiptBus.ts+20-2
......@@ -1,3 +1,13 @@
1/**
2 * RuntimeReceiptBus layers.
3 *
4 * `RuntimeReceiptBusLive` is the production default and intentionally does not
5 * retain or broadcast receipts. `RuntimeReceiptBusTest` installs the in-memory
6 * PubSub-backed implementation used by integration tests that need to await
7 * checkpoint-reactor milestones precisely.
8 *
9 * @module RuntimeReceiptBus
10 */
111import { Effect, Layer, PubSub, Stream } from "effect";
212
313import {
......@@ -6,13 +16,21 @@ import {
616 type OrchestrationRuntimeReceipt,
717} from "../Services/RuntimeReceiptBus.ts";
818
9const makeRuntimeReceiptBus = Effect.gen(function* () {
19const makeRuntimeReceiptBus = Effect.succeed({
20 publish: () => Effect.void,
21 streamEventsForTest: Stream.empty,
22} satisfies RuntimeReceiptBusShape);
23
24const makeRuntimeReceiptBusTest = Effect.gen(function* () {
1025 const pubSub = yield* PubSub.unbounded<OrchestrationRuntimeReceipt>();
1126
1227 return {
1328 publish: (receipt) => PubSub.publish(pubSub, receipt).pipe(Effect.asVoid),
14 stream: Stream.fromPubSub(pubSub),
29 get streamEventsForTest() {
30 return Stream.fromPubSub(pubSub);
31 },
1532 } satisfies RuntimeReceiptBusShape;
1633});
1734
1835export const RuntimeReceiptBusLive = Layer.effect(RuntimeReceiptBus, makeRuntimeReceiptBus);
36export const RuntimeReceiptBusTest = Layer.effect(RuntimeReceiptBus, makeRuntimeReceiptBusTest);
apps/server/src/orchestration/Services/RuntimeReceiptBus.ts+17-1
......@@ -1,3 +1,19 @@
1/**
2 * RuntimeReceiptBus - Internal checkpoint-reactor synchronization receipts.
3 *
4 * This service exists to expose short-lived orchestration milestones that are
5 * useful in tests and harnesses but are not part of the production runtime
6 * event model. `CheckpointReactor` publishes receipts such as baseline capture,
7 * diff finalization, and turn-processing quiescence so integration tests can
8 * wait for those exact points without inferring them indirectly from persisted
9 * state.
10 *
11 * Production code should only call `publish`. Test code may subscribe via
12 * `streamEventsForTest`, which is intentionally named to make the intended
13 * usage explicit.
14 *
15 * @module RuntimeReceiptBus
16 */
117import { CheckpointRef, IsoDateTime, NonNegativeInt, ThreadId, TurnId } from "@t3tools/contracts";
218import { Schema, ServiceMap } from "effect";
319import type { Effect, Stream } from "effect";
......@@ -40,7 +56,7 @@ export type OrchestrationRuntimeReceipt = typeof OrchestrationRuntimeReceipt.Typ
4056
4157export interface RuntimeReceiptBusShape {
4258 readonly publish: (receipt: OrchestrationRuntimeReceipt) => Effect.Effect<void>;
43 readonly stream: Stream.Stream<OrchestrationRuntimeReceipt>;
59 readonly streamEventsForTest: Stream.Stream<OrchestrationRuntimeReceipt>;
4460}
4561
4662export class RuntimeReceiptBus extends ServiceMap.Service<
apps/server/src/provider/Layers/ClaudeAdapter.ts+3-1
......@@ -3054,7 +3054,9 @@ const makeClaudeAdapter = Effect.fn("makeClaudeAdapter")(function* (
30543054 listSessions,
30553055 hasSession,
30563056 stopAll,
3057 streamEvents: Stream.fromQueue(runtimeEventQueue),
3057 get streamEvents() {
3058 return Stream.fromQueue(runtimeEventQueue);
3059 },
30583060 } satisfies ClaudeAdapterShape;
30593061});
30603062
apps/server/src/provider/Layers/CodexAdapter.ts+3-1
......@@ -1631,7 +1631,9 @@ const makeCodexAdapter = Effect.fn("makeCodexAdapter")(function* (
16311631 listSessions,
16321632 hasSession,
16331633 stopAll,
1634 streamEvents: Stream.fromQueue(runtimeEventQueue),
1634 get streamEvents() {
1635 return Stream.fromQueue(runtimeEventQueue);
1636 },
16351637 } satisfies CodexAdapterShape;
16361638});
16371639
apps/server/src/provider/Layers/EventNdjsonLogger.test.ts+43
......@@ -112,6 +112,49 @@ describe("EventNdjsonLogger", () => {
112112 }),
113113 );
114114
115 it.effect("serializes concurrent first writes for the same segment", () =>
116 Effect.gen(function* () {
117 const tempDir = fs.mkdtempSync(path.join(os.tmpdir(), "t3-provider-log-"));
118 const basePath = path.join(tempDir, "provider-canonical.ndjson");
119
120 try {
121 const logger = yield* makeEventNdjsonLogger(basePath, {
122 stream: "canonical",
123 batchWindowMs: 0,
124 });
125 assert.notEqual(logger, undefined);
126 if (!logger) {
127 return;
128 }
129
130 yield* Effect.all(
131 [
132 logger.write({ id: "evt-concurrent-1" }, null),
133 logger.write({ id: "evt-concurrent-2" }, null),
134 ],
135 { concurrency: "unbounded" },
136 );
137 yield* logger.close();
138
139 const globalPath = path.join(tempDir, "_global.log");
140 assert.equal(fs.existsSync(globalPath), true);
141 const lines = fs
142 .readFileSync(globalPath, "utf8")
143 .trim()
144 .split("\n")
145 .map((line) => parseLogLine(line));
146
147 assert.equal(lines.length, 2);
148 assert.deepEqual(lines.map((line) => line.payload).toSorted(), [
149 '{"id":"evt-concurrent-1"}',
150 '{"id":"evt-concurrent-2"}',
151 ]);
152 } finally {
153 fs.rmSync(tempDir, { recursive: true, force: true });
154 }
155 }),
156 );
157
115158 it.effect("rotates per-thread files when max size is exceeded", () =>
116159 Effect.gen(function* () {
117160 const tempDir = fs.mkdtempSync(path.join(os.tmpdir(), "t3-provider-log-"));
apps/server/src/provider/Layers/EventNdjsonLogger.ts+65-27
......@@ -10,7 +10,7 @@ import path from "node:path";
1010
1111import type { ThreadId } from "@t3tools/contracts";
1212import { RotatingFileSink } from "@t3tools/shared/logging";
13import { Effect, Exit, Logger, Scope } from "effect";
13import { Effect, Exit, Logger, Scope, SynchronizedRef } from "effect";
1414
1515import { toSafeThreadAttachmentSegment } from "../../attachmentStore.ts";
1616
......@@ -40,6 +40,11 @@ interface ThreadWriter {
4040 close: () => Effect.Effect<void>;
4141}
4242
43interface LoggerState {
44 readonly threadWriters: Map<string, ThreadWriter>;
45 readonly failedSegments: Set<string>;
46}
47
4348function logWarning(message: string, context: Record<string, unknown>): Effect.Effect<void> {
4449 return Effect.logWarning(message, context).pipe(Effect.annotateLogs({ scope: LOG_SCOPE }));
4550}
......@@ -193,34 +198,56 @@ export const makeEventNdjsonLogger = Effect.fn("makeEventNdjsonLogger")(function
193198 return undefined;
194199 }
195200
196 const threadWriters = new Map<string, ThreadWriter>();
197 const failedSegments = new Set<string>();
201 const stateRef = yield* SynchronizedRef.make<LoggerState>({
202 threadWriters: new Map(),
203 failedSegments: new Set(),
204 });
198205
199206 const resolveThreadWriter = Effect.fn("resolveThreadWriter")(function* (
200207 threadSegment: string,
201208 ): Effect.fn.Return<ThreadWriter | undefined> {
202 if (failedSegments.has(threadSegment)) {
203 return undefined;
204 }
205 const existing = threadWriters.get(threadSegment);
206 if (existing) {
207 return existing;
208 }
209 return yield* SynchronizedRef.modifyEffect(stateRef, (state) => {
210 if (state.failedSegments.has(threadSegment)) {
211 return Effect.succeed([undefined, state] as const);
212 }
209213
210 const writer = yield* makeThreadWriter({
211 filePath: path.join(path.dirname(filePath), `${threadSegment}.log`),
212 maxBytes,
213 maxFiles,
214 batchWindowMs,
215 streamLabel,
216 });
217 if (!writer) {
218 failedSegments.add(threadSegment);
219 return undefined;
220 }
214 const existing = state.threadWriters.get(threadSegment);
215 if (existing) {
216 return Effect.succeed([existing, state] as const);
217 }
218
219 return makeThreadWriter({
220 filePath: path.join(path.dirname(filePath), `${threadSegment}.log`),
221 maxBytes,
222 maxFiles,
223 batchWindowMs,
224 streamLabel,
225 }).pipe(
226 Effect.map((writer) => {
227 if (!writer) {
228 const nextFailedSegments = new Set(state.failedSegments);
229 nextFailedSegments.add(threadSegment);
230 return [
231 undefined,
232 {
233 ...state,
234 failedSegments: nextFailedSegments,
235 },
236 ] as const;
237 }
221238
222 threadWriters.set(threadSegment, writer);
223 return writer;
239 const nextThreadWriters = new Map(state.threadWriters);
240 nextThreadWriters.set(threadSegment, writer);
241 return [
242 writer,
243 {
244 ...state,
245 threadWriters: nextThreadWriters,
246 },
247 ] as const;
248 }),
249 );
250 });
224251 });
225252
226253 const write = Effect.fn("write")(function* (event: unknown, threadId: ThreadId | null) {
......@@ -239,10 +266,21 @@ export const makeEventNdjsonLogger = Effect.fn("makeEventNdjsonLogger")(function
239266 });
240267
241268 const close = Effect.fn("close")(function* () {
242 for (const writer of threadWriters.values()) {
243 yield* writer.close();
244 }
245 threadWriters.clear();
269 yield* SynchronizedRef.modifyEffect(stateRef, (state) =>
270 Effect.gen(function* () {
271 for (const writer of state.threadWriters.values()) {
272 yield* writer.close();
273 }
274
275 return [
276 undefined,
277 {
278 threadWriters: new Map<string, ThreadWriter>(),
279 failedSegments: new Set<string>(),
280 },
281 ] as const;
282 }),
283 );
246284 });
247285
248286 return {
apps/server/src/provider/Layers/ProviderRegistry.test.ts+3-1
......@@ -115,7 +115,9 @@ function makeMutableServerSettingsService(
115115 yield* PubSub.publish(changes, next);
116116 return next;
117117 }),
118 streamChanges: Stream.fromPubSub(changes),
118 get streamChanges() {
119 return Stream.fromPubSub(changes);
120 },
119121 } satisfies ServerSettingsShape;
120122 });
121123}
apps/server/src/provider/Layers/ProviderService.test.ts+3-1
......@@ -191,7 +191,9 @@ function makeFakeCodexAdapter(provider: ProviderKind = "codex") {
191191 readThread,
192192 rollbackThread,
193193 stopAll,
194 streamEvents: Stream.fromPubSub(runtimeEventPubSub),
194 get streamEvents() {
195 return Stream.fromPubSub(runtimeEventPubSub);
196 },
195197 };
196198
197199 const emit = (event: LegacyProviderRuntimeEvent): void => {
apps/server/src/provider/Layers/ProviderService.ts+2-10
......@@ -22,7 +22,7 @@ import {
2222 type ProviderRuntimeEvent,
2323 type ProviderSession,
2424} from "@t3tools/contracts";
25import { Effect, Layer, Option, PubSub, Queue, Schema, SchemaIssue, Stream } from "effect";
25import { Effect, Layer, Option, PubSub, Schema, SchemaIssue, Stream } from "effect";
2626
2727import {
2828 increment,
......@@ -156,7 +156,6 @@ const makeProviderService = Effect.fn("makeProviderService")(function* (
156156
157157 const registry = yield* ProviderAdapterRegistry;
158158 const directory = yield* ProviderSessionDirectory;
159 const runtimeEventQueue = yield* Queue.unbounded<ProviderRuntimeEvent>();
160159 const runtimeEventPubSub = yield* PubSub.unbounded<ProviderRuntimeEvent>();
161160
162161 const publishRuntimeEvent = (event: ProviderRuntimeEvent): Effect.Effect<void> =>
......@@ -194,15 +193,8 @@ const makeProviderService = Effect.fn("makeProviderService")(function* (
194193 eventType: event.type,
195194 }).pipe(Effect.andThen(publishRuntimeEvent(event)));
196195
197 const worker = Effect.forever(
198 Queue.take(runtimeEventQueue).pipe(Effect.flatMap(processRuntimeEvent)),
199 );
200 yield* Effect.forkScoped(worker);
201
202196 yield* Effect.forEach(adapters, (adapter) =>
203 Stream.runForEach(adapter.streamEvents, (event) =>
204 Queue.offer(runtimeEventQueue, event).pipe(Effect.asVoid),
205 ).pipe(Effect.forkScoped),
197 Stream.runForEach(adapter.streamEvents, processRuntimeEvent).pipe(Effect.forkScoped),
206198 ).pipe(Effect.asVoid);
207199
208200 const recoverSessionForThread = Effect.fn("recoverSessionForThread")(function* (input: {