authorgravatar for julius0216@outlook.comJulius Marminge <julius0216@outlook.com> 2026-02-22 23:40:52-08:00
committergravatar for jmarminge@gmail.comJulius Marminge <jmarminge@gmail.com> 2026-02-26 13:20:36-08:00
logfcc94b3b2ffdf3d96fe09fc224b11f0540d2443c
treee5ba518d82d9d60998772f38cf9578f7df6750bb
parent42b510e8c876d97bf505c3e637e521f7c59ef73c

back in a pretty good state


40 files changed, 2490 insertions(+), 635 deletions(-)

.plans/13-provider-service-integration-tests.md created+110
......@@ -0,0 +1,110 @@
1# ProviderService Integration Test Plan
2
3Goal:
4- Validate end-to-end `ProviderService` behavior with real layers:
5 - `ProviderServiceLive`
6 - `CheckpointServiceLive`
7 - `CheckpointStoreLive`
8 - `CheckpointRepositoryLive` (sqlite in-memory)
9 - `ProviderSessionDirectoryLive`
10- Only fake the adapter event source (deterministic Codex-like stream).
11- Avoid mocking checkpointing/persistence orchestration logic.
12
13## Test Harness
14
15Build a deterministic `TestProviderAdapterLive` in `apps/server/src/provider/Layers/TestProviderAdapter.integration.ts`:
16- Service contract: `ProviderAdapterShape<ProviderAdapterError>`.
17- Internal state:
18 - session registry (session + cwd + threadId)
19 - thread snapshot store (`threadId`, `turns`)
20 - event subscribers
21- Behavior:
22 - `startSession`: creates session with threadId.
23 - `sendTurn`: appends a deterministic turn snapshot and emits ordered events:
24 - `turn/started`
25 - `item/started` / `item/completed` (tool + approval variants depending on scenario)
26 - `item/agentMessage/delta` chunks
27 - `turn/completed`
28 - optional "mutator" callback per turn to change workspace files before completion.
29 - `readThread`, `rollbackThread`, `stopSession`, `stopAll`.
30
31Use real git-backed temporary workspaces in integration tests:
32- initialize repo with baseline commit
33- run provider turn in workspace
34- assert checkpoint diffs against real git refs
35
36## Core Integration Specs
37
381) `startSession` initializes checkpoint root exactly once
39- Arrange:
40 - start provider session in git repo.
41- Assert:
42 - `provider_checkpoints` contains root row (turn 0).
43 - checkpoint ref exists in git.
44 - second `startSession` for new session creates a new independent root.
45
462) Turn without filesystem change
47- Arrange:
48 - emit normal turn events, no file mutation.
49- Assert:
50 - provider subscribers receive:
51 - `turn/started`
52 - `turn/completed`
53 - synthetic `checkpoint/captured`
54 - `listCheckpoints` returns root + turn 1.
55 - `getCheckpointDiff(0 -> 1)` returns empty/no-op diff.
56
573) Turn with filesystem change
58- Arrange:
59 - mutate `README.md` during turn.
60- Assert:
61 - `listCheckpoints` returns root + turn 1.
62 - `getCheckpointDiff(0 -> 1)` contains file path and hunk.
63 - persisted checkpoint metadata includes non-empty `checkpointRef`.
64
654) Multi-turn sequencing and checkpoint monotonicity
66- Arrange:
67 - turn 1: no file change
68 - turn 2: file change
69 - turn 3: file change
70- Assert:
71 - turn counts are monotonic and contiguous in DB (0,1,2,3).
72 - latest checkpoint is marked current.
73 - diffs for adjacent turns map to expected filesystem deltas.
74
755) Revert to checkpoint
76- Arrange:
77 - execute 3 turns with at least one file-changing turn.
78 - call `revertToCheckpoint(turnCount=1)`.
79- Assert:
80 - workspace content matches turn 1 state.
81 - adapter `rollbackThread` called with `numTurns=2`.
82 - DB rows for turns >1 are removed.
83 - later refs are deleted from git.
84
856) Capture failure surface
86- Arrange:
87 - adapter emits `turn/completed`, but file mutation leaves invalid repo state or store capture fails.
88- Assert:
89 - `ProviderService` emits `checkpoint/captureError`.
90 - no partial metadata/ref divergence is left behind.
91
92## WebSocket Coverage (Thin Integration)
93
94Add one ws server integration spec:
95- Subscribe to `providers.event`.
96- Run a deterministic provider turn through ws methods.
97- Assert push stream includes:
98 - `turn/started`, tool events, `turn/completed`, `checkpoint/captured`.
99- Assert orchestration projection still updates assistant message and turn diff summary.
100
101## Proposed PR Split
102
103PR A:
104- Test adapter harness + shared integration fixtures (repo setup, runtime/layer setup).
105
106PR B:
107- Core ProviderService integration specs (cases 1-4).
108
109PR C:
110- Revert + failure-path specs (cases 5-6) + ws thin integration spec.
apps/server/integration/TestProviderAdapter.integration.ts created+310
......@@ -0,0 +1,310 @@
1import { randomUUID } from "node:crypto";
2
3import type {
4 ProviderRuntimeEvent,
5 ProviderSession,
6 ProviderTurnStartResult,
7} from "@t3tools/contracts";
8import { Effect, Queue, Stream } from "effect";
9
10import {
11 ProviderAdapterSessionNotFoundError,
12 ProviderAdapterValidationError,
13 type ProviderAdapterError,
14} from "../src/provider/Errors.ts";
15import type {
16 ProviderAdapterShape,
17 ProviderThreadSnapshot,
18 ProviderThreadTurnSnapshot,
19} from "../src/provider/Services/ProviderAdapter.ts";
20
21export interface TestTurnResponse {
22 readonly events: ReadonlyArray<ProviderRuntimeEvent>;
23 readonly mutateWorkspace?: (input: {
24 readonly cwd: string;
25 readonly turnCount: number;
26 }) => Effect.Effect<void, never>;
27}
28
29interface SessionState {
30 readonly session: ProviderSession;
31 snapshot: ProviderThreadSnapshot;
32 turnCount: number;
33 readonly queuedResponses: Array<TestTurnResponse>;
34 readonly rollbackCalls: Array<number>;
35}
36
37export interface TestProviderAdapterHarness {
38 readonly adapter: ProviderAdapterShape<ProviderAdapterError>;
39 readonly queueTurnResponse: (
40 sessionId: string,
41 response: TestTurnResponse,
42 ) => Effect.Effect<void, ProviderAdapterSessionNotFoundError>;
43 readonly getRollbackCalls: (sessionId: string) => ReadonlyArray<number>;
44}
45
46const PROVIDER = "codex" as const;
47
48function nowIso(): string {
49 return new Date().toISOString();
50}
51
52function sessionNotFound(sessionId: string): ProviderAdapterSessionNotFoundError {
53 return new ProviderAdapterSessionNotFoundError({
54 provider: PROVIDER,
55 sessionId,
56 });
57}
58
59function missingSessionEffect(sessionId: string): Effect.Effect<never, ProviderAdapterError> {
60 return Effect.fail(sessionNotFound(sessionId));
61}
62
63export const makeTestProviderAdapterHarness = Effect.gen(function* () {
64 const runtimeEvents = yield* Queue.unbounded<ProviderRuntimeEvent>();
65 let sessionCount = 0;
66 const sessions = new Map<string, SessionState>();
67
68 const emit = (event: ProviderRuntimeEvent) => Queue.offer(runtimeEvents, event);
69
70 const startSession: ProviderAdapterShape<ProviderAdapterError>["startSession"] = (input) =>
71 Effect.gen(function* () {
72 if (input.provider !== undefined && input.provider !== PROVIDER) {
73 return yield* Effect.fail(
74 new ProviderAdapterValidationError({
75 provider: PROVIDER,
76 operation: "startSession",
77 issue: `Expected provider '${PROVIDER}' but received '${input.provider}'.`,
78 }),
79 );
80 }
81
82 sessionCount += 1;
83 const sessionId = `test-session-${sessionCount}`;
84 const threadId = `test-thread-${sessionCount}`;
85 const createdAt = nowIso();
86
87 const session: ProviderSession = {
88 sessionId,
89 provider: PROVIDER,
90 status: "ready",
91 threadId,
92 cwd: input.cwd,
93 createdAt,
94 updatedAt: createdAt,
95 };
96
97 sessions.set(sessionId, {
98 session,
99 snapshot: {
100 threadId,
101 turns: [],
102 },
103 turnCount: 0,
104 queuedResponses: [],
105 rollbackCalls: [],
106 });
107
108 return session;
109 });
110
111 const sendTurn: ProviderAdapterShape<ProviderAdapterError>["sendTurn"] = (input) =>
112 Effect.gen(function* () {
113 const state = sessions.get(input.sessionId);
114 if (!state) {
115 return yield* missingSessionEffect(input.sessionId);
116 }
117
118 state.turnCount += 1;
119 const turnCount = state.turnCount;
120 const turnId = `turn-${turnCount}`;
121
122 const response = state.queuedResponses.shift();
123 if (!response) {
124 return yield* Effect.fail(
125 new ProviderAdapterValidationError({
126 provider: PROVIDER,
127 operation: "sendTurn",
128 issue: `No queued turn response for session ${input.sessionId}.`,
129 }),
130 );
131 }
132
133 const assistantDeltas: string[] = [];
134 let hasCompletedEvent = false;
135 for (const fixtureEvent of response.events) {
136 const rawEvent: Record<string, unknown> = {
137 ...(fixtureEvent as Record<string, unknown>),
138 eventId: randomUUID(),
139 provider: PROVIDER,
140 sessionId: input.sessionId,
141 createdAt: nowIso(),
142 };
143 if (Object.hasOwn(rawEvent, "threadId")) {
144 rawEvent.threadId = state.snapshot.threadId;
145 }
146 if (Object.hasOwn(rawEvent, "turnId")) {
147 rawEvent.turnId = turnId;
148 }
149
150 const runtimeEvent = rawEvent as ProviderRuntimeEvent;
151 if (runtimeEvent.type === "message.delta") {
152 assistantDeltas.push(runtimeEvent.delta);
153 }
154 if (runtimeEvent.type === "turn.completed") {
155 hasCompletedEvent = true;
156 }
157
158 yield* emit(runtimeEvent);
159 }
160
161 if (response.mutateWorkspace && state.session.cwd) {
162 yield* response.mutateWorkspace({ cwd: state.session.cwd!, turnCount });
163 }
164
165 const userItem = {
166 type: "userMessage",
167 content: [{ type: "text", text: input.input }],
168 } as const;
169 const assistantText = assistantDeltas.join("");
170 const nextItems: Array<unknown> =
171 assistantText.length > 0
172 ? [userItem, { type: "agentMessage", text: assistantText }]
173 : [userItem];
174
175 const nextTurn: ProviderThreadTurnSnapshot = {
176 id: turnId,
177 items: nextItems,
178 };
179
180 state.snapshot = {
181 threadId: state.snapshot.threadId,
182 turns: [...state.snapshot.turns, nextTurn],
183 };
184
185 if (!hasCompletedEvent) {
186 yield* emit({
187 type: "turn.completed",
188 eventId: randomUUID(),
189 provider: PROVIDER,
190 sessionId: input.sessionId,
191 createdAt: nowIso(),
192 threadId: state.snapshot.threadId,
193 turnId,
194 status: "completed",
195 });
196 }
197
198 return {
199 threadId: state.snapshot.threadId,
200 turnId,
201 } satisfies ProviderTurnStartResult;
202 });
203
204 const interruptTurn: ProviderAdapterShape<ProviderAdapterError>["interruptTurn"] = (
205 sessionId,
206 _turnId,
207 ) => (sessions.has(sessionId) ? Effect.void : missingSessionEffect(sessionId));
208
209 const respondToRequest: ProviderAdapterShape<ProviderAdapterError>["respondToRequest"] = (
210 sessionId,
211 _requestId,
212 _decision,
213 ) => (sessions.has(sessionId) ? Effect.void : missingSessionEffect(sessionId));
214
215 const stopSession: ProviderAdapterShape<ProviderAdapterError>["stopSession"] = (sessionId) =>
216 Effect.sync(() => {
217 sessions.delete(sessionId);
218 });
219
220 const listSessions: ProviderAdapterShape<ProviderAdapterError>["listSessions"] = () =>
221 Effect.sync(() => Array.from(sessions.values(), (state) => state.session));
222
223 const hasSession: ProviderAdapterShape<ProviderAdapterError>["hasSession"] = (sessionId) =>
224 Effect.succeed(sessions.has(sessionId));
225
226 const readThread: ProviderAdapterShape<ProviderAdapterError>["readThread"] = (sessionId) => {
227 const state = sessions.get(sessionId);
228 if (!state) {
229 return missingSessionEffect(sessionId);
230 }
231 return Effect.succeed(state.snapshot);
232 };
233
234 const rollbackThread: ProviderAdapterShape<ProviderAdapterError>["rollbackThread"] = (
235 sessionId,
236 numTurns,
237 ) => {
238 const state = sessions.get(sessionId);
239 if (!state) {
240 return missingSessionEffect(sessionId);
241 }
242 if (!Number.isInteger(numTurns) || numTurns < 0 || numTurns > state.snapshot.turns.length) {
243 return Effect.fail(
244 new ProviderAdapterValidationError({
245 provider: PROVIDER,
246 operation: "rollbackThread",
247 issue: "numTurns must be an integer between 0 and current turn count.",
248 }),
249 );
250 }
251
252 return Effect.sync(() => {
253 state.rollbackCalls.push(numTurns);
254 state.snapshot = {
255 threadId: state.snapshot.threadId,
256 turns: state.snapshot.turns.slice(0, state.snapshot.turns.length - numTurns),
257 };
258 state.turnCount = state.snapshot.turns.length;
259 return state.snapshot;
260 });
261 };
262
263 const stopAll: ProviderAdapterShape<ProviderAdapterError>["stopAll"] = () =>
264 Effect.sync(() => {
265 sessions.clear();
266 });
267
268 const adapter: ProviderAdapterShape<ProviderAdapterError> = {
269 provider: PROVIDER,
270 startSession,
271 sendTurn,
272 interruptTurn,
273 respondToRequest,
274 stopSession,
275 listSessions,
276 hasSession,
277 readThread,
278 rollbackThread,
279 stopAll,
280 streamEvents: Stream.fromQueue(runtimeEvents),
281 };
282
283 const queueTurnResponse = (
284 sessionId: string,
285 response: TestTurnResponse,
286 ): Effect.Effect<void, ProviderAdapterSessionNotFoundError> =>
287 Effect.sync(() => sessions.get(sessionId)).pipe(
288 Effect.flatMap((state) =>
289 state
290 ? Effect.sync(() => {
291 state.queuedResponses.push(response);
292 })
293 : Effect.fail(sessionNotFound(sessionId)),
294 ),
295 );
296
297 const getRollbackCalls = (sessionId: string): ReadonlyArray<number> => {
298 const state = sessions.get(sessionId);
299 if (!state) {
300 return [];
301 }
302 return [...state.rollbackCalls];
303 };
304
305 return {
306 adapter,
307 queueTurnResponse,
308 getRollbackCalls,
309 } satisfies TestProviderAdapterHarness;
310});
apps/server/integration/fixtures/providerRuntime.ts created+130
......@@ -0,0 +1,130 @@
1import type { ProviderRuntimeEvent } from "@t3tools/contracts";
2
3const PROVIDER = "codex" as const;
4const SESSION_ID = "fixture-session";
5const THREAD_ID = "fixture-thread";
6const TURN_ID = "fixture-turn";
7
8function baseEvent(eventId: string, createdAt: string): Pick<
9 ProviderRuntimeEvent,
10 "eventId" | "provider" | "sessionId" | "createdAt"
11> {
12 return {
13 eventId,
14 provider: PROVIDER,
15 sessionId: SESSION_ID,
16 createdAt,
17 };
18}
19
20export const codexTurnTextFixture = [
21 {
22 type: "turn.started",
23 ...baseEvent("evt-1", "2026-02-23T00:00:00.000Z"),
24 threadId: THREAD_ID,
25 turnId: TURN_ID,
26 },
27 {
28 type: "message.delta",
29 ...baseEvent("evt-2", "2026-02-23T00:00:00.100Z"),
30 threadId: THREAD_ID,
31 turnId: TURN_ID,
32 delta: "I will make a small update.\n",
33 },
34 {
35 type: "message.delta",
36 ...baseEvent("evt-3", "2026-02-23T00:00:00.200Z"),
37 threadId: THREAD_ID,
38 turnId: TURN_ID,
39 delta: "Done.\n",
40 },
41 {
42 type: "turn.completed",
43 ...baseEvent("evt-4", "2026-02-23T00:00:00.300Z"),
44 threadId: THREAD_ID,
45 turnId: TURN_ID,
46 status: "completed",
47 },
48] satisfies ReadonlyArray<ProviderRuntimeEvent>;
49
50export const codexTurnToolFixture = [
51 {
52 type: "turn.started",
53 ...baseEvent("evt-11", "2026-02-23T00:01:00.000Z"),
54 threadId: THREAD_ID,
55 turnId: TURN_ID,
56 },
57 {
58 type: "tool.started",
59 ...baseEvent("evt-12", "2026-02-23T00:01:00.100Z"),
60 threadId: THREAD_ID,
61 turnId: TURN_ID,
62 toolKind: "command",
63 title: "Command run",
64 detail: "echo integration",
65 },
66 {
67 type: "tool.completed",
68 ...baseEvent("evt-13", "2026-02-23T00:01:00.200Z"),
69 threadId: THREAD_ID,
70 turnId: TURN_ID,
71 toolKind: "command",
72 title: "Command run",
73 detail: "echo integration",
74 },
75 {
76 type: "message.delta",
77 ...baseEvent("evt-14", "2026-02-23T00:01:00.300Z"),
78 threadId: THREAD_ID,
79 turnId: TURN_ID,
80 delta: "Applied the requested edit.\n",
81 },
82 {
83 type: "turn.completed",
84 ...baseEvent("evt-15", "2026-02-23T00:01:00.400Z"),
85 threadId: THREAD_ID,
86 turnId: TURN_ID,
87 status: "completed",
88 },
89] satisfies ReadonlyArray<ProviderRuntimeEvent>;
90
91export const codexTurnApprovalFixture = [
92 {
93 type: "turn.started",
94 ...baseEvent("evt-21", "2026-02-23T00:02:00.000Z"),
95 threadId: THREAD_ID,
96 turnId: TURN_ID,
97 },
98 {
99 type: "approval.requested",
100 ...baseEvent("evt-22", "2026-02-23T00:02:00.100Z"),
101 threadId: THREAD_ID,
102 turnId: TURN_ID,
103 requestId: "req-1",
104 requestKind: "command",
105 detail: "Please approve command",
106 },
107 {
108 type: "approval.resolved",
109 ...baseEvent("evt-23", "2026-02-23T00:02:00.200Z"),
110 threadId: THREAD_ID,
111 turnId: TURN_ID,
112 requestId: "req-1",
113 requestKind: "command",
114 decision: "accept",
115 },
116 {
117 type: "message.delta",
118 ...baseEvent("evt-24", "2026-02-23T00:02:00.300Z"),
119 threadId: THREAD_ID,
120 turnId: TURN_ID,
121 delta: "Approval received and command executed.\n",
122 },
123 {
124 type: "turn.completed",
125 ...baseEvent("evt-25", "2026-02-23T00:02:00.400Z"),
126 threadId: THREAD_ID,
127 turnId: TURN_ID,
128 status: "completed",
129 },
130] satisfies ReadonlyArray<ProviderRuntimeEvent>;
apps/server/integration/providerService.integration.test.ts created+361
......@@ -0,0 +1,361 @@
1import type { ProviderRuntimeEvent } from "@t3tools/contracts";
2import { NodeServices } from "@effect/platform-node";
3import { it, assert } from "@effect/vitest";
4import { Effect, FileSystem, Layer, Path, Queue, Stream } from "effect";
5
6import { runGit as runGitProcess } from "../src/git/Process.ts";
7import { CheckpointServiceLive } from "../src/checkpointing/Layers/CheckpointService.ts";
8import { CheckpointStoreLive } from "../src/checkpointing/Layers/CheckpointStore.ts";
9import { ProviderUnsupportedError } from "../src/provider/Errors.ts";
10import { ProviderAdapterRegistry } from "../src/provider/Services/ProviderAdapterRegistry.ts";
11import { ProviderSessionDirectoryLive } from "../src/provider/Layers/ProviderSessionDirectory.ts";
12import { ProviderServiceLive } from "../src/provider/Layers/ProviderService.ts";
13import {
14 ProviderService,
15 type ProviderServiceShape,
16} from "../src/provider/Services/ProviderService.ts";
17import { SqlitePersistenceMemory } from "../src/persistence/Layers/Sqlite.ts";
18import { CheckpointRepositoryLive } from "../src/persistence/Layers/Checkpoints.ts";
19
20import {
21 makeTestProviderAdapterHarness,
22 type TestProviderAdapterHarness,
23 type TestTurnResponse,
24} from "./TestProviderAdapter.integration.ts";
25import {
26 codexTurnApprovalFixture,
27 codexTurnToolFixture,
28 codexTurnTextFixture,
29} from "./fixtures/providerRuntime.ts";
30
31const runGit = (cwd: string, args: ReadonlyArray<string>, allowNonZeroExit = false) =>
32 runGitProcess({
33 operation: "providerService.integration.git",
34 cwd,
35 args,
36 allowNonZeroExit,
37 }).pipe(Effect.provide(NodeServices.layer));
38
39const gitRefExists = (cwd: string, ref: string) =>
40 runGit(cwd, ["show-ref", "--verify", "--quiet", ref], true).pipe(
41 Effect.map((result) => result.code === 0),
42 );
43
44const checkpointRefForTurn = (threadId: string, turnCount: number) =>
45 `refs/t3/checkpoints/${Buffer.from(threadId, "utf8").toString("base64url")}/turn/${turnCount}`;
46
47const makeGitRepository = Effect.gen(function* () {
48 const fs = yield* FileSystem.FileSystem;
49 const pathService = yield* Path.Path;
50 const cwd = yield* fs.makeTempDirectory();
51
52 yield* runGit(cwd, ["init", "--initial-branch=main"]);
53 yield* runGit(cwd, ["config", "user.email", "test@example.com"]);
54 yield* runGit(cwd, ["config", "user.name", "Test User"]);
55 yield* fs.writeFileString(pathService.join(cwd, "README.md"), "v1\n");
56 yield* runGit(cwd, ["add", "."]);
57 yield* runGit(cwd, ["commit", "-m", "Initial"]);
58
59 return cwd;
60}).pipe(Effect.provide(NodeServices.layer));
61
62interface IntegrationFixture {
63 readonly cwd: string;
64 readonly harness: TestProviderAdapterHarness;
65 readonly layer: Layer.Layer<ProviderService, unknown, never>;
66}
67
68const makeIntegrationFixture = Effect.gen(function* () {
69 const cwd = yield* makeGitRepository;
70 const harness = yield* makeTestProviderAdapterHarness;
71
72 const registry: typeof ProviderAdapterRegistry.Service = {
73 getByProvider: (provider) =>
74 provider === "codex"
75 ? Effect.succeed(harness.adapter)
76 : Effect.fail(new ProviderUnsupportedError({ provider })),
77 listProviders: () => Effect.succeed(["codex"]),
78 };
79
80 const shared = Layer.mergeAll(
81 ProviderSessionDirectoryLive,
82 Layer.succeed(ProviderAdapterRegistry, registry),
83 Layer.provide(CheckpointStoreLive, NodeServices.layer),
84 Layer.provide(CheckpointRepositoryLive, SqlitePersistenceMemory),
85 );
86
87 const checkpointLayer = CheckpointServiceLive.pipe(Layer.provide(shared));
88
89 const layer = ProviderServiceLive.pipe(
90 Layer.provide(Layer.mergeAll(shared, checkpointLayer)),
91 Layer.merge(NodeServices.layer),
92 );
93
94 return {
95 cwd,
96 harness,
97 layer,
98 } satisfies IntegrationFixture;
99});
100
101const collectEventsDuring = <A, E, R>(
102 stream: Stream.Stream<ProviderRuntimeEvent>,
103 count: number,
104 action: Effect.Effect<A, E, R>,
105) =>
106 Effect.gen(function* () {
107 const queue = yield* Queue.unbounded<ProviderRuntimeEvent>();
108 yield* Stream.runForEach(stream, (event) => Queue.offer(queue, event).pipe(Effect.asVoid)).pipe(
109 Effect.forkScoped,
110 );
111
112 yield* action;
113
114 return yield* Effect.forEach(
115 Array.from({ length: count }, () => undefined),
116 () => Queue.take(queue),
117 { discard: false },
118 );
119 });
120
121const runTurn = (input: {
122 readonly provider: ProviderServiceShape;
123 readonly harness: TestProviderAdapterHarness;
124 readonly sessionId: string;
125 readonly userText: string;
126 readonly response: TestTurnResponse;
127}) =>
128 Effect.gen(function* () {
129 yield* input.harness.queueTurnResponse(input.sessionId, input.response);
130
131 return yield* collectEventsDuring(
132 input.provider.streamEvents,
133 input.response.events.length + 1,
134 input.provider.sendTurn({
135 sessionId: input.sessionId,
136 input: input.userText,
137 attachments: [],
138 }),
139 );
140 });
141
142it.effect("replays typed runtime fixture and emits checkpoint.captured", () =>
143 Effect.gen(function* () {
144 const fixture = yield* makeIntegrationFixture;
145
146 yield* Effect.gen(function* () {
147 const provider = yield* ProviderService;
148 const session = yield* provider.startSession({
149 provider: "codex",
150 cwd: fixture.cwd,
151 });
152 const threadId = session.threadId ?? "";
153 assert.equal(threadId.length > 0, true);
154
155 const observedEvents = yield* runTurn({
156 provider,
157 harness: fixture.harness,
158 sessionId: session.sessionId,
159 userText: "hello",
160 response: { events: codexTurnTextFixture },
161 });
162
163 assert.deepEqual(
164 observedEvents.map((event) => event.type),
165 [...codexTurnTextFixture.map((event) => event.type), "checkpoint.captured"],
166 );
167
168 const checkpoints = yield* provider.listCheckpoints({ sessionId: session.sessionId });
169 assert.deepEqual(
170 checkpoints.checkpoints.map((checkpoint) => checkpoint.turnCount),
171 [0, 1],
172 );
173
174 const diff = yield* provider.getCheckpointDiff({
175 sessionId: session.sessionId,
176 fromTurnCount: 0,
177 toTurnCount: 1,
178 });
179 assert.equal(diff.diff.trim(), "");
180
181 const turnOneRefExists = yield* gitRefExists(fixture.cwd, checkpointRefForTurn(threadId, 1));
182 assert.equal(turnOneRefExists, true);
183 }).pipe(Effect.provide(fixture.layer));
184 }),
185);
186
187it.effect("captures file-changing fixture turn and produces a non-empty diff", () =>
188 Effect.gen(function* () {
189 const fixture = yield* makeIntegrationFixture;
190 const { join } = yield* Path.Path;
191 const { writeFileString } = yield* FileSystem.FileSystem;
192
193 yield* Effect.gen(function* () {
194 const provider = yield* ProviderService;
195 const session = yield* provider.startSession({
196 provider: "codex",
197 cwd: fixture.cwd,
198 });
199 const threadId = session.threadId ?? "";
200 assert.equal(threadId.length > 0, true);
201
202 const observedEvents = yield* runTurn({
203 provider,
204 harness: fixture.harness,
205 sessionId: session.sessionId,
206 userText: "make a small change",
207 response: {
208 events: codexTurnToolFixture,
209 mutateWorkspace: ({ cwd }) =>
210 writeFileString(join(cwd, "README.md"), "v2\n").pipe(Effect.asVoid, Effect.ignore),
211 },
212 });
213
214 assert.deepEqual(
215 observedEvents.map((event) => event.type),
216 [...codexTurnToolFixture.map((event) => event.type), "checkpoint.captured"],
217 );
218
219 const diff = yield* provider.getCheckpointDiff({
220 sessionId: session.sessionId,
221 fromTurnCount: 0,
222 toTurnCount: 1,
223 });
224 assert.equal(diff.diff.includes("README.md"), true);
225
226 const turnOneRefExists = yield* gitRefExists(fixture.cwd, checkpointRefForTurn(threadId, 1));
227 assert.equal(turnOneRefExists, true);
228 }).pipe(Effect.provide(fixture.layer));
229 }).pipe(Effect.provide(NodeServices.layer)),
230);
231
232it.effect("runs multi-turn tool/approval flow with contiguous checkpoints", () =>
233 Effect.gen(function* () {
234 const fixture = yield* makeIntegrationFixture;
235 const { join } = yield* Path.Path;
236 const { writeFileString } = yield* FileSystem.FileSystem;
237
238 yield* Effect.gen(function* () {
239 const provider = yield* ProviderService;
240 const session = yield* provider.startSession({
241 provider: "codex",
242 cwd: fixture.cwd,
243 });
244 const threadId = session.threadId ?? "";
245 assert.equal(threadId.length > 0, true);
246
247 const firstTurnEvents = yield* runTurn({
248 provider,
249 harness: fixture.harness,
250 sessionId: session.sessionId,
251 userText: "turn 1",
252 response: {
253 events: codexTurnToolFixture,
254 mutateWorkspace: ({ cwd }) =>
255 writeFileString(join(cwd, "README.md"), "v2\n").pipe(Effect.asVoid, Effect.ignore),
256 },
257 });
258 assert.deepEqual(
259 firstTurnEvents.map((event) => event.type),
260 [...codexTurnToolFixture.map((event) => event.type), "checkpoint.captured"],
261 );
262
263 const secondTurnEvents = yield* runTurn({
264 provider,
265 harness: fixture.harness,
266 sessionId: session.sessionId,
267 userText: "turn 2 approval",
268 response: {
269 events: codexTurnApprovalFixture,
270 mutateWorkspace: ({ cwd }) =>
271 writeFileString(join(cwd, "README.md"), "v3\n").pipe(Effect.asVoid, Effect.ignore),
272 },
273 });
274 assert.deepEqual(
275 secondTurnEvents.map((event) => event.type),
276 [...codexTurnApprovalFixture.map((event) => event.type), "checkpoint.captured"],
277 );
278
279 const turnStateBeforeRevert = yield* provider.listCheckpoints({
280 sessionId: session.sessionId,
281 });
282 assert.deepEqual(
283 turnStateBeforeRevert.checkpoints.map((checkpoint) => checkpoint.turnCount),
284 [0, 1, 2],
285 );
286 const current =
287 turnStateBeforeRevert.checkpoints.find((checkpoint) => checkpoint.isCurrent) ??
288 turnStateBeforeRevert.checkpoints[turnStateBeforeRevert.checkpoints.length - 1];
289 assert.equal(current?.turnCount, 2);
290 }).pipe(Effect.provide(fixture.layer));
291 }).pipe(Effect.provide(NodeServices.layer)),
292);
293
294it.effect("reverts to turn 1 and prunes later checkpoint refs", () =>
295 Effect.gen(function* () {
296 const fixture = yield* makeIntegrationFixture;
297 const { join } = yield* Path.Path;
298 const { writeFileString, readFileString } = yield* FileSystem.FileSystem;
299
300 yield* Effect.gen(function* () {
301 const provider = yield* ProviderService;
302 const session = yield* provider.startSession({
303 provider: "codex",
304 cwd: fixture.cwd,
305 });
306 const threadId = session.threadId ?? "";
307 assert.equal(threadId.length > 0, true);
308
309 yield* runTurn({
310 provider,
311 harness: fixture.harness,
312 sessionId: session.sessionId,
313 userText: "turn 1",
314 response: {
315 events: codexTurnToolFixture,
316 mutateWorkspace: ({ cwd }) =>
317 writeFileString(join(cwd, "README.md"), "v2\n").pipe(Effect.asVoid, Effect.ignore),
318 },
319 });
320
321 yield* runTurn({
322 provider,
323 harness: fixture.harness,
324 sessionId: session.sessionId,
325 userText: "turn 2 approval",
326 response: {
327 events: codexTurnApprovalFixture,
328 mutateWorkspace: ({ cwd }) =>
329 writeFileString(join(cwd, "README.md"), "v3\n").pipe(Effect.asVoid, Effect.ignore),
330 },
331 });
332
333 const turnTwoRef = checkpointRefForTurn(threadId, 2);
334 const turnTwoRefBefore = yield* gitRefExists(fixture.cwd, turnTwoRef);
335 assert.equal(turnTwoRefBefore, true);
336
337 const revert = yield* provider.revertToCheckpoint({
338 sessionId: session.sessionId,
339 turnCount: 1,
340 });
341 assert.equal(revert.rolledBackTurns, 1);
342
343 const rollbackCalls = fixture.harness.getRollbackCalls(session.sessionId);
344 assert.deepEqual(rollbackCalls, [1]);
345
346 const readme = yield* readFileString(join(fixture.cwd, "README.md"));
347 assert.equal(readme, "v2\n");
348
349 const afterRevertCheckpoints = yield* provider.listCheckpoints({
350 sessionId: session.sessionId,
351 });
352 assert.deepEqual(
353 afterRevertCheckpoints.checkpoints.map((checkpoint) => checkpoint.turnCount),
354 [0, 1],
355 );
356
357 const turnTwoRefAfter = yield* gitRefExists(fixture.cwd, turnTwoRef);
358 assert.equal(turnTwoRefAfter, false);
359 }).pipe(Effect.provide(fixture.layer));
360 }).pipe(Effect.provide(NodeServices.layer)),
361);
apps/server/package.json+1-1
......@@ -11,7 +11,7 @@
1111 "main": "./dist/index.mjs",
1212 "scripts": {
1313 "dev": "VITE_DEV_SERVER_URL=${VITE_DEV_SERVER_URL:-http://localhost:5173} bun run src/index.ts",
14 "build": "tsdown && node scripts/bundle-client.mjs",
14 "build": "tsdown && node scripts/bundle-client.ts",
1515 "start": "node dist/index.mjs",
1616 "typecheck": "tsc --noEmit",
1717 "test": "vitest run"
apps/server/scripts/bundle-client.mjs deleted-21
......@@ -1,21 +0,0 @@
1/**
2 * Copies the built web app into dist/client/ so the published npm package
3 * includes the web UI. This runs as a post-build step.
4 */
5import fs from "node:fs";
6import path from "node:path";
7import { fileURLToPath } from "node:url";
8
9const __dirname = path.dirname(fileURLToPath(import.meta.url));
10const webDist = path.resolve(__dirname, "../../web/dist");
11const target = path.resolve(__dirname, "../dist/client");
12
13if (!fs.existsSync(webDist)) {
14 console.log(
15 "âš  Web dist not found — skipping client bundle. Run `bun run --cwd apps/web build` first.",
16 );
17 process.exit(0);
18}
19
20fs.cpSync(webDist, target, { recursive: true });
21console.log("✓ Bundled web app into dist/client");
apps/server/scripts/bundle-client.ts created+19
......@@ -0,0 +1,19 @@
1/**
2 * Copies the built web app into dist/client/ so the published npm package
3 * includes the web UI. This runs as a post-build step.
4 */
5import * as fs from "node:fs";
6import * as path from "node:path";
7
8const webDist = path.resolve(import.meta.dirname, "../../web/dist");
9const target = path.resolve(import.meta.dirname, "../dist/client");
10
11if (!fs.existsSync(webDist)) {
12 console.log(
13 "âš  Web dist not found — skipping client bundle. Run `bun run --cwd apps/web build` first.",
14 );
15 process.exit(0);
16}
17
18fs.cpSync(webDist, target, { recursive: true });
19console.log("✓ Bundled web app into dist/client");
apps/server/src/checkpointing/Layers/CheckpointService.test.ts+2-5
......@@ -1,6 +1,5 @@
11import type {
22 ProviderApprovalDecision,
3 ProviderEvent,
43 ProviderKind,
54 ProviderSendTurnInput,
65 ProviderSession,
......@@ -9,7 +8,7 @@ import type {
98} from "@t3tools/contracts";
109import { NodeServices } from "@effect/platform-node";
1110import { it, assert } from "@effect/vitest";
12import { Effect, FileSystem, Layer, Path } from "effect";
11import { Effect, FileSystem, Layer, Path, Stream } from "effect";
1312
1413import { runGit as runGitProcess } from "../../git/Process.ts";
1514import { CheckpointServiceLive } from "./CheckpointService.ts";
......@@ -194,9 +193,7 @@ class InMemoryProviderAdapter implements ProviderAdapterShape<ProviderAdapterErr
194193 this.active = false;
195194 });
196195
197 readonly subscribeToEvents = (
198 _callback: (event: ProviderEvent) => void,
199 ): Effect.Effect<() => void, ProviderAdapterError> => Effect.succeed(() => undefined);
196 readonly streamEvents = Stream.empty;
200197}
201198
202199function makeFixture(turns: ReadonlyArray<ProviderThreadTurnSnapshot>) {
apps/server/src/checkpointing/Layers/CheckpointService.ts+22
......@@ -377,6 +377,16 @@ const makeCheckpointService = Effect.gen(function* () {
377377 return next;
378378 });
379379
380 const clearSessionSemaphore = (sessionId: string) =>
381 Ref.update(sessionSemaphoresRef, (current) => {
382 if (!current.has(sessionId)) {
383 return current;
384 }
385 const next = new Map(current);
386 next.delete(sessionId);
387 return next;
388 });
389
380390 const getTrackedCheckpointCwd = (sessionId: string) =>
381391 Ref.get(sessionCheckpointCwdsRef).pipe(
382392 Effect.map((current) => Option.fromNullishOr(current.get(sessionId))),
......@@ -1024,12 +1034,24 @@ const makeCheckpointService = Effect.gen(function* () {
10241034 ),
10251035 );
10261036
1037 const releaseSession: CheckpointServiceShape["releaseSession"] = (input) =>
1038 Effect.gen(function* () {
1039 const providerSessionId = yield* normalizeNonEmpty(
1040 input.providerSessionId,
1041 "CheckpointService.releaseSession",
1042 "providerSessionId",
1043 );
1044 yield* clearTrackedCheckpointCwd(providerSessionId);
1045 yield* clearSessionSemaphore(providerSessionId);
1046 });
1047
10271048 return {
10281049 initializeForSession,
10291050 captureCurrentTurn,
10301051 listCheckpoints,
10311052 getCheckpointDiff,
10321053 revertToCheckpoint,
1054 releaseSession,
10331055 } satisfies CheckpointServiceShape;
10341056});
10351057
apps/server/src/checkpointing/Services/CheckpointService.ts+10
......@@ -69,6 +69,16 @@ export interface CheckpointServiceShape {
6969 readonly revertToCheckpoint: (
7070 input: ProviderRevertToCheckpointInput,
7171 ) => Effect.Effect<ProviderRevertToCheckpointResult, CheckpointServiceError>;
72
73 /**
74 * Release in-memory checkpoint session state after session shutdown.
75 *
76 * This only clears local coordination state (locks/cwd tracking). It does not
77 * delete durable checkpoint metadata or git refs.
78 */
79 readonly releaseSession: (input: {
80 readonly providerSessionId: string;
81 }) => Effect.Effect<void, CheckpointServiceError>;
7282}
7383
7484/**
apps/server/src/orchestration/Layer.ts+13
......@@ -143,6 +143,19 @@ function mapCommandToEvent(command: OrchestrationCommand): Omit<OrchestrationEve
143143 : {}),
144144 },
145145 };
146 case "thread.activity.append":
147 return {
148 eventId,
149 type: "thread.activity-appended",
150 aggregateType: "thread",
151 aggregateId: command.threadId,
152 occurredAt: command.createdAt,
153 commandId: command.commandId,
154 payload: {
155 threadId: command.threadId,
156 activity: command.activity,
157 },
158 };
146159 case "thread.revert":
147160 return {
148161 eventId,
apps/server/src/orchestration/Service.ts+6-6
......@@ -27,7 +27,7 @@ export interface OrchestrationEngineShape {
2727 *
2828 * @returns Effect containing the latest read model.
2929 */
30 readonly getSnapshot: () => Effect.Effect<OrchestrationReadModel>;
30 readonly getSnapshot: () => Effect.Effect<OrchestrationReadModel, never, never>;
3131
3232 /**
3333 * Replay persisted orchestration events from an exclusive sequence cursor.
......@@ -37,7 +37,7 @@ export interface OrchestrationEngineShape {
3737 */
3838 readonly replayEvents: (
3939 fromSequenceExclusive: number,
40 ) => Effect.Effect<OrchestrationEvent[], OrchestrationEventRepositoryError>;
40 ) => Effect.Effect<OrchestrationEvent[], OrchestrationEventRepositoryError, never>;
4141
4242 /**
4343 * Validate and dispatch an unknown command payload.
......@@ -49,7 +49,7 @@ export interface OrchestrationEngineShape {
4949 */
5050 readonly dispatchUnknown: (
5151 command: unknown,
52 ) => Effect.Effect<{ sequence: number }, OrchestrationEngineError>;
52 ) => Effect.Effect<{ sequence: number }, OrchestrationEngineError, never>;
5353
5454 /**
5555 * Dispatch a validated orchestration command.
......@@ -59,7 +59,7 @@ export interface OrchestrationEngineShape {
5959 */
6060 readonly dispatch: (
6161 command: OrchestrationCommand,
62 ) => Effect.Effect<{ sequence: number }, OrchestrationDispatchError>;
62 ) => Effect.Effect<{ sequence: number }, OrchestrationDispatchError, never>;
6363
6464 /**
6565 * Subscribe to read model updates.
......@@ -69,7 +69,7 @@ export interface OrchestrationEngineShape {
6969 */
7070 readonly subscribeToReadModel: (
7171 callback: (snapshot: OrchestrationReadModel) => void,
72 ) => Effect.Effect<() => void>;
72 ) => Effect.Effect<() => void, never, never>;
7373
7474 /**
7575 * Subscribe to domain event fan-out.
......@@ -79,7 +79,7 @@ export interface OrchestrationEngineShape {
7979 */
8080 readonly subscribeToDomainEvents: (
8181 callback: (event: OrchestrationEvent) => void,
82 ) => Effect.Effect<() => void>;
82 ) => Effect.Effect<() => void, never, never>;
8383}
8484
8585/**
apps/server/src/orchestration/reducer.test.ts+1
......@@ -46,6 +46,7 @@ describe("orchestration reducer", () => {
4646 latestTurnDurationMs: null,
4747 messages: [],
4848 session: null,
49 activities: [],
4950 turnDiffSummaries: [],
5051 error: null,
5152 },
apps/server/src/orchestration/reducer.ts+50-2
......@@ -7,6 +7,7 @@ import type {
77 OrchestrationThread,
88} from "@t3tools/contracts";
99import {
10 OrchestrationThreadActivitySchema,
1011 OrchestrationGitReadModelSchema,
1112 OrchestrationMessageSchema,
1213 OrchestrationSessionSchema,
......@@ -89,6 +90,11 @@ const ThreadRevertedPayloadSchema = Schema.Struct({
8990 messageCount: Schema.Number,
9091});
9192
93const ThreadActivityAppendedPayloadSchema = Schema.Struct({
94 threadId: Schema.String,
95 activity: OrchestrationThreadActivitySchema,
96});
97
9298export function createEmptyReadModel(nowIso: string): OrchestrationReadModel {
9399 return {
94100 sequence: 0,
......@@ -135,6 +141,7 @@ export function reduceEvent(
135141 messages: [],
136142 session: null,
137143 turnDiffSummaries: [],
144 activities: [],
138145 error: null,
139146 },
140147 event.type,
......@@ -214,11 +221,13 @@ export function reduceEvent(
214221 ? message.text
215222 : entry.text,
216223 streaming: message.streaming,
217 createdAt: message.createdAt,
224 createdAt: entry.createdAt,
218225 }
219226 : entry,
220227 )
221 : [...targetThread.messages, message];
228 : !message.streaming && message.text.length === 0
229 ? targetThread.messages
230 : [...targetThread.messages, message];
222231
223232 return {
224233 ...nextBase,
......@@ -334,6 +343,16 @@ export function reduceEvent(
334343 .toSorted((left, right) => left.completedAt.localeCompare(right.completedAt));
335344 const latestSummary =
336345 turnDiffSummaries.length > 0 ? turnDiffSummaries[turnDiffSummaries.length - 1] : null;
346 const activities = thread.activities.filter((activity) => {
347 if (!activity.turnId) {
348 return true;
349 }
350 const checkpointTurnCount = inferredTurnCountByTurnId.get(activity.turnId);
351 if (checkpointTurnCount === undefined) {
352 return true;
353 }
354 return checkpointTurnCount <= targetTurnCount;
355 });
337356
338357 return {
339358 ...nextBase,
......@@ -347,6 +366,35 @@ export function reduceEvent(
347366 latestTurnStartedAt: null,
348367 latestTurnCompletedAt: latestSummary?.completedAt ?? null,
349368 latestTurnDurationMs: null,
369 activities,
370 updatedAt: event.occurredAt,
371 }),
372 };
373 }),
374 );
375 case "thread.activity-appended":
376 return decodeForEvent(
377 ThreadActivityAppendedPayloadSchema,
378 event.payload,
379 event.type,
380 "payload",
381 ).pipe(
382 Effect.map((payload) => {
383 const thread = nextBase.threads.find((entry) => entry.id === payload.threadId);
384 if (!thread) {
385 return nextBase;
386 }
387 const activities = [
388 ...thread.activities.filter((activity) => activity.id !== payload.activity.id),
389 payload.activity,
390 ]
391 .toSorted((left, right) => left.createdAt.localeCompare(right.createdAt))
392 .slice(-500);
393
394 return {
395 ...nextBase,
396 threads: updateThread(nextBase.threads, payload.threadId, {
397 activities,
350398 updatedAt: event.occurredAt,
351399 }),
352400 };
apps/server/src/provider/Layers/CodexAdapter.test.ts+12-6
......@@ -9,7 +9,7 @@ import type {
99import { afterAll, assert, it, vi } from "@effect/vitest";
1010import { assertFailure } from "@effect/vitest/utils";
1111
12import { Effect } from "effect";
12import { Effect, Fiber, Stream } from "effect";
1313
1414import { CodexAppServerManager } from "../../codexAppServerManager.ts";
1515import { ProviderAdapterValidationError } from "../Errors.ts";
......@@ -170,11 +170,15 @@ const lifecycleManager = new FakeCodexManager();
170170const lifecycleLayer = it.layer(makeCodexAdapterLive({ manager: lifecycleManager }));
171171
172172lifecycleLayer("CodexAdapterLive lifecycle", (it) => {
173 it.effect("subscribes and unsubscribes from manager events", () =>
173 it.effect("emits canonical events on stream and supports consumer interruption", () =>
174174 Effect.gen(function* () {
175175 const adapter = yield* CodexAdapter;
176 const onEvent = vi.fn((_event: ProviderEvent) => undefined);
177 const unsubscribe = yield* adapter.subscribeToEvents(onEvent);
176 const onEvent = vi.fn(() => undefined);
177 const consumer = yield* Stream.runForEach(adapter.streamEvents, () =>
178 Effect.sync(() => {
179 onEvent();
180 }),
181 ).pipe(Effect.forkChild);
178182
179183 const event: ProviderEvent = {
180184 id: "evt-1",
......@@ -183,14 +187,16 @@ lifecycleLayer("CodexAdapterLive lifecycle", (it) => {
183187 sessionId: "sess-1",
184188 createdAt: new Date().toISOString(),
185189 method: "turn/started",
190 turnId: "turn-1",
186191 };
187192
188193 lifecycleManager.emit("event", event);
194 yield* Effect.promise(() => new Promise<void>((resolve) => setTimeout(resolve, 0)));
189195 assert.equal(onEvent.mock.calls.length, 1);
190 assert.equal(onEvent.mock.calls[0]?.[0]?.method, "turn/started");
191196
192 unsubscribe();
197 yield* Fiber.interrupt(consumer);
193198 lifecycleManager.emit("event", event);
199 yield* Effect.promise(() => new Promise<void>((resolve) => setTimeout(resolve, 0)));
194200 assert.equal(onEvent.mock.calls.length, 1);
195201 assert.equal(lifecycleManager.stopAllImpl.mock.calls.length, 0);
196202 }),
apps/server/src/provider/Layers/CodexAdapter.ts+329-13
......@@ -6,8 +6,13 @@
66 *
77 * @module CodexAdapterLive
88 */
9import type { ProviderEvent } from "@t3tools/contracts";
10import { Effect, Layer } from "effect";
9import type {
10 ProviderEvent,
11 ProviderRuntimeEvent,
12 ProviderRuntimeTurnStatus,
13 ProviderRuntimeToolKind,
14} from "@t3tools/contracts";
15import { Effect, Layer, Queue, Stream } from "effect";
1116
1217import {
1318 ProviderAdapterProcessError,
......@@ -19,12 +24,14 @@ import {
1924} from "../Errors.ts";
2025import { CodexAdapter, type CodexAdapterShape } from "../Services/CodexAdapter.ts";
2126import { CodexAppServerManager } from "../../codexAppServerManager.ts";
27import { makeEventNdjsonLogger } from "./EventNdjsonLogger.ts";
2228
2329const PROVIDER = "codex" as const;
2430
2531export interface CodexAdapterLiveOptions {
2632 readonly manager?: CodexAppServerManager;
2733 readonly makeManager?: () => CodexAppServerManager;
34 readonly nativeEventLogPath?: string;
2835}
2936
3037function toMessage(cause: unknown, fallback: string): string {
......@@ -69,8 +76,309 @@ function toRequestError(sessionId: string, method: string, cause: unknown): Prov
6976 });
7077}
7178
79function asObject(value: unknown): Record<string, unknown> | undefined {
80 if (!value || typeof value !== "object") {
81 return undefined;
82 }
83 return value as Record<string, unknown>;
84}
85
86function asString(value: unknown): string | undefined {
87 return typeof value === "string" ? value : undefined;
88}
89
90function toTurnStatus(value: unknown): ProviderRuntimeTurnStatus | undefined {
91 switch (value) {
92 case "completed":
93 case "failed":
94 case "cancelled":
95 case "interrupted":
96 return value;
97 default:
98 return undefined;
99 }
100}
101
102function normalizeItemType(raw: unknown): string {
103 const type = asString(raw);
104 if (!type) return "item";
105 return type
106 .replace(/([a-z0-9])([A-Z])/g, "$1 $2")
107 .replace(/[._/-]/g, " ")
108 .replace(/\s+/g, " ")
109 .trim()
110 .toLowerCase();
111}
112
113function shouldDropItemType(type: string): boolean {
114 if (type.includes("preamble") || type.includes("reasoning") || type.includes("thought")) {
115 return true;
116 }
117 return type === "work" || type.startsWith("work ");
118}
119
120function toolMeta(type: string):
121 | {
122 readonly toolKind: ProviderRuntimeToolKind;
123 readonly title: string;
124 }
125 | undefined {
126 if (type.includes("command")) {
127 return { toolKind: "command", title: "Command run" };
128 }
129 if (type.includes("file change")) {
130 return { toolKind: "file-change", title: "File change" };
131 }
132 if (type.includes("tool")) {
133 return { toolKind: "other", title: "Tool call" };
134 }
135 return undefined;
136}
137
138function itemDetail(
139 item: Record<string, unknown>,
140 payload: Record<string, unknown>,
141): string | undefined {
142 const nestedResult = asObject(item.result);
143 const candidates = [
144 asString(item.command),
145 asString(item.title),
146 asString(item.summary),
147 asString(item.text),
148 asString(item.path),
149 asString(item.prompt),
150 asString(nestedResult?.command),
151 asString(payload.command),
152 asString(payload.message),
153 asString(payload.prompt),
154 ];
155 for (const candidate of candidates) {
156 if (!candidate) continue;
157 const trimmed = candidate.trim();
158 if (trimmed.length === 0) continue;
159 return trimmed;
160 }
161 return undefined;
162}
163
164function mapToolEvent(event: ProviderEvent): ProviderRuntimeEvent | undefined {
165 if (event.method !== "item/started" && event.method !== "item/completed") {
166 return undefined;
167 }
168 const payload = asObject(event.payload);
169 const item = asObject(payload?.item);
170 if (!item) {
171 return undefined;
172 }
173 const normalizedType = normalizeItemType(item.type ?? item.kind);
174 if (shouldDropItemType(normalizedType)) {
175 return undefined;
176 }
177 const meta = toolMeta(normalizedType);
178 if (!meta) {
179 return undefined;
180 }
181
182 const eventBase = {
183 eventId: event.id,
184 provider: event.provider,
185 sessionId: event.sessionId,
186 createdAt: event.createdAt,
187 ...(event.threadId ? { threadId: event.threadId } : {}),
188 ...(event.turnId ? { turnId: event.turnId } : {}),
189 ...(event.itemId ? { itemId: event.itemId } : {}),
190 toolKind: meta.toolKind,
191 title: meta.title,
192 ...(payload ? { detail: itemDetail(item, payload) } : {}),
193 } as const;
194
195 if (event.method === "item/started") {
196 return {
197 type: "tool.started",
198 ...eventBase,
199 };
200 }
201
202 return {
203 type: "tool.completed",
204 ...eventBase,
205 };
206}
207
208function mapToRuntimeEvents(event: ProviderEvent): ReadonlyArray<ProviderRuntimeEvent> {
209 const payload = asObject(event.payload);
210 const turn = asObject(payload?.turn);
211
212 if (event.kind === "error") {
213 if (!event.message) {
214 return [];
215 }
216 return [
217 {
218 type: "runtime.error",
219 eventId: event.id,
220 provider: event.provider,
221 sessionId: event.sessionId,
222 createdAt: event.createdAt,
223 ...(event.threadId ? { threadId: event.threadId } : {}),
224 ...(event.turnId ? { turnId: event.turnId } : {}),
225 ...(event.itemId ? { itemId: event.itemId } : {}),
226 message: event.message,
227 },
228 ];
229 }
230
231 if (event.kind === "request" && event.requestId && event.requestKind) {
232 const detail =
233 asString(payload?.command) ?? asString(payload?.reason) ?? asString(payload?.prompt);
234 return [
235 {
236 type: "approval.requested",
237 eventId: event.id,
238 provider: event.provider,
239 sessionId: event.sessionId,
240 createdAt: event.createdAt,
241 ...(event.threadId ? { threadId: event.threadId } : {}),
242 ...(event.turnId ? { turnId: event.turnId } : {}),
243 ...(event.itemId ? { itemId: event.itemId } : {}),
244 requestId: event.requestId,
245 requestKind: event.requestKind,
246 ...(detail ? { detail } : {}),
247 },
248 ];
249 }
250
251 if (event.method === "item/requestApproval/decision" && event.requestId) {
252 const decision = asString(payload?.decision);
253 return [
254 {
255 type: "approval.resolved",
256 eventId: event.id,
257 provider: event.provider,
258 sessionId: event.sessionId,
259 createdAt: event.createdAt,
260 ...(event.threadId ? { threadId: event.threadId } : {}),
261 ...(event.turnId ? { turnId: event.turnId } : {}),
262 ...(event.itemId ? { itemId: event.itemId } : {}),
263 requestId: event.requestId,
264 ...(event.requestKind ? { requestKind: event.requestKind } : {}),
265 ...(decision ? { decision } : {}),
266 },
267 ];
268 }
269
270 const tool = mapToolEvent(event);
271 if (tool) {
272 return [tool];
273 }
274
275 if (event.method === "session/started") {
276 return [
277 {
278 type: "session.started",
279 eventId: event.id,
280 provider: event.provider,
281 sessionId: event.sessionId,
282 createdAt: event.createdAt,
283 ...(event.threadId ? { threadId: event.threadId } : {}),
284 ...(event.message ? { message: event.message } : {}),
285 },
286 ];
287 }
288
289 if (event.method === "session/exited") {
290 return [
291 {
292 type: "session.exited",
293 eventId: event.id,
294 provider: event.provider,
295 sessionId: event.sessionId,
296 createdAt: event.createdAt,
297 ...(event.threadId ? { threadId: event.threadId } : {}),
298 ...(event.message ? { message: event.message } : {}),
299 },
300 ];
301 }
302
303 if (event.method === "thread/started") {
304 const threadId = event.threadId ?? asString(asObject(payload?.thread)?.id);
305 if (!threadId) {
306 return [];
307 }
308 return [
309 {
310 type: "thread.started",
311 eventId: event.id,
312 provider: event.provider,
313 sessionId: event.sessionId,
314 createdAt: event.createdAt,
315 threadId,
316 },
317 ];
318 }
319
320 if (event.method === "turn/started") {
321 const turnId = event.turnId ?? asString(turn?.id);
322 if (!turnId) {
323 return [];
324 }
325 return [
326 {
327 type: "turn.started",
328 eventId: event.id,
329 provider: event.provider,
330 sessionId: event.sessionId,
331 createdAt: event.createdAt,
332 ...(event.threadId ? { threadId: event.threadId } : {}),
333 turnId,
334 },
335 ];
336 }
337
338 if (event.method === "turn/completed") {
339 return [
340 {
341 type: "turn.completed",
342 eventId: event.id,
343 provider: event.provider,
344 sessionId: event.sessionId,
345 createdAt: event.createdAt,
346 ...(event.threadId ? { threadId: event.threadId } : {}),
347 ...(event.turnId ? { turnId: event.turnId } : {}),
348 ...(toTurnStatus(turn?.status) ? { status: toTurnStatus(turn?.status) } : {}),
349 ...(asString(asObject(turn?.error)?.message)
350 ? { errorMessage: asString(asObject(turn?.error)?.message) }
351 : {}),
352 },
353 ];
354 }
355
356 if (event.method === "item/agentMessage/delta" && event.textDelta && event.textDelta.length > 0) {
357 return [
358 {
359 type: "message.delta",
360 eventId: event.id,
361 provider: event.provider,
362 sessionId: event.sessionId,
363 createdAt: event.createdAt,
364 ...(event.threadId ? { threadId: event.threadId } : {}),
365 ...(event.turnId ? { turnId: event.turnId } : {}),
366 ...(event.itemId ? { itemId: event.itemId } : {}),
367 delta: event.textDelta,
368 },
369 ];
370 }
371
372 return [];
373}
374
72375const makeCodexAdapter = (options?: CodexAdapterLiveOptions) =>
73376 Effect.gen(function* () {
377 const nativeEventLogger =
378 options?.nativeEventLogPath !== undefined
379 ? makeEventNdjsonLogger(options.nativeEventLogPath)
380 : undefined;
381
74382 const manager = yield* Effect.acquireRelease(
75383 Effect.sync(() => {
76384 if (options?.manager) {
......@@ -175,20 +483,28 @@ const makeCodexAdapter = (options?: CodexAdapterLiveOptions) =>
175483 manager.stopAll();
176484 });
177485
178 const subscribeToEvents: CodexAdapterShape["subscribeToEvents"] = (callback) =>
486 const runtimeEventQueue = yield* Queue.unbounded<ProviderRuntimeEvent>();
487
488 yield* Effect.acquireRelease(
179489 Effect.sync(() => {
180490 const listener = (event: ProviderEvent) => {
181 try {
182 callback(event);
183 } catch {
184 // Adapter subscribers must not destabilize provider event processing.
185 }
491 nativeEventLogger?.write({
492 observedAt: new Date().toISOString(),
493 event,
494 });
495 Queue.offerAllUnsafe(runtimeEventQueue, mapToRuntimeEvents(event));
186496 };
187497 manager.on("event", listener);
188 return () => {
189 manager.off("event", listener);
190 };
191 });
498 return listener;
499 }),
500 (listener) =>
501 Effect.gen(function* () {
502 yield* Effect.sync(() => {
503 manager.off("event", listener);
504 });
505 yield* Queue.shutdown(runtimeEventQueue);
506 }),
507 );
192508
193509 return {
194510 provider: PROVIDER,
......@@ -202,7 +518,7 @@ const makeCodexAdapter = (options?: CodexAdapterLiveOptions) =>
202518 listSessions,
203519 hasSession,
204520 stopAll,
205 subscribeToEvents,
521 streamEvents: Stream.fromQueue(runtimeEventQueue),
206522 } satisfies CodexAdapterShape;
207523 });
208524
apps/server/src/provider/Layers/EventNdjsonLogger.test.ts created+37
......@@ -0,0 +1,37 @@
1import fs from "node:fs";
2import os from "node:os";
3import path from "node:path";
4
5import { describe, expect, it } from "vitest";
6
7import { makeEventNdjsonLogger } from "./EventNdjsonLogger.ts";
8
9describe("EventNdjsonLogger", () => {
10 it("writes NDJSON records to disk", () => {
11 const tempDir = fs.mkdtempSync(path.join(os.tmpdir(), "t3-provider-log-"));
12 const filePath = path.join(tempDir, "provider-native.ndjson");
13
14 try {
15 const logger = makeEventNdjsonLogger(filePath);
16 expect(logger).toBeDefined();
17 if (!logger) {
18 return;
19 }
20
21 logger.write({ type: "native", payload: { id: "evt-1" } });
22 logger.write({ type: "canonical", payload: { id: "evt-2" } });
23 logger.close();
24
25 const raw = fs.readFileSync(filePath, "utf8").trim();
26 const lines = raw.split("\n");
27 expect(lines.length).toBe(2);
28
29 const first = JSON.parse(lines[0] ?? "{}");
30 const second = JSON.parse(lines[1] ?? "{}");
31 expect(first).toEqual({ type: "native", payload: { id: "evt-1" } });
32 expect(second).toEqual({ type: "canonical", payload: { id: "evt-2" } });
33 } finally {
34 fs.rmSync(tempDir, { recursive: true, force: true });
35 }
36 });
37});
apps/server/src/provider/Layers/EventNdjsonLogger.ts created+66
......@@ -0,0 +1,66 @@
1/**
2 * Provider event NDJSON logger helper.
3 *
4 * Best-effort writer for observability logs. Failures are downgraded to warnings
5 * so provider runtime behavior is unaffected.
6 */
7import fs from "node:fs";
8import path from "node:path";
9
10import { createLogger } from "../../logger.ts";
11
12const logger = createLogger("provider-observability");
13
14export interface EventNdjsonLogger {
15 readonly filePath: string;
16 write: (record: unknown) => void;
17 close: () => void;
18}
19
20function stringifyRecord(record: unknown): string | undefined {
21 try {
22 return JSON.stringify(record);
23 } catch (error) {
24 logger.warn("failed to serialize provider event log record", { error });
25 return undefined;
26 }
27}
28
29export function makeEventNdjsonLogger(filePath: string): EventNdjsonLogger | undefined {
30 try {
31 fs.mkdirSync(path.dirname(filePath), { recursive: true });
32 fs.appendFileSync(filePath, "", { encoding: "utf8", flag: "a" });
33 } catch (error) {
34 logger.warn("failed to create provider event log directory", { filePath, error });
35 return undefined;
36 }
37
38 try {
39 let broken = false;
40
41 return {
42 filePath,
43 write(record: unknown) {
44 if (broken) {
45 return;
46 }
47 const serialized = stringifyRecord(record);
48 if (!serialized) {
49 return;
50 }
51 try {
52 fs.appendFileSync(filePath, `${serialized}\n`, { encoding: "utf8", flag: "a" });
53 } catch (error) {
54 broken = true;
55 logger.warn("provider event log write failed", { filePath, error });
56 }
57 },
58 close() {
59 // no-op for sync append logger
60 },
61 };
62 } catch (error) {
63 logger.warn("failed to create provider event log file", { filePath, error });
64 return undefined;
65 }
66}
apps/server/src/provider/Layers/ProviderAdapterRegistry.test.ts+2-2
......@@ -2,7 +2,7 @@ import type { ProviderKind } from "@t3tools/contracts";
22import { it, assert, vi } from "@effect/vitest";
33import { assertFailure } from "@effect/vitest/utils";
44
5import { Effect, Layer } from "effect";
5import { Effect, Layer, Stream } from "effect";
66
77import { CodexAdapter, CodexAdapterShape } from "../Services/CodexAdapter.ts";
88import { ProviderAdapterRegistry } from "../Services/ProviderAdapterRegistry.ts";
......@@ -22,7 +22,7 @@ const fakeCodexAdapter: CodexAdapterShape = {
2222 readThread: vi.fn(),
2323 rollbackThread: vi.fn(),
2424 stopAll: vi.fn(),
25 subscribeToEvents: vi.fn(),
25 streamEvents: Stream.empty,
2626};
2727
2828const layer = it.layer(
apps/server/src/provider/Layers/ProviderService.test.ts+124-37
......@@ -1,11 +1,11 @@
11import type {
22 ProviderApprovalDecision,
3 ProviderEvent,
43 ProviderGetCheckpointDiffInput,
54 ProviderGetCheckpointDiffResult,
65 ProviderListCheckpointsInput,
76 ProviderListCheckpointsResult,
87 ProviderRevertToCheckpointInput,
8 ProviderRuntimeEvent,
99 ProviderRevertToCheckpointResult,
1010 ProviderSendTurnInput,
1111 ProviderSession,
......@@ -16,7 +16,7 @@ import { providerSessionStartInputSchema } from "@t3tools/contracts";
1616import { it, assert, vi } from "@effect/vitest";
1717import { assertFailure } from "@effect/vitest/utils";
1818
19import { Effect, Layer } from "effect";
19import { Effect, Fiber, Layer, PubSub, Ref, Stream } from "effect";
2020
2121import { CheckpointService } from "../../checkpointing/Services/CheckpointService.ts";
2222import {
......@@ -35,7 +35,7 @@ import { NodeServices } from "@effect/platform-node";
3535
3636function makeFakeCodexAdapter() {
3737 const sessions = new Map<string, ProviderSession>();
38 const eventListeners = new Set<(event: ProviderEvent) => void>();
38 const runtimeEventPubSub = Effect.runSync(PubSub.unbounded<ProviderRuntimeEvent>());
3939 let nextSession = 1;
4040
4141 const startSession = vi.fn((input: ProviderSessionStartInput) =>
......@@ -127,16 +127,6 @@ function makeFakeCodexAdapter() {
127127 }),
128128 );
129129
130 const subscribeToEvents = vi.fn(
131 (callback: (event: ProviderEvent) => void): Effect.Effect<() => void, ProviderAdapterError> =>
132 Effect.sync(() => {
133 eventListeners.add(callback);
134 return () => {
135 eventListeners.delete(callback);
136 };
137 }),
138 );
139
140130 const adapter: ProviderAdapterShape<ProviderAdapterError> = {
141131 provider: "codex",
142132 startSession,
......@@ -149,13 +139,11 @@ function makeFakeCodexAdapter() {
149139 readThread,
150140 rollbackThread,
151141 stopAll,
152 subscribeToEvents,
142 streamEvents: Stream.fromPubSub(runtimeEventPubSub),
153143 };
154144
155 const emit = (event: ProviderEvent): void => {
156 for (const listener of eventListeners) {
157 listener(event);
158 }
145 const emit = (event: ProviderRuntimeEvent): void => {
146 Effect.runSync(PubSub.publish(runtimeEventPubSub, event));
159147 };
160148
161149 return {
......@@ -171,7 +159,6 @@ function makeFakeCodexAdapter() {
171159 readThread,
172160 rollbackThread,
173161 stopAll,
174 subscribeToEvents,
175162 };
176163}
177164
......@@ -224,6 +211,7 @@ function makeCheckpointServiceDouble() {
224211 ],
225212 }),
226213 );
214 const releaseSession = vi.fn((_input: { providerSessionId: string }) => Effect.void);
227215
228216 const service: typeof CheckpointService.Service = {
229217 initializeForSession,
......@@ -231,6 +219,7 @@ function makeCheckpointServiceDouble() {
231219 listCheckpoints,
232220 getCheckpointDiff,
233221 revertToCheckpoint,
222 releaseSession,
234223 };
235224
236225 return {
......@@ -240,6 +229,7 @@ function makeCheckpointServiceDouble() {
240229 listCheckpoints,
241230 getCheckpointDiff,
242231 revertToCheckpoint,
232 releaseSession,
243233 };
244234}
245235
......@@ -334,6 +324,9 @@ routing.layer("ProviderServiceLive routing", (it) => {
334324 ]);
335325
336326 yield* provider.stopSession({ sessionId: session.sessionId });
327 assert.deepEqual(routing.checkpoint.releaseSession.mock.calls, [
328 [{ providerSessionId: session.sessionId }],
329 ]);
337330 const sendAfterStop = yield* Effect.result(
338331 provider.sendTurn({
339332 sessionId: session.sessionId,
......@@ -347,6 +340,29 @@ routing.layer("ProviderServiceLive routing", (it) => {
347340 );
348341 }),
349342 );
343
344 it.effect("releases checkpoint session state when stopping all sessions", () =>
345 Effect.gen(function* () {
346 const provider = yield* ProviderService;
347
348 const first = yield* provider.startSession({
349 provider: "codex",
350 });
351 const second = yield* provider.startSession({
352 provider: "codex",
353 });
354
355 yield* provider.stopAll();
356
357 assert.deepEqual(routing.checkpoint.releaseSession.mock.calls.slice(-2), [
358 [{ providerSessionId: first.sessionId }],
359 [{ providerSessionId: second.sessionId }],
360 ]);
361
362 const remaining = yield* provider.listSessions();
363 assert.equal(remaining.length, 0);
364 }),
365 );
350366});
351367
352368const fanout = makeProviderServiceLayer();
......@@ -358,40 +374,111 @@ fanout.layer("ProviderServiceLive fanout", (it) => {
358374 provider: "codex",
359375 });
360376
361 const onEvent = vi.fn((_event: ProviderEvent) => undefined);
362 const unsubscribe = yield* provider.subscribeToEvents(onEvent);
377 const eventsRef = yield* Ref.make<Array<ProviderRuntimeEvent>>([]);
378 const consumer = yield* Stream.runForEach(provider.streamEvents, (event) =>
379 Ref.update(eventsRef, (current) => [...current, event]),
380 ).pipe(
381 Effect.forkChild,
382 );
383 yield* Effect.promise(() => new Promise<void>((resolve) => setTimeout(resolve, 0)));
363384
364 const completedEvent: ProviderEvent = {
365 id: "evt-1",
366 kind: "notification",
385 const completedEvent: ProviderRuntimeEvent = {
386 type: "turn.completed",
387 eventId: "evt-1",
367388 provider: "codex",
368389 sessionId: session.sessionId,
369390 createdAt: new Date().toISOString(),
370 method: "turn/completed",
371391 threadId: "thread-1",
372392 turnId: "turn-1",
373 payload: {
374 turn: {
375 status: "completed",
376 },
377 },
393 status: "completed",
378394 };
379395
380396 fanout.codex.emit(completedEvent);
381397 yield* Effect.promise(() => new Promise<void>((resolve) => setTimeout(resolve, 0)));
382398
383 assert.equal(onEvent.mock.calls.length, 1);
384 assert.equal(onEvent.mock.calls[0]?.[0]?.method, "turn/completed");
399 const events = yield* Ref.get(eventsRef);
400 yield* Fiber.interrupt(consumer);
401
402 assert.equal(events.length >= 2, true);
403 assert.equal(events[0]?.type, "turn.completed");
404 assert.equal(events[1]?.type, "checkpoint.captured");
385405 assert.deepEqual(fanout.checkpoint.captureCurrentTurn.mock.calls, [
386406 [{ providerSessionId: session.sessionId, turnId: "turn-1", status: "completed" }],
387407 ]);
408 }),
409 );
388410
389 unsubscribe();
390 fanout.codex.emit({
391 ...completedEvent,
392 id: "evt-2",
411 it.effect("keeps subscriber delivery ordered and isolates failing subscribers", () =>
412 Effect.gen(function* () {
413 const provider = yield* ProviderService;
414 const session = yield* provider.startSession({
415 provider: "codex",
393416 });
394 assert.equal(onEvent.mock.calls.length, 1);
417 fanout.checkpoint.captureCurrentTurn.mockClear();
418
419 const receivedByHealthy: string[] = [];
420 const healthyFiber = yield* Stream.take(provider.streamEvents, 3).pipe(
421 Stream.runForEach((event) =>
422 Effect.sync(() => {
423 receivedByHealthy.push(event.eventId);
424 }),
425 ),
426 Effect.forkChild,
427 );
428 const failingFiber = yield* Stream.take(provider.streamEvents, 1).pipe(
429 Stream.runForEach(() => Effect.fail("listener crash")),
430 Effect.forkChild,
431 );
432 yield* Effect.promise(() => new Promise<void>((resolve) => setTimeout(resolve, 0)));
433
434 const events: ReadonlyArray<ProviderRuntimeEvent> = [
435 {
436 type: "tool.completed",
437 eventId: "evt-ordered-1",
438 provider: "codex",
439 sessionId: session.sessionId,
440 createdAt: new Date().toISOString(),
441 threadId: "thread-1",
442 turnId: "turn-1",
443 toolKind: "command",
444 title: "Command run",
445 detail: "echo one",
446 },
447 {
448 type: "message.delta",
449 eventId: "evt-ordered-2",
450 provider: "codex",
451 sessionId: session.sessionId,
452 createdAt: new Date().toISOString(),
453 threadId: "thread-1",
454 turnId: "turn-1",
455 delta: "hello",
456 },
457 {
458 type: "turn.completed",
459 eventId: "evt-ordered-3",
460 provider: "codex",
461 sessionId: session.sessionId,
462 createdAt: new Date().toISOString(),
463 threadId: "thread-1",
464 turnId: "turn-1",
465 status: "completed",
466 },
467 ];
468
469 for (const event of events) {
470 fanout.codex.emit(event);
471 }
472 const failingResult = yield* Effect.result(Fiber.join(failingFiber));
473 assert.equal(failingResult._tag, "Failure");
474 yield* Fiber.join(healthyFiber);
475
476 assert.deepEqual(receivedByHealthy.slice(0, 3), [
477 "evt-ordered-1",
478 "evt-ordered-2",
479 "evt-ordered-3",
480 ]);
481 assert.equal(receivedByHealthy.length, 3);
395482 }),
396483 );
397484});
apps/server/src/provider/Layers/ProviderService.ts+122-92
......@@ -12,7 +12,6 @@
1212 */
1313import { randomUUID } from "node:crypto";
1414
15import type { ProviderEvent } from "@t3tools/contracts";
1615import {
1716 providerGetCheckpointDiffInputSchema,
1817 providerInterruptTurnInputSchema,
......@@ -22,46 +21,54 @@ import {
2221 providerSendTurnInputSchema,
2322 providerSessionStartInputSchema,
2423 providerStopSessionInputSchema,
24 type ProviderRuntimeEvent,
25 type ProviderRuntimeTurnCompletedEvent,
2526} from "@t3tools/contracts";
26import { Effect, Layer } from "effect";
27import { Effect, Layer, PubSub, Queue, Stream } from "effect";
2728
2829import { CheckpointService } from "../../checkpointing/Services/CheckpointService.ts";
2930import { ProviderValidationError } from "../Errors.ts";
3031import { ProviderAdapterRegistry } from "../Services/ProviderAdapterRegistry.ts";
3132import { ProviderService, type ProviderServiceShape } from "../Services/ProviderService.ts";
3233import { ProviderSessionDirectory } from "../Services/ProviderSessionDirectory.ts";
34import { makeEventNdjsonLogger } from "./EventNdjsonLogger.ts";
3335
34function asObject(value: unknown): Record<string, unknown> | undefined {
35 if (!value || typeof value !== "object") {
36 return undefined;
37 }
38 return value as Record<string, unknown>;
39}
40
41function asString(value: unknown): string | undefined {
42 return typeof value === "string" ? value : undefined;
43}
44
45function readTurnCompletionStatus(event: ProviderEvent): string | undefined {
46 const payload = asObject(event.payload);
47 const turn = asObject(payload?.turn);
48 return asString(turn?.status);
36export interface ProviderServiceLiveOptions {
37 readonly canonicalEventLogPath?: string;
4938}
5039
5140function toCheckpointCaptureErrorEvent(
52 event: ProviderEvent,
41 event: ProviderRuntimeTurnCompletedEvent,
5342 error: { readonly message: string },
54): ProviderEvent {
43): ProviderRuntimeEvent {
5544 return {
56 id: randomUUID(),
57 kind: "error",
45 type: "runtime.error",
46 eventId: randomUUID(),
5847 provider: event.provider,
5948 sessionId: event.sessionId,
6049 createdAt: new Date().toISOString(),
61 method: "checkpoint/captureError",
62 message: error.message,
6350 ...(event.threadId !== undefined ? { threadId: event.threadId } : {}),
6451 ...(event.turnId !== undefined ? { turnId: event.turnId } : {}),
52 message: error.message,
53 };
54}
55
56function toCheckpointCapturedEvent(input: {
57 readonly event: ProviderRuntimeTurnCompletedEvent;
58 readonly threadId: string;
59 readonly turnCount: number;
60}): ProviderRuntimeEvent {
61 const { event, threadId, turnCount } = input;
62 return {
63 type: "checkpoint.captured",
64 eventId: randomUUID(),
65 provider: event.provider,
66 sessionId: event.sessionId,
67 createdAt: new Date().toISOString(),
68 threadId,
69 ...(event.turnId !== undefined ? { turnId: event.turnId } : {}),
70 turnCount,
71 ...(event.status !== undefined ? { status: event.status } : {}),
6572 };
6673}
6774
......@@ -77,58 +84,80 @@ function toValidationError(
7784 });
7885}
7986
80const makeProviderService = Effect.gen(function* () {
81 const registry = yield* ProviderAdapterRegistry;
82 const directory = yield* ProviderSessionDirectory;
83 const checkpointService = yield* CheckpointService;
84
85 const eventSubscribers = new Set<(event: ProviderEvent) => void>();
86 const notifySubscribers = (event: ProviderEvent): void => {
87 for (const callback of eventSubscribers) {
88 try {
89 callback(event);
90 } catch {
91 // Subscriber callbacks must not destabilize provider event fanout.
92 }
93 }
94 };
87const makeProviderService = (options?: ProviderServiceLiveOptions) =>
88 Effect.gen(function* () {
89 const canonicalEventLogger =
90 options?.canonicalEventLogPath !== undefined
91 ? makeEventNdjsonLogger(options.canonicalEventLogPath)
92 : undefined;
93
94 const registry = yield* ProviderAdapterRegistry;
95 const directory = yield* ProviderSessionDirectory;
96 const checkpointService = yield* CheckpointService;
97
98 const runtimeEventQueue = yield* Queue.unbounded<ProviderRuntimeEvent>();
99 const runtimeEventPubSub = yield* PubSub.unbounded<ProviderRuntimeEvent>();
100
101 const publishRuntimeEvent = (event: ProviderRuntimeEvent): Effect.Effect<void> =>
102 Effect.sync(() => {
103 canonicalEventLogger?.write({
104 observedAt: new Date().toISOString(),
105 event,
106 });
107 }).pipe(
108 Effect.flatMap(() => PubSub.publish(runtimeEventPubSub, event)),
109 Effect.asVoid,
110 );
95111
96 const providers = yield* registry.listProviders();
97 const adapters = yield* Effect.forEach(providers, (provider) => registry.getByProvider(provider));
112 const providers = yield* registry.listProviders();
113 const adapters = yield* Effect.forEach(providers, (provider) => registry.getByProvider(provider));
114
115 const onTurnCompleted = (event: ProviderRuntimeTurnCompletedEvent): Effect.Effect<void> =>
116 checkpointService
117 .captureCurrentTurn({
118 providerSessionId: event.sessionId,
119 ...(event.turnId !== undefined ? { turnId: event.turnId } : {}),
120 ...(event.status !== undefined ? { status: event.status } : {}),
121 })
122 .pipe(
123 Effect.flatMap(() => checkpointService.listCheckpoints({ sessionId: event.sessionId })),
124 Effect.flatMap((result) => {
125 const currentCheckpoint =
126 result.checkpoints.find((checkpoint) => checkpoint.isCurrent) ??
127 result.checkpoints[result.checkpoints.length - 1];
128 if (!currentCheckpoint) {
129 return Effect.void;
130 }
131
132 return publishRuntimeEvent(
133 toCheckpointCapturedEvent({
134 event,
135 threadId: result.threadId,
136 turnCount: currentCheckpoint.turnCount,
137 }),
138 );
139 }),
140 Effect.catch((error) => publishRuntimeEvent(toCheckpointCaptureErrorEvent(event, error))),
141 );
98142
99 yield* Effect.acquireRelease(
100 Effect.forEach(adapters, (adapter) =>
101 adapter.subscribeToEvents((event) => {
102 notifySubscribers(event);
103 if (event.kind !== "notification" || event.method !== "turn/completed") {
104 return;
143 const processRuntimeEvent = (event: ProviderRuntimeEvent): Effect.Effect<void> =>
144 publishRuntimeEvent(event).pipe(
145 Effect.flatMap(() => {
146 if (event.type !== "turn.completed") {
147 return Effect.void;
105148 }
106
107 const status = readTurnCompletionStatus(event);
108 const captureInput = {
109 providerSessionId: event.sessionId,
110 ...(event.turnId !== undefined ? { turnId: event.turnId } : {}),
111 ...(status !== undefined ? { status } : {}),
112 };
113
114 Effect.runFork(
115 checkpointService.captureCurrentTurn(captureInput).pipe(
116 Effect.catch((error) =>
117 Effect.sync(() => {
118 notifySubscribers(toCheckpointCaptureErrorEvent(event, error));
119 }),
120 ),
121 ),
122 );
149 return onTurnCompleted(event);
123150 }),
124 ),
125 (unsubscribeAll) =>
126 Effect.forEach(unsubscribeAll, (unsubscribe) =>
127 Effect.sync(() => {
128 unsubscribe();
129 }),
130 ).pipe(Effect.asVoid),
131 );
151 );
152
153 const worker = Effect.forever(Queue.take(runtimeEventQueue).pipe(Effect.flatMap(processRuntimeEvent)));
154 yield* Effect.forkScoped(worker);
155
156 yield* Effect.forEach(adapters, (adapter) =>
157 Stream.runForEach(adapter.streamEvents, (event) =>
158 Queue.offer(runtimeEventQueue, event).pipe(Effect.asVoid),
159 ).pipe(Effect.forkScoped),
160 ).pipe(Effect.asVoid);
132161
133162 const adapterForSession = (sessionId: string) =>
134163 directory
......@@ -252,6 +281,7 @@ const makeProviderService = Effect.gen(function* () {
252281 const input = parsed.data;
253282 const adapter = yield* adapterForSession(input.sessionId);
254283 yield* adapter.stopSession(input.sessionId);
284 yield* checkpointService.releaseSession({ providerSessionId: input.sessionId });
255285 yield* directory.remove(input.sessionId);
256286 });
257287
......@@ -316,34 +346,34 @@ const makeProviderService = Effect.gen(function* () {
316346
317347 const stopAll: ProviderServiceShape["stopAll"] = () =>
318348 Effect.gen(function* () {
319 yield* Effect.forEach(adapters, (adapter) => adapter.stopAll()).pipe(Effect.asVoid);
320349 const sessionIds = yield* directory.listSessionIds();
321 yield* Effect.forEach(sessionIds, (sessionId) => directory.remove(sessionId)).pipe(
322 Effect.asVoid,
323 );
324 });
325
326 const subscribeToEvents: ProviderServiceShape["subscribeToEvents"] = (callback) =>
327 Effect.sync(() => {
328 eventSubscribers.add(callback);
329 return () => {
330 eventSubscribers.delete(callback);
331 };
350 yield* Effect.forEach(adapters, (adapter) => adapter.stopAll()).pipe(Effect.asVoid);
351 yield* Effect.forEach(
352 sessionIds,
353 (sessionId) =>
354 checkpointService.releaseSession({ providerSessionId: sessionId }).pipe(
355 Effect.flatMap(() => directory.remove(sessionId)),
356 ),
357 ).pipe(Effect.asVoid);
332358 });
333359
334 return {
335 startSession,
336 sendTurn,
337 interruptTurn,
360 return {
361 startSession,
362 sendTurn,
363 interruptTurn,
338364 respondToRequest,
339365 stopSession,
340366 listSessions,
341367 listCheckpoints,
342368 getCheckpointDiff,
343369 revertToCheckpoint,
344 stopAll,
345 subscribeToEvents,
346 } satisfies ProviderServiceShape;
347});
370 stopAll,
371 streamEvents: Stream.fromPubSub(runtimeEventPubSub),
372 } satisfies ProviderServiceShape;
373 });
348374
349export const ProviderServiceLive = Layer.effect(ProviderService, makeProviderService);
375export const ProviderServiceLive = Layer.effect(ProviderService, makeProviderService());
376
377export function makeProviderServiceLive(options?: ProviderServiceLiveOptions) {
378 return Layer.effect(ProviderService, makeProviderService(options));
379}
apps/server/src/provider/Services/ProviderAdapter.ts+4-5
......@@ -9,14 +9,15 @@
99 */
1010import type {
1111 ProviderApprovalDecision,
12 ProviderEvent,
1312 ProviderKind,
13 ProviderRuntimeEvent,
1414 ProviderSendTurnInput,
1515 ProviderSession,
1616 ProviderSessionStartInput,
1717 ProviderTurnStartResult,
1818} from "@t3tools/contracts";
1919import type { Effect } from "effect";
20import type { Stream } from "effect";
2021
2122export interface ProviderThreadTurnSnapshot {
2223 readonly id: string;
......@@ -96,9 +97,7 @@ export interface ProviderAdapterShape<TError> {
9697 readonly stopAll: () => Effect.Effect<void, TError>;
9798
9899 /**
99 * Subscribe to provider events emitted by this adapter.
100 * Canonical runtime event stream emitted by this adapter.
100101 */
101 readonly subscribeToEvents: (
102 callback: (event: ProviderEvent) => void,
103 ) => Effect.Effect<() => void, TError>;
102 readonly streamEvents: Stream.Stream<ProviderRuntimeEvent>;
104103}
apps/server/src/provider/Services/ProviderService.ts+4-6
......@@ -12,7 +12,6 @@
1212 * @module ProviderService
1313 */
1414import type {
15 ProviderEvent,
1615 ProviderGetCheckpointDiffInput,
1716 ProviderGetCheckpointDiffResult,
1817 ProviderInterruptTurnInput,
......@@ -26,9 +25,10 @@ import type {
2625 ProviderSessionStartInput,
2726 ProviderStopSessionInput,
2827 ProviderTurnStartResult,
28 ProviderRuntimeEvent,
2929} from "@t3tools/contracts";
3030import { ServiceMap } from "effect";
31import type { Effect } from "effect";
31import type { Effect, Stream } from "effect";
3232
3333import type { ProviderServiceError } from "../Errors.ts";
3434
......@@ -100,13 +100,11 @@ export interface ProviderServiceShape {
100100 readonly stopAll: () => Effect.Effect<void, ProviderServiceError>;
101101
102102 /**
103 * Subscribe to provider event stream.
103 * Canonical provider runtime event stream.
104104 *
105105 * Fan-out is owned by ProviderService (not by a standalone event-bus service).
106106 */
107 readonly subscribeToEvents: (
108 callback: (event: ProviderEvent) => void,
109 ) => Effect.Effect<() => void, ProviderServiceError>;
107 readonly streamEvents: Stream.Stream<ProviderRuntimeEvent>;
110108}
111109
112110/**
apps/server/src/provider/contracts.test.ts deleted-245
......@@ -1,245 +0,0 @@
1import { describe, expect, it } from "vitest";
2
3import { Effect, Layer, ManagedRuntime, Option } from "effect";
4
5import {
6 ProviderAdapterValidationError,
7 ProviderFilesystemError,
8 ProviderSessionNotFoundError,
9 ProviderUnsupportedError,
10 type ProviderAdapterError,
11 type ProviderServiceError,
12} from "./Errors.ts";
13import {
14 CheckpointGitCommandError,
15 CheckpointServiceValidationError,
16 type CheckpointServiceError,
17 type CheckpointStoreError,
18} from "../checkpointing/Errors.ts";
19import {
20 CheckpointRepositoryValidationError,
21 type CheckpointRepositoryError,
22} from "../persistence/Errors.ts";
23import { CheckpointService } from "../checkpointing/Services/CheckpointService.ts";
24import { CheckpointStore } from "../checkpointing/Services/CheckpointStore.ts";
25import { CheckpointRepository } from "../persistence/Services/Checkpoints.ts";
26import { CodexAdapter } from "./Services/CodexAdapter.ts";
27import { ProviderAdapterRegistry } from "./Services/ProviderAdapterRegistry.ts";
28import { ProviderService } from "./Services/ProviderService.ts";
29import { ProviderSessionDirectory } from "./Services/ProviderSessionDirectory.ts";
30
31describe("provider/checkpoint service contracts", () => {
32 it("resolves all service tags from ServiceMap", async () => {
33 const providerLive: typeof ProviderService.Service = {
34 startSession: () =>
35 Effect.fail<ProviderServiceError>(new ProviderUnsupportedError({ provider: "x" })),
36 sendTurn: () =>
37 Effect.fail<ProviderServiceError>(new ProviderUnsupportedError({ provider: "x" })),
38 interruptTurn: () =>
39 Effect.fail<ProviderServiceError>(new ProviderUnsupportedError({ provider: "x" })),
40 respondToRequest: () =>
41 Effect.fail<ProviderServiceError>(new ProviderUnsupportedError({ provider: "x" })),
42 stopSession: () => Effect.void,
43 listSessions: () => Effect.succeed([]),
44 listCheckpoints: () =>
45 Effect.fail<ProviderServiceError>(new ProviderUnsupportedError({ provider: "x" })),
46 getCheckpointDiff: () =>
47 Effect.fail<ProviderServiceError>(new ProviderUnsupportedError({ provider: "x" })),
48 revertToCheckpoint: () =>
49 Effect.fail<ProviderServiceError>(new ProviderUnsupportedError({ provider: "x" })),
50 stopAll: () => Effect.void,
51 subscribeToEvents: () => Effect.succeed(() => undefined),
52 };
53
54 const codexLive: typeof CodexAdapter.Service = {
55 provider: "codex",
56 startSession: () =>
57 Effect.fail<ProviderAdapterError>(
58 new ProviderAdapterValidationError({
59 provider: "codex",
60 operation: "startSession",
61 issue: "not implemented",
62 }),
63 ),
64 sendTurn: () =>
65 Effect.fail<ProviderAdapterError>(
66 new ProviderAdapterValidationError({
67 provider: "codex",
68 operation: "sendTurn",
69 issue: "not implemented",
70 }),
71 ),
72 interruptTurn: () =>
73 Effect.fail<ProviderAdapterError>(
74 new ProviderAdapterValidationError({
75 provider: "codex",
76 operation: "interruptTurn",
77 issue: "not implemented",
78 }),
79 ),
80 readThread: () =>
81 Effect.fail<ProviderAdapterError>(
82 new ProviderAdapterValidationError({
83 provider: "codex",
84 operation: "readThread",
85 issue: "not implemented",
86 }),
87 ),
88 rollbackThread: () =>
89 Effect.fail<ProviderAdapterError>(
90 new ProviderAdapterValidationError({
91 provider: "codex",
92 operation: "rollbackThread",
93 issue: "not implemented",
94 }),
95 ),
96 respondToRequest: () =>
97 Effect.fail<ProviderAdapterError>(
98 new ProviderAdapterValidationError({
99 provider: "codex",
100 operation: "respondToRequest",
101 issue: "not implemented",
102 }),
103 ),
104 stopSession: () => Effect.void,
105 listSessions: () => Effect.succeed([]),
106 hasSession: () => Effect.succeed(false),
107 stopAll: () => Effect.void,
108 subscribeToEvents: () => Effect.succeed(() => undefined),
109 };
110
111 const checkpointStoreLive: typeof CheckpointStore.Service = {
112 isGitRepository: () => Effect.succeed(false),
113 captureCheckpoint: () => Effect.void,
114 hasCheckpointRef: () => Effect.succeed(false),
115 ensureRootCheckpoint: () => Effect.succeed(false),
116 restoreCheckpoint: () => Effect.succeed(false),
117 diffCheckpoints: () =>
118 Effect.fail<CheckpointStoreError>(
119 new CheckpointGitCommandError({
120 operation: "diffCheckpoints",
121 command: "git diff",
122 cwd: "/tmp",
123 detail: "not implemented",
124 }),
125 ),
126 deleteCheckpointRefs: () => Effect.void,
127 };
128
129 const checkpointRepositoryLive: typeof CheckpointRepository.Service = {
130 upsertCheckpoint: () => Effect.void,
131 listCheckpoints: () => Effect.succeed([]),
132 getCheckpoint: () => Effect.succeed(Option.none()),
133 deleteAfterTurn: () => Effect.void,
134 deleteAllForSession: () => Effect.void,
135 };
136
137 const checkpointServiceLive: typeof CheckpointService.Service = {
138 initializeForSession: () => Effect.void,
139 captureCurrentTurn: () => Effect.void,
140 listCheckpoints: () =>
141 Effect.fail<CheckpointServiceError>(
142 new CheckpointServiceValidationError({
143 operation: "listCheckpoints",
144 issue: "not implemented",
145 }),
146 ),
147 getCheckpointDiff: () =>
148 Effect.fail<CheckpointServiceError>(
149 new CheckpointServiceValidationError({
150 operation: "getCheckpointDiff",
151 issue: "not implemented",
152 }),
153 ),
154 revertToCheckpoint: () =>
155 Effect.fail<CheckpointServiceError>(
156 new CheckpointServiceValidationError({
157 operation: "revertToCheckpoint",
158 issue: "not implemented",
159 }),
160 ),
161 };
162
163 const providerAdapterRegistryLive: typeof ProviderAdapterRegistry.Service = {
164 getByProvider: (provider) =>
165 provider === "codex"
166 ? Effect.succeed(codexLive)
167 : Effect.fail<ProviderUnsupportedError>(new ProviderUnsupportedError({ provider })),
168 listProviders: () => Effect.succeed(["codex"]),
169 };
170
171 const providerSessionDirectoryLive: typeof ProviderSessionDirectory.Service = {
172 upsert: () => Effect.void,
173 getProvider: (sessionId) =>
174 Effect.fail<ProviderSessionNotFoundError>(new ProviderSessionNotFoundError({ sessionId })),
175 getThreadId: (sessionId) =>
176 Effect.fail<ProviderSessionNotFoundError>(new ProviderSessionNotFoundError({ sessionId })),
177 remove: () => Effect.void,
178 listSessionIds: () => Effect.succeed([]),
179 };
180
181 const runtime = ManagedRuntime.make(
182 Layer.mergeAll(
183 Layer.succeed(ProviderService, providerLive),
184 Layer.succeed(CodexAdapter, codexLive),
185 Layer.succeed(CheckpointStore, checkpointStoreLive),
186 Layer.succeed(CheckpointRepository, checkpointRepositoryLive),
187 Layer.succeed(CheckpointService, checkpointServiceLive),
188 Layer.succeed(ProviderAdapterRegistry, providerAdapterRegistryLive),
189 Layer.succeed(ProviderSessionDirectory, providerSessionDirectoryLive),
190 ),
191 );
192
193 const [
194 provider,
195 codex,
196 checkpointStore,
197 checkpointRepository,
198 checkpointService,
199 providerAdapterRegistry,
200 providerSessionDirectory,
201 ] = await Promise.all([
202 runtime.runPromise(Effect.service(ProviderService)),
203 runtime.runPromise(Effect.service(CodexAdapter)),
204 runtime.runPromise(Effect.service(CheckpointStore)),
205 runtime.runPromise(Effect.service(CheckpointRepository)),
206 runtime.runPromise(Effect.service(CheckpointService)),
207 runtime.runPromise(Effect.service(ProviderAdapterRegistry)),
208 runtime.runPromise(Effect.service(ProviderSessionDirectory)),
209 ]);
210
211 expect(provider).toBe(providerLive);
212 expect(codex).toBe(codexLive);
213 expect(checkpointStore).toBe(checkpointStoreLive);
214 expect(checkpointRepository).toBe(checkpointRepositoryLive);
215 expect(checkpointService).toBe(checkpointServiceLive);
216 expect(providerAdapterRegistry).toBe(providerAdapterRegistryLive);
217 expect(providerSessionDirectory).toBe(providerSessionDirectoryLive);
218
219 await runtime.dispose();
220 });
221});
222
223describe("provider/checkpoint errors", () => {
224 it("preserves cause and message on tagged errors", () => {
225 const fsCause = new Error("filesystem failed");
226 const fsError = new ProviderFilesystemError({
227 sessionId: "sess-1",
228 detail: "capture failed",
229 cause: fsCause,
230 });
231
232 expect(fsError._tag).toBe("ProviderFilesystemError");
233 expect(fsError.message).toContain("sess-1");
234 expect(fsError.cause).toBe(fsCause);
235 });
236
237 it("accepts checkpoint repository error channel type", () => {
238 const error: CheckpointRepositoryError = new CheckpointRepositoryValidationError({
239 operation: "upsertCheckpoint",
240 issue: "turnCount must be >= 0",
241 });
242
243 expect(error._tag).toBe("CheckpointRepositoryValidationError");
244 });
245});
apps/server/src/wsServer.test.ts+1-2
......@@ -628,8 +628,7 @@ describe("WebSocket Server", () => {
628628 const ws = await connectWs(port);
629629 connections.push(ws);
630630
631 // Consume welcome + initial orchestration snapshot pushes
632 await waitForMessage(ws);
631 // Consume welcome push
633632 await waitForMessage(ws);
634633
635634 const response = await sendRequest(ws, "nonexistent.method");
apps/server/src/wsServer.ts+284-152
......@@ -11,7 +11,7 @@ import {
1111 ORCHESTRATION_WS_METHODS,
1212 WS_CHANNELS,
1313 WS_METHODS,
14 type OrchestrationReadModelPush,
14 type ProviderRuntimeEvent,
1515 type TerminalEvent,
1616 type WsPush,
1717 type WsRequest,
......@@ -19,7 +19,7 @@ import {
1919 wsRequestSchema,
2020} from "@t3tools/contracts";
2121import { NodeServices } from "@effect/platform-node";
22import { Effect, Layer, ManagedRuntime } from "effect";
22import { Effect, Exit, Layer, ManagedRuntime, Queue, Scope, Stream } from "effect";
2323import { WebSocketServer, type WebSocket } from "ws";
2424
2525import { createLogger } from "./logger";
......@@ -43,14 +43,18 @@ import { makeSqlitePersistenceLive } from "./persistence/Layers/Sqlite";
4343import { ProjectRepository, type ProjectRepositoryShape } from "./persistence/Services/Projects";
4444import assert from "node:assert";
4545import { OrchestrationEventRepositoryLive } from "./persistence/Layers/OrchestrationEvents";
46import { CodexAdapterLive } from "./provider/Layers/CodexAdapter";
46import { makeCodexAdapterLive } from "./provider/Layers/CodexAdapter";
4747import { ProviderAdapterRegistryLive } from "./provider/Layers/ProviderAdapterRegistry";
48import { ProviderServiceLive } from "./provider/Layers/ProviderService";
48import { makeProviderServiceLive } from "./provider/Layers/ProviderService";
4949import { ProviderSessionDirectoryLive } from "./provider/Layers/ProviderSessionDirectory";
5050import { ProviderService, type ProviderServiceShape } from "./provider/Services/ProviderService";
5151import { CheckpointStoreLive } from "./checkpointing/Layers/CheckpointStore";
5252import { CheckpointServiceLive } from "./checkpointing/Layers/CheckpointService";
5353import { CheckpointRepositoryLive } from "./persistence/Layers/Checkpoints";
54import {
55 makeEventNdjsonLogger,
56 type EventNdjsonLogger,
57} from "./provider/Layers/EventNdjsonLogger";
5458import * as SqlClient from "effect/unstable/sql/SqlClient";
5559import * as SqlError from "effect/unstable/sql/SqlError";
5660import * as Migrator from "effect/unstable/sql/Migrator";
......@@ -114,17 +118,90 @@ function parseBooleanEnv(value: string | undefined): boolean | undefined {
114118 return undefined;
115119}
116120
117function asObject(value: unknown): Record<string, unknown> | null {
118 if (!value || typeof value !== "object") return null;
119 return value as Record<string, unknown>;
121const providerTurnKey = (sessionId: string, turnId: string) => `${sessionId}:${turnId}`;
122
123type ActivityTone = "thinking" | "tool" | "info" | "error";
124
125interface ThreadActivityInput {
126 readonly id: string;
127 readonly createdAt: string;
128 readonly label: string;
129 readonly detail?: string;
130 readonly tone: ActivityTone;
131 readonly turnId?: string;
132 readonly requestId?: string;
133 readonly requestKind?: "command" | "file-change";
120134}
121135
122function asString(value: unknown): string | null {
123 return typeof value === "string" ? value : null;
136function truncateDetail(value: string, limit = 180): string {
137 return value.length > limit ? `${value.slice(0, limit - 1)}...` : value;
124138}
125139
126function asNumber(value: unknown): number | null {
127 return typeof value === "number" && Number.isFinite(value) ? value : null;
140function runtimeEventToActivities(event: ProviderRuntimeEvent): ReadonlyArray<ThreadActivityInput> {
141 switch (event.type) {
142 case "approval.requested":
143 return [
144 {
145 id: event.eventId,
146 createdAt: event.createdAt,
147 label:
148 event.requestKind === "command"
149 ? "Command approval requested"
150 : "File-change approval requested",
151 ...(event.detail ? { detail: truncateDetail(event.detail) } : {}),
152 tone: "tool",
153 ...(event.turnId ? { turnId: event.turnId } : {}),
154 requestId: event.requestId,
155 requestKind: event.requestKind,
156 },
157 ];
158 case "approval.resolved":
159 return [
160 {
161 id: event.eventId,
162 createdAt: event.createdAt,
163 label: "Approval resolved",
164 tone: "info",
165 ...(event.turnId ? { turnId: event.turnId } : {}),
166 requestId: event.requestId,
167 ...(event.requestKind ? { requestKind: event.requestKind } : {}),
168 },
169 ];
170 case "runtime.error":
171 return [
172 {
173 id: event.eventId,
174 createdAt: event.createdAt,
175 label: "Runtime error",
176 detail: truncateDetail(event.message),
177 tone: "error",
178 ...(event.turnId ? { turnId: event.turnId } : {}),
179 },
180 ];
181 case "checkpoint.captured":
182 return [
183 {
184 id: event.eventId,
185 createdAt: event.createdAt,
186 label: "Checkpoint captured",
187 tone: "info",
188 ...(event.turnId ? { turnId: event.turnId } : {}),
189 },
190 ];
191 case "tool.completed":
192 return [
193 {
194 id: event.eventId,
195 createdAt: event.createdAt,
196 label: `${event.title} complete`,
197 ...(event.detail ? { detail: truncateDetail(event.detail) } : {}),
198 tone: "tool",
199 ...(event.turnId ? { turnId: event.turnId } : {}),
200 },
201 ];
202 default:
203 return [];
204 }
128205}
129206
130207const noop = () => {};
......@@ -157,7 +234,8 @@ export function createServer(options: ServerOptions) {
157234 const logWebSocketEvents =
158235 explicitLogWsEvents ?? parseBooleanEnv(process.env.T3CODE_LOG_WS_EVENTS) ?? Boolean(devUrl);
159236 let keybindingsConfig = loadResolvedKeybindingsConfig(logger);
160 let unsubscribeReadModel = noop;
237 let orchestrationDomainEventLogger: EventNdjsonLogger | undefined;
238 let orchestrationCommandLogger: EventNdjsonLogger | undefined;
161239 let unsubscribeDomainEvents = noop;
162240 let unsubscribeProviderEvents = noop;
163241
......@@ -197,30 +275,12 @@ export function createServer(options: ServerOptions) {
197275 >,
198276 orchestrationEngine: OrchestrationEngineShape,
199277 ) {
200 unsubscribeReadModel = await runtime.runPromise(
201 orchestrationEngine.subscribeToReadModel((snapshot) => {
202 const push: WsPush = {
203 type: "push",
204 channel: ORCHESTRATION_WS_CHANNELS.readModel,
205 data: {
206 sequence: snapshot.sequence,
207 snapshot,
208 } satisfies OrchestrationReadModelPush,
209 };
210 const message = JSON.stringify(push);
211 let recipients = 0;
212 for (const client of clients) {
213 if (client.readyState === client.OPEN) {
214 client.send(message);
215 recipients += 1;
216 }
217 }
218 logOutgoingPush(push, recipients);
219 }),
220 );
221
222278 unsubscribeDomainEvents = await runtime.runPromise(
223279 orchestrationEngine.subscribeToDomainEvents((event) => {
280 orchestrationDomainEventLogger?.write({
281 observedAt: new Date().toISOString(),
282 event,
283 });
224284 const push: WsPush = {
225285 type: "push",
226286 channel: ORCHESTRATION_WS_CHANNELS.domainEvent,
......@@ -247,53 +307,96 @@ export function createServer(options: ServerOptions) {
247307 providerService: ProviderServiceShape,
248308 orchestrationEngine: OrchestrationEngineShape,
249309 ) {
250 unsubscribeProviderEvents = await runtime.runPromise(
251 providerService.subscribeToEvents((event) => {
252 void (async () => {
253 const snapshot = await runtime.runPromise(orchestrationEngine.getSnapshot());
254 const thread = snapshot.threads.find(
255 (entry) => entry.session?.sessionId === event.sessionId,
256 );
257 if (!thread) return;
258 const now = event.createdAt;
259 if (event.method === "turn/started" || event.method === "turn/completed") {
260 await runtime.runPromise(
261 orchestrationEngine.dispatch({
262 type: "thread.session",
263 commandId: crypto.randomUUID(),
264 threadId: thread.id,
265 session: {
266 sessionId: event.sessionId,
267 provider: event.provider,
268 status: event.method === "turn/started" ? "running" : "ready",
269 threadId: thread.id,
270 activeTurnId: event.method === "turn/started" ? (event.turnId ?? null) : null,
271 createdAt: thread.session?.createdAt ?? now,
272 updatedAt: now,
273 lastError: null,
274 },
275 createdAt: now,
276 }),
277 );
278 }
279 if (event.textDelta && event.textDelta.length > 0) {
280 const assistantMessageId = `assistant:${event.turnId ?? event.itemId ?? event.sessionId}`;
281 await runtime.runPromise(
282 orchestrationEngine.dispatch({
283 type: "message.send",
284 commandId: crypto.randomUUID(),
285 threadId: thread.id,
286 messageId: assistantMessageId,
287 role: "assistant",
288 text: event.textDelta,
289 streaming: true,
290 createdAt: now,
291 }),
292 );
310 const turnMessageIdsByTurnKey = new Map<string, Set<string>>();
311 const latestMessageIdByTurnKey = new Map<string, string>();
312
313 const rememberAssistantMessageId = (sessionId: string, turnId: string, messageId: string) => {
314 const key = providerTurnKey(sessionId, turnId);
315 const existingIds = turnMessageIdsByTurnKey.get(key);
316 if (existingIds) {
317 existingIds.add(messageId);
318 } else {
319 turnMessageIdsByTurnKey.set(key, new Set([messageId]));
320 }
321 latestMessageIdByTurnKey.set(key, messageId);
322 };
323
324 const getAssistantMessageIdsForTurn = (sessionId: string, turnId: string) => {
325 return turnMessageIdsByTurnKey.get(providerTurnKey(sessionId, turnId)) ?? new Set<string>();
326 };
327
328 const clearAssistantMessageIdsForTurn = (sessionId: string, turnId: string) => {
329 turnMessageIdsByTurnKey.delete(providerTurnKey(sessionId, turnId));
330 };
331
332 const getLatestAssistantMessageIdForTurn = (sessionId: string, turnId: string) => {
333 return latestMessageIdByTurnKey.get(providerTurnKey(sessionId, turnId));
334 };
335
336 const clearTurnStateForSession = (sessionId: string) => {
337 const prefix = `${sessionId}:`;
338 for (const key of turnMessageIdsByTurnKey.keys()) {
339 if (key.startsWith(prefix)) {
340 turnMessageIdsByTurnKey.delete(key);
341 }
342 }
343 for (const key of latestMessageIdByTurnKey.keys()) {
344 if (key.startsWith(prefix)) {
345 latestMessageIdByTurnKey.delete(key);
346 }
347 }
348 };
349
350 const processEvent = (event: ProviderRuntimeEvent) =>
351 Effect.gen(function* () {
352 const snapshot = yield* orchestrationEngine.getSnapshot();
353 const thread = snapshot.threads.find((entry) => entry.session?.sessionId === event.sessionId);
354 if (!thread) return;
355
356 const now = event.createdAt;
357 if (event.type === "turn.started" || event.type === "turn.completed") {
358 yield* orchestrationEngine.dispatch({
359 type: "thread.session",
360 commandId: crypto.randomUUID(),
361 threadId: thread.id,
362 session: {
363 sessionId: event.sessionId,
364 provider: event.provider,
365 status: event.type === "turn.started" ? "running" : "ready",
366 threadId: thread.id,
367 activeTurnId: event.type === "turn.started" ? event.turnId : null,
368 createdAt: thread.session?.createdAt ?? now,
369 updatedAt: now,
370 lastError:
371 event.type === "turn.completed" && event.status === "failed"
372 ? (event.errorMessage ?? null)
373 : null,
374 },
375 createdAt: now,
376 });
377 }
378
379 if (event.type === "message.delta" && event.delta.length > 0) {
380 const assistantMessageId = `assistant:${event.itemId ?? event.turnId ?? event.sessionId}`;
381 if (event.turnId) {
382 rememberAssistantMessageId(event.sessionId, event.turnId, assistantMessageId);
293383 }
294 if (event.method === "turn/completed") {
295 const assistantMessageId = `assistant:${event.turnId ?? event.itemId ?? event.sessionId}`;
296 await runtime.runPromise(
384 yield* orchestrationEngine.dispatch({
385 type: "message.send",
386 commandId: crypto.randomUUID(),
387 threadId: thread.id,
388 messageId: assistantMessageId,
389 role: "assistant",
390 text: event.delta,
391 streaming: true,
392 createdAt: now,
393 });
394 }
395
396 if (event.type === "turn.completed") {
397 if (event.turnId) {
398 const assistantMessageIds = getAssistantMessageIdsForTurn(event.sessionId, event.turnId);
399 yield* Effect.forEach(assistantMessageIds, (assistantMessageId) =>
297400 orchestrationEngine.dispatch({
298401 type: "message.send",
299402 commandId: crypto.randomUUID(),
......@@ -304,55 +407,85 @@ export function createServer(options: ServerOptions) {
304407 streaming: false,
305408 createdAt: now,
306409 }),
307 );
308 }
309 if (event.method === "checkpoint/captured") {
310 const payload = asObject(event.payload);
311 const turnId = event.turnId ?? asString(payload?.turnId);
312 if (turnId) {
313 await runtime.runPromise(
314 orchestrationEngine.dispatch({
315 type: "thread.turnDiff.complete",
316 commandId: crypto.randomUUID(),
317 threadId: thread.id,
318 turnId,
319 completedAt: now,
320 ...(asString(payload?.status) !== null
321 ? { status: asString(payload?.status) ?? undefined }
322 : {}),
323 files: [],
324 assistantMessageId: `assistant:${turnId}`,
325 ...(asNumber(payload?.turnCount) !== null
326 ? { checkpointTurnCount: asNumber(payload?.turnCount) ?? undefined }
327 : {}),
328 createdAt: now,
329 }),
330 );
331 }
332 }
333 if (event.kind === "error" && event.message) {
334 await runtime.runPromise(
335 orchestrationEngine.dispatch({
336 type: "thread.session",
337 commandId: crypto.randomUUID(),
338 threadId: thread.id,
339 session: {
340 sessionId: event.sessionId,
341 provider: event.provider,
342 status: "error",
343 threadId: thread.id,
344 activeTurnId: event.turnId ?? null,
345 createdAt: thread.session?.createdAt ?? now,
346 updatedAt: now,
347 lastError: event.message,
348 },
349 createdAt: now,
350 }),
351 );
410 ).pipe(Effect.asVoid);
411 clearAssistantMessageIdsForTurn(event.sessionId, event.turnId);
352412 }
353 })().catch(() => undefined);
354 }),
413 }
414
415 if (event.type === "session.exited") {
416 clearTurnStateForSession(event.sessionId);
417 }
418
419 if (event.type === "checkpoint.captured" && event.turnId) {
420 const assistantMessageId =
421 getLatestAssistantMessageIdForTurn(event.sessionId, event.turnId) ??
422 `assistant:${event.turnId}`;
423 yield* orchestrationEngine.dispatch({
424 type: "thread.turnDiff.complete",
425 commandId: crypto.randomUUID(),
426 threadId: thread.id,
427 turnId: event.turnId,
428 completedAt: now,
429 ...(event.status ? { status: event.status } : {}),
430 files: [],
431 assistantMessageId,
432 checkpointTurnCount: event.turnCount,
433 createdAt: now,
434 });
435 }
436
437 if (event.type === "runtime.error") {
438 yield* orchestrationEngine.dispatch({
439 type: "thread.session",
440 commandId: crypto.randomUUID(),
441 threadId: thread.id,
442 session: {
443 sessionId: event.sessionId,
444 provider: event.provider,
445 status: "error",
446 threadId: thread.id,
447 activeTurnId: event.turnId ?? null,
448 createdAt: thread.session?.createdAt ?? now,
449 updatedAt: now,
450 lastError: event.message,
451 },
452 createdAt: now,
453 });
454 }
455
456 const activities = runtimeEventToActivities(event);
457 yield* Effect.forEach(activities, (activity) =>
458 orchestrationEngine.dispatch({
459 type: "thread.activity.append",
460 commandId: crypto.randomUUID(),
461 threadId: thread.id,
462 activity,
463 createdAt: activity.createdAt,
464 }),
465 ).pipe(Effect.asVoid);
466 });
467
468 const subscriptionScope = await runtime.runPromise(Scope.make("sequential"));
469 await runtime.runPromise(
470 Effect.gen(function* () {
471 const providerEventQueue = yield* Queue.unbounded<ProviderRuntimeEvent>();
472
473 yield* Effect.addFinalizer(() => Queue.shutdown(providerEventQueue).pipe(Effect.asVoid));
474
475 yield* Effect.forkScoped(
476 Effect.forever(Queue.take(providerEventQueue).pipe(Effect.flatMap(processEvent))),
477 );
478 yield* Effect.forkScoped(
479 Stream.runForEach(providerService.streamEvents, (event) =>
480 Queue.offer(providerEventQueue, event).pipe(Effect.asVoid),
481 ),
482 );
483 }).pipe(Scope.provide(subscriptionScope)),
355484 );
485
486 unsubscribeProviderEvents = () => {
487 runtime.runFork(Scope.close(subscriptionScope, Exit.void));
488 };
356489 }
357490
358491 const onTerminalEvent = (event: TerminalEvent) => {
......@@ -456,8 +589,6 @@ export function createServer(options: ServerOptions) {
456589
457590 wss.on("connection", (ws) => {
458591 clients.add(ws);
459 const orchestrationEngine = getOrchestrationEngine();
460 const runtime = getEffectRuntime();
461592
462593 // Send welcome message with project info
463594 const segments = cwd.split(/[/\\]/).filter(Boolean);
......@@ -471,25 +602,6 @@ export function createServer(options: ServerOptions) {
471602 logOutgoingPush(welcome, 1);
472603 ws.send(JSON.stringify(welcome));
473604
474 void runtime
475 .runPromise(orchestrationEngine.getSnapshot())
476 .then((snapshot) => {
477 if (ws.readyState !== ws.OPEN) {
478 return;
479 }
480 const snapshotPush: WsPush = {
481 type: "push",
482 channel: ORCHESTRATION_WS_CHANNELS.readModel,
483 data: {
484 sequence: snapshot.sequence,
485 snapshot,
486 } satisfies OrchestrationReadModelPush,
487 };
488 logOutgoingPush(snapshotPush, 1);
489 ws.send(JSON.stringify(snapshotPush));
490 })
491 .catch(() => undefined);
492
493605 ws.on("message", (raw) => {
494606 void handleMessage(ws, raw);
495607 });
......@@ -745,6 +857,11 @@ export function createServer(options: ServerOptions) {
745857 return effectRuntime.runPromise(orchestrationEngine.getSnapshot());
746858
747859 case ORCHESTRATION_WS_METHODS.dispatchCommand:
860 orchestrationCommandLogger?.write({
861 observedAt: new Date().toISOString(),
862 requestId: request.id,
863 command: request.params,
864 });
748865 return effectRuntime.runPromise(orchestrationEngine.dispatchUnknown(request.params));
749866
750867 case ORCHESTRATION_WS_METHODS.replayEvents:
......@@ -769,12 +886,23 @@ export function createServer(options: ServerOptions) {
769886
770887 async function createEffectRuntime() {
771888 const dbPath = path.join(resolvedStateDir, "orchestration.sqlite");
889 const providerLogsDir = path.join(resolvedStateDir, "logs", "providers");
890 const orchestrationLogsDir = path.join(resolvedStateDir, "logs", "orchestration");
891 orchestrationDomainEventLogger = makeEventNdjsonLogger(
892 path.join(orchestrationLogsDir, "orchestration-domain.ndjson"),
893 );
894 orchestrationCommandLogger = makeEventNdjsonLogger(
895 path.join(orchestrationLogsDir, "orchestration-command.ndjson"),
896 );
772897 const persistenceLayer = customPersistenceLayer ?? makeSqlitePersistenceLive(dbPath);
773898 const orchestrationLayer = Layer.provide(
774899 OrchestrationEngineLive,
775900 OrchestrationEventRepositoryLive,
776901 );
777 const adapterRegistryLayer = ProviderAdapterRegistryLive.pipe(Layer.provide(CodexAdapterLive));
902 const codexAdapterLayer = makeCodexAdapterLive({
903 nativeEventLogPath: path.join(providerLogsDir, "provider-native.ndjson"),
904 });
905 const adapterRegistryLayer = ProviderAdapterRegistryLive.pipe(Layer.provide(codexAdapterLayer));
778906 const checkpointStoreLayer = CheckpointStoreLive.pipe(Layer.provide(NodeServices.layer));
779907 const checkpointDependenciesLayer = Layer.mergeAll(
780908 checkpointStoreLayer,
......@@ -785,7 +913,9 @@ export function createServer(options: ServerOptions) {
785913 const checkpointServiceLayer = CheckpointServiceLive.pipe(
786914 Layer.provideMerge(checkpointDependenciesLayer),
787915 );
788 const providerLayer = ProviderServiceLive.pipe(Layer.provideMerge(checkpointServiceLayer));
916 const providerLayer = makeProviderServiceLive({
917 canonicalEventLogPath: path.join(providerLogsDir, "provider-canonical.ndjson"),
918 }).pipe(Layer.provideMerge(checkpointServiceLayer));
789919
790920 const layer = Layer.mergeAll(orchestrationLayer, ProjectRepositoryLive, providerLayer).pipe(
791921 Layer.provide(persistenceLayer),
......@@ -814,12 +944,14 @@ export function createServer(options: ServerOptions) {
814944 }
815945
816946 async function disposeEffectRuntime() {
817 unsubscribeReadModel();
818947 unsubscribeDomainEvents();
819948 unsubscribeProviderEvents();
820 unsubscribeReadModel = noop;
821949 unsubscribeDomainEvents = noop;
822950 unsubscribeProviderEvents = noop;
951 orchestrationDomainEventLogger?.close();
952 orchestrationDomainEventLogger = undefined;
953 orchestrationCommandLogger?.close();
954 orchestrationCommandLogger = undefined;
823955
824956 const runtime = effectRuntime;
825957 const liveProviderService = providerService;
apps/server/tsconfig.json+1-1
......@@ -7,5 +7,5 @@
77 "noEmit": true,
88 "allowImportingTsExtensions": true
99 },
10 "include": ["src", "tsdown.config.ts"]
10 "include": ["src", "integration", "tsdown.config.ts", "scripts"]
1111}
apps/web/src/components/ChatView.tsx+20-5
......@@ -432,20 +432,35 @@ export default function ChatView({ threadId }: ChatViewProps) {
432432 );
433433 const phase = derivePhase(activeThread?.session ?? null);
434434 const isWorking = phase === "running" || isSending || isConnecting || isRevertingCheckpoint;
435 const activeTurnId = activeThread?.session?.activeTurnId ?? undefined;
435436 const nowIso = new Date(nowTick).toISOString();
436 const workLogEntries = useMemo(() => deriveWorkLogEntries([], undefined), []);
437 const threadActivities = activeThread?.activities ?? [];
438 const workLogEntries = useMemo(
439 () => deriveWorkLogEntries(threadActivities, undefined),
440 [threadActivities],
441 );
437442 const latestTurnWorkEntries = useMemo(
438 () => deriveWorkLogEntries([], activeThread?.latestTurnId),
439 [activeThread?.latestTurnId],
443 () => deriveWorkLogEntries(threadActivities, activeTurnId ?? activeThread?.latestTurnId),
444 [activeThread?.latestTurnId, activeTurnId, threadActivities],
445 );
446 const pendingApprovals = useMemo(
447 () => derivePendingApprovals(threadActivities),
448 [threadActivities],
440449 );
441 const pendingApprovals = useMemo(() => derivePendingApprovals([]), []);
442 const assistantCompletionByItemId = useMemo(() => new Map<string, string>(), []);
443450 const timelineEntries = useMemo(
444451 () => deriveTimelineEntries(activeThread?.messages ?? [], workLogEntries),
445452 [activeThread?.messages, workLogEntries],
446453 );
447454 const { turnDiffSummaries, inferredCheckpointTurnCountByTurnId } =
448455 useTurnDiffSummaries(activeThread);
456 const assistantCompletionByItemId = useMemo(() => {
457 const completionByMessageId = new Map<string, string>();
458 for (const summary of turnDiffSummaries) {
459 if (!summary.assistantMessageId) continue;
460 completionByMessageId.set(summary.assistantMessageId, summary.completedAt);
461 }
462 return completionByMessageId;
463 }, [turnDiffSummaries]);
449464 const turnDiffSummaryByAssistantMessageId = useMemo(() => {
450465 const byMessageId = new Map<string, TurnDiffSummary>();
451466 for (const summary of turnDiffSummaries) {
apps/web/src/routes/__root.tsx+48-6
......@@ -126,17 +126,59 @@ function EventRouter() {
126126
127127 useEffect(() => {
128128 if (!api) return;
129 void Promise.all([api.projects.list(), api.orchestration.getSnapshot()])
130 .then(([projects, snapshot]) => {
129 let disposed = false;
130 let latestSequence = 0;
131 let syncing = false;
132 let pending = false;
133
134 const flushSnapshotSync = async (): Promise<void> => {
135 const snapshot = await api.orchestration.getSnapshot();
136 if (disposed) return;
137 latestSequence = Math.max(latestSequence, snapshot.sequence);
138 dispatch({ type: "SYNC_SERVER_READ_MODEL", readModel: snapshot });
139 if (pending) {
140 pending = false;
141 await flushSnapshotSync();
142 }
143 };
144
145 const syncSnapshot = async () => {
146 if (syncing) {
147 pending = true;
148 return;
149 }
150 syncing = true;
151 try {
152 pending = false;
153 await flushSnapshotSync();
154 } catch {
155 // Keep prior state and wait for next domain event to trigger a resync.
156 } finally {
157 syncing = false;
158 }
159 };
160
161 void Promise.all([api.projects.list(), syncSnapshot()])
162 .then(([projects]) => {
163 if (disposed) return;
131164 dispatch({ type: "SYNC_PROJECTS", projects });
132 dispatch({ type: "SYNC_SERVER_READ_MODEL", readModel: snapshot });
133165 })
134166 .catch(() => undefined);
135 const unsubReadModel = api.orchestration.onReadModel((snapshot) => {
136 dispatch({ type: "SYNC_SERVER_READ_MODEL", readModel: snapshot });
167
168 const unsubDomainEvent = api.orchestration.onDomainEvent((event) => {
169 if (event.sequence <= latestSequence) {
170 return;
171 }
172 latestSequence = event.sequence;
173 void syncSnapshot();
174 });
175 const unsubWelcome = onServerWelcome(() => {
176 void syncSnapshot();
137177 });
138178 return () => {
139 unsubReadModel();
179 disposed = true;
180 unsubDomainEvent();
181 unsubWelcome();
140182 };
141183 }, [api, dispatch]);
142184
apps/web/src/session-logic.test.ts created+90
......@@ -0,0 +1,90 @@
1import type { OrchestrationThreadActivity } from "@t3tools/contracts";
2import { describe, expect, it } from "vitest";
3
4import { derivePendingApprovals, deriveWorkLogEntries } from "./session-logic";
5
6function makeActivity(
7 overrides: Partial<OrchestrationThreadActivity>,
8): OrchestrationThreadActivity {
9 return {
10 id: overrides.id ?? crypto.randomUUID(),
11 createdAt: overrides.createdAt ?? "2026-02-23T00:00:00.000Z",
12 label: overrides.label ?? "Tool call",
13 tone: overrides.tone ?? "tool",
14 ...(overrides.detail !== undefined ? { detail: overrides.detail } : {}),
15 ...(overrides.turnId !== undefined ? { turnId: overrides.turnId } : {}),
16 ...(overrides.requestId !== undefined ? { requestId: overrides.requestId } : {}),
17 ...(overrides.requestKind !== undefined ? { requestKind: overrides.requestKind } : {}),
18 };
19}
20
21describe("derivePendingApprovals", () => {
22 it("tracks open approvals and removes resolved ones", () => {
23 const activities: OrchestrationThreadActivity[] = [
24 makeActivity({
25 id: "approval-open",
26 createdAt: "2026-02-23T00:00:01.000Z",
27 label: "Command approval requested",
28 detail: "bun run lint",
29 tone: "tool",
30 requestId: "req-1",
31 requestKind: "command",
32 }),
33 makeActivity({
34 id: "approval-close",
35 createdAt: "2026-02-23T00:00:02.000Z",
36 label: "Approval resolved",
37 tone: "info",
38 requestId: "req-2",
39 }),
40 makeActivity({
41 id: "approval-closed-request",
42 createdAt: "2026-02-23T00:00:01.500Z",
43 label: "File-change approval requested",
44 tone: "tool",
45 requestId: "req-2",
46 requestKind: "file-change",
47 }),
48 ];
49
50 expect(derivePendingApprovals(activities)).toEqual([
51 {
52 requestId: "req-1",
53 requestKind: "command",
54 createdAt: "2026-02-23T00:00:01.000Z",
55 detail: "bun run lint",
56 },
57 ]);
58 });
59});
60
61describe("deriveWorkLogEntries", () => {
62 it("returns entries in chronological order", () => {
63 const activities: OrchestrationThreadActivity[] = [
64 makeActivity({
65 id: "tool-complete",
66 createdAt: "2026-02-23T00:00:03.000Z",
67 label: "Tool call complete",
68 }),
69 makeActivity({
70 id: "tool-start",
71 createdAt: "2026-02-23T00:00:02.000Z",
72 label: "Tool call",
73 }),
74 ];
75
76 const entries = deriveWorkLogEntries(activities, undefined);
77 expect(entries.map((entry) => entry.id)).toEqual(["tool-start", "tool-complete"]);
78 });
79
80 it("filters by turn id when provided", () => {
81 const activities: OrchestrationThreadActivity[] = [
82 makeActivity({ id: "turn-1", turnId: "turn-1", label: "Tool call" }),
83 makeActivity({ id: "turn-2", turnId: "turn-2", label: "Tool call complete" }),
84 makeActivity({ id: "no-turn", label: "Checkpoint captured", tone: "info" }),
85 ];
86
87 const entries = deriveWorkLogEntries(activities, "turn-2");
88 expect(entries.map((entry) => entry.id)).toEqual(["turn-2"]);
89 });
90});
apps/web/src/session-logic.ts+54-9
......@@ -1,4 +1,8 @@
1import type { ProviderKind, ProviderSession } from "@t3tools/contracts";
1import type {
2 OrchestrationThreadActivity,
3 ProviderKind,
4 ProviderSession,
5} from "@t3tools/contracts";
26
37import type { ChatMessage, SessionPhase, TurnDiffFileChange, TurnDiffSummary } from "./types";
48
......@@ -70,15 +74,56 @@ export function formatElapsed(startIso: string, endIso: string | undefined): str
7074 return formatDuration(endedAt - startedAt);
7175}
7276
73export function derivePendingApprovals(_events: unknown[]): PendingApproval[] {
74 return [];
77export function derivePendingApprovals(
78 activities: ReadonlyArray<OrchestrationThreadActivity>,
79): PendingApproval[] {
80 const openByRequestId = new Map<string, PendingApproval>();
81 const ordered = [...activities].toSorted((left, right) =>
82 left.createdAt.localeCompare(right.createdAt),
83 );
84
85 for (const activity of ordered) {
86 if (activity.requestId && activity.requestKind) {
87 openByRequestId.set(activity.requestId, {
88 requestId: activity.requestId,
89 requestKind: activity.requestKind,
90 createdAt: activity.createdAt,
91 ...(activity.detail ? { detail: activity.detail } : {}),
92 });
93 continue;
94 }
95
96 if (activity.requestId && activity.label === "Approval resolved") {
97 openByRequestId.delete(activity.requestId);
98 }
99 }
100
101 return [...openByRequestId.values()].toSorted((left, right) =>
102 left.createdAt.localeCompare(right.createdAt),
103 );
75104}
76105
77106export function deriveWorkLogEntries(
78 _events: unknown[],
79 _latestTurnId: string | undefined,
107 activities: ReadonlyArray<OrchestrationThreadActivity>,
108 latestTurnId: string | undefined,
80109): WorkLogEntry[] {
81 return [];
110 const ordered = [...activities].toSorted((left, right) =>
111 left.createdAt.localeCompare(right.createdAt),
112 );
113 return ordered
114 .filter((activity) => (latestTurnId ? activity.turnId === latestTurnId : true))
115 .map((activity) => {
116 const entry: WorkLogEntry = {
117 id: activity.id,
118 createdAt: activity.createdAt,
119 label: activity.label,
120 tone: activity.tone,
121 };
122 if (activity.detail) {
123 entry.detail = activity.detail;
124 }
125 return entry;
126 });
82127}
83128
84129export function deriveTimelineEntries(
......@@ -97,7 +142,7 @@ export function deriveTimelineEntries(
97142 createdAt: entry.createdAt,
98143 entry,
99144 }));
100 return [...messageRows, ...workRows].sort((a, b) => a.createdAt.localeCompare(b.createdAt));
145 return [...messageRows, ...workRows].toSorted((a, b) => a.createdAt.localeCompare(b.createdAt));
101146}
102147
103148export function deriveTurnDiffFilesFromUnifiedDiff(diff: string): TurnDiffFileChange[] {
......@@ -116,7 +161,7 @@ export function deriveTurnDiffFilesFromUnifiedDiff(diff: string): TurnDiffFileCh
116161export function inferCheckpointTurnCountByTurnId(
117162 summaries: TurnDiffSummary[],
118163): Record<string, number> {
119 const sorted = [...summaries].sort((a, b) => a.completedAt.localeCompare(b.completedAt));
164 const sorted = [...summaries].toSorted((a, b) => a.completedAt.localeCompare(b.completedAt));
120165 const result: Record<string, number> = {};
121166 for (let index = 0; index < sorted.length; index += 1) {
122167 const summary = sorted[index];
......@@ -126,7 +171,7 @@ export function inferCheckpointTurnCountByTurnId(
126171 return result;
127172}
128173
129export function deriveTurnDiffSummaries(_events: unknown[]): TurnDiffSummary[] {
174export function deriveTurnDiffSummaries(_events: ReadonlyArray<OrchestrationThreadActivity>): TurnDiffSummary[] {
130175 return [];
131176}
132177
apps/web/src/store.ts+1
......@@ -398,6 +398,7 @@ export function reducer(state: AppState, action: Action): AppState {
398398 ...summary,
399399 files: summary.files.map((file) => ({ ...file })),
400400 })),
401 activities: thread.activities.map((activity) => ({ ...activity })),
401402 });
402403 });
403404 return {
apps/web/src/threadFactory.ts+1
......@@ -33,6 +33,7 @@ export function createThread(projectId: string, options: CreateThreadOptions = {
3333 session: null,
3434 messages: [],
3535 turnDiffSummaries: [],
36 activities: [],
3637 error: null,
3738 createdAt,
3839 branch: options.branch ?? null,
apps/web/src/types.ts+2
......@@ -1,4 +1,5 @@
11import type {
2 OrchestrationThreadActivity,
23 ProjectScript as ContractProjectScript,
34 ProviderSession,
45} from "@t3tools/contracts";
......@@ -86,4 +87,5 @@ export interface Thread {
8687 branch: string | null;
8788 worktreePath: string | null;
8889 turnDiffSummaries: TurnDiffSummary[];
90 activities: OrchestrationThreadActivity[];
8991}
apps/web/src/worktreeCleanup.test.ts+1
......@@ -25,6 +25,7 @@ function makeThread(overrides: Partial<Thread> = {}): Thread {
2525 session: null,
2626 messages: [],
2727 turnDiffSummaries: [],
28 activities: [],
2829 error: null,
2930 createdAt: "2026-02-13T00:00:00.000Z",
3031 branch: null,
apps/web/src/wsNativeApi.ts-10
......@@ -4,8 +4,6 @@ import {
44 type NativeApi,
55 WS_CHANNELS,
66 WS_METHODS,
7 type OrchestrationEvent,
8 type OrchestrationReadModel,
97 type WsWelcomePayload,
108} from "@t3tools/contracts";
119
......@@ -161,14 +159,6 @@ export function createWsNativeApi(): NativeApi {
161159 transport.request(ORCHESTRATION_WS_METHODS.dispatchCommand, command),
162160 replayEvents: (fromSequenceExclusive) =>
163161 transport.request(ORCHESTRATION_WS_METHODS.replayEvents, { fromSequenceExclusive }),
164 onReadModel: (callback) =>
165 transport.subscribe(
166 ORCHESTRATION_WS_CHANNELS.readModel,
167 (data: unknown) => {
168 const payload = data as { snapshot: OrchestrationReadModel };
169 callback(payload.snapshot);
170 },
171 ),
172162 onDomainEvent: (callback) =>
173163 transport.subscribe(
174164 ORCHESTRATION_WS_CHANNELS.domainEvent,
packages/contracts/src/index.ts+1
......@@ -3,6 +3,7 @@ export * from "./ipc";
33export * from "./agent";
44export * from "./terminal";
55export * from "./provider";
6export * from "./providerRuntime";
67export * from "./model";
78export * from "./ws";
89export * from "./keybindings";
packages/contracts/src/ipc.ts-1
......@@ -146,7 +146,6 @@ export interface NativeApi {
146146 getSnapshot: () => Promise<OrchestrationReadModel>;
147147 dispatchCommand: (command: OrchestrationCommand) => Promise<{ sequence: number }>;
148148 replayEvents: (fromSequenceExclusive: number) => Promise<OrchestrationEvent[]>;
149 onReadModel: (callback: (snapshot: OrchestrationReadModel) => void) => () => void;
150149 onDomainEvent: (callback: (event: OrchestrationEvent) => void) => () => void;
151150 };
152151}
packages/contracts/src/orchestration.ts+34-8
......@@ -7,7 +7,6 @@ export const ORCHESTRATION_WS_METHODS = {
77} as const;
88
99export const ORCHESTRATION_WS_CHANNELS = {
10 readModel: "orchestration.readModel",
1110 domainEvent: "orchestration.domainEvent",
1211} as const;
1312
......@@ -59,6 +58,30 @@ export type OrchestrationTurnDiffSummary = Schema.Schema.Type<
5958 typeof OrchestrationTurnDiffSummarySchema
6059>;
6160
61export const OrchestrationThreadActivityToneSchema = Schema.Literals([
62 "thinking",
63 "tool",
64 "info",
65 "error",
66]);
67
68export type OrchestrationThreadActivityTone = Schema.Schema.Type<
69 typeof OrchestrationThreadActivityToneSchema
70>;
71
72export const OrchestrationThreadActivitySchema = Schema.Struct({
73 id: IdSchema,
74 createdAt: IsoDateTimeSchema,
75 label: Schema.String,
76 detail: Schema.optional(Schema.String),
77 tone: OrchestrationThreadActivityToneSchema,
78 turnId: Schema.optional(Schema.String),
79 requestId: Schema.optional(Schema.String),
80 requestKind: Schema.optional(Schema.Literals(["command", "file-change"])),
81});
82
83export type OrchestrationThreadActivity = Schema.Schema.Type<typeof OrchestrationThreadActivitySchema>;
84
6285export const OrchestrationThreadSchema = Schema.Struct({
6386 id: IdSchema,
6487 projectId: IdSchema,
......@@ -75,6 +98,7 @@ export const OrchestrationThreadSchema = Schema.Struct({
7598 messages: Schema.Array(OrchestrationMessageSchema),
7699 session: Schema.NullOr(OrchestrationSessionSchema),
77100 turnDiffSummaries: Schema.Array(OrchestrationTurnDiffSummarySchema),
101 activities: Schema.Array(OrchestrationThreadActivitySchema),
78102 error: Schema.NullOr(Schema.String),
79103});
80104
......@@ -182,6 +206,14 @@ export const RevertThreadCommandSchema = Schema.Struct({
182206 createdAt: IsoDateTimeSchema,
183207});
184208
209export const AppendThreadActivityCommandSchema = Schema.Struct({
210 type: Schema.Literal("thread.activity.append"),
211 commandId: IdSchema,
212 threadId: IdSchema,
213 activity: OrchestrationThreadActivitySchema,
214 createdAt: IsoDateTimeSchema,
215});
216
185217export const OrchestrationCommandSchema = Schema.Union([
186218 CreateThreadCommandSchema,
187219 DeleteThreadCommandSchema,
......@@ -190,6 +222,7 @@ export const OrchestrationCommandSchema = Schema.Union([
190222 SetThreadSessionSchema,
191223 UpsertGitReadModelCommandSchema,
192224 CompleteThreadTurnDiffCommandSchema,
225 AppendThreadActivityCommandSchema,
193226 RevertThreadCommandSchema,]
194227);
195228
......@@ -207,10 +240,3 @@ export const OrchestrationEventSchema = Schema.Struct({
207240});
208241
209242export type OrchestrationEvent = Schema.Schema.Type<typeof OrchestrationEventSchema>;
210
211export const OrchestrationReadModelPushSchema = Schema.Struct({
212 sequence: Schema.Number,
213 snapshot: OrchestrationReadModelSchema,
214});
215
216export type OrchestrationReadModelPush = Schema.Schema.Type<typeof OrchestrationReadModelPushSchema>;
packages/contracts/src/providerRuntime.ts created+217
......@@ -0,0 +1,217 @@
1import { Schema } from "effect";
2
3const IsoDateTimeSchema = Schema.String;
4const IdSchema = Schema.String;
5
6export const ProviderRuntimeProviderSchema = Schema.Literals(["codex", "claudeCode"]);
7export type ProviderRuntimeProvider = Schema.Schema.Type<typeof ProviderRuntimeProviderSchema>;
8
9export const ProviderRuntimeApprovalKindSchema = Schema.Literals(["command", "file-change"]);
10export type ProviderRuntimeApprovalKind = Schema.Schema.Type<
11 typeof ProviderRuntimeApprovalKindSchema
12>;
13
14export const ProviderRuntimeToolKindSchema = Schema.Literals(["command", "file-change", "other"]);
15export type ProviderRuntimeToolKind = Schema.Schema.Type<typeof ProviderRuntimeToolKindSchema>;
16
17export const ProviderRuntimeTurnStatusSchema = Schema.Literals([
18 "completed",
19 "failed",
20 "cancelled",
21 "interrupted",
22]);
23export type ProviderRuntimeTurnStatus = Schema.Schema.Type<typeof ProviderRuntimeTurnStatusSchema>;
24
25export const ProviderRuntimeSessionStartedEventSchema = Schema.Struct({
26 type: Schema.Literal("session.started"),
27 eventId: IdSchema,
28 provider: ProviderRuntimeProviderSchema,
29 sessionId: IdSchema,
30 createdAt: IsoDateTimeSchema,
31 threadId: Schema.optional(Schema.String),
32 message: Schema.optional(Schema.String),
33});
34
35export const ProviderRuntimeSessionExitedEventSchema = Schema.Struct({
36 type: Schema.Literal("session.exited"),
37 eventId: IdSchema,
38 provider: ProviderRuntimeProviderSchema,
39 sessionId: IdSchema,
40 createdAt: IsoDateTimeSchema,
41 threadId: Schema.optional(Schema.String),
42 message: Schema.optional(Schema.String),
43});
44
45export const ProviderRuntimeThreadStartedEventSchema = Schema.Struct({
46 type: Schema.Literal("thread.started"),
47 eventId: IdSchema,
48 provider: ProviderRuntimeProviderSchema,
49 sessionId: IdSchema,
50 createdAt: IsoDateTimeSchema,
51 threadId: IdSchema,
52});
53
54export const ProviderRuntimeTurnStartedEventSchema = Schema.Struct({
55 type: Schema.Literal("turn.started"),
56 eventId: IdSchema,
57 provider: ProviderRuntimeProviderSchema,
58 sessionId: IdSchema,
59 createdAt: IsoDateTimeSchema,
60 threadId: Schema.optional(Schema.String),
61 turnId: IdSchema,
62});
63
64export const ProviderRuntimeTurnCompletedEventSchema = Schema.Struct({
65 type: Schema.Literal("turn.completed"),
66 eventId: IdSchema,
67 provider: ProviderRuntimeProviderSchema,
68 sessionId: IdSchema,
69 createdAt: IsoDateTimeSchema,
70 threadId: Schema.optional(Schema.String),
71 turnId: Schema.optional(Schema.String),
72 status: Schema.optional(ProviderRuntimeTurnStatusSchema),
73 errorMessage: Schema.optional(Schema.String),
74});
75
76export const ProviderRuntimeMessageDeltaEventSchema = Schema.Struct({
77 type: Schema.Literal("message.delta"),
78 eventId: IdSchema,
79 provider: ProviderRuntimeProviderSchema,
80 sessionId: IdSchema,
81 createdAt: IsoDateTimeSchema,
82 threadId: Schema.optional(Schema.String),
83 turnId: Schema.optional(Schema.String),
84 itemId: Schema.optional(Schema.String),
85 delta: Schema.String,
86});
87
88export const ProviderRuntimeToolStartedEventSchema = Schema.Struct({
89 type: Schema.Literal("tool.started"),
90 eventId: IdSchema,
91 provider: ProviderRuntimeProviderSchema,
92 sessionId: IdSchema,
93 createdAt: IsoDateTimeSchema,
94 threadId: Schema.optional(Schema.String),
95 turnId: Schema.optional(Schema.String),
96 itemId: Schema.optional(Schema.String),
97 toolKind: ProviderRuntimeToolKindSchema,
98 title: Schema.String,
99 detail: Schema.optional(Schema.String),
100});
101
102export const ProviderRuntimeToolCompletedEventSchema = Schema.Struct({
103 type: Schema.Literal("tool.completed"),
104 eventId: IdSchema,
105 provider: ProviderRuntimeProviderSchema,
106 sessionId: IdSchema,
107 createdAt: IsoDateTimeSchema,
108 threadId: Schema.optional(Schema.String),
109 turnId: Schema.optional(Schema.String),
110 itemId: Schema.optional(Schema.String),
111 toolKind: ProviderRuntimeToolKindSchema,
112 title: Schema.String,
113 detail: Schema.optional(Schema.String),
114});
115
116export const ProviderRuntimeApprovalRequestedEventSchema = Schema.Struct({
117 type: Schema.Literal("approval.requested"),
118 eventId: IdSchema,
119 provider: ProviderRuntimeProviderSchema,
120 sessionId: IdSchema,
121 createdAt: IsoDateTimeSchema,
122 threadId: Schema.optional(Schema.String),
123 turnId: Schema.optional(Schema.String),
124 itemId: Schema.optional(Schema.String),
125 requestId: IdSchema,
126 requestKind: ProviderRuntimeApprovalKindSchema,
127 detail: Schema.optional(Schema.String),
128});
129
130export const ProviderRuntimeApprovalResolvedEventSchema = Schema.Struct({
131 type: Schema.Literal("approval.resolved"),
132 eventId: IdSchema,
133 provider: ProviderRuntimeProviderSchema,
134 sessionId: IdSchema,
135 createdAt: IsoDateTimeSchema,
136 threadId: Schema.optional(Schema.String),
137 turnId: Schema.optional(Schema.String),
138 itemId: Schema.optional(Schema.String),
139 requestId: IdSchema,
140 requestKind: Schema.optional(ProviderRuntimeApprovalKindSchema),
141 decision: Schema.optional(Schema.String),
142});
143
144export const ProviderRuntimeCheckpointCapturedEventSchema = Schema.Struct({
145 type: Schema.Literal("checkpoint.captured"),
146 eventId: IdSchema,
147 provider: ProviderRuntimeProviderSchema,
148 sessionId: IdSchema,
149 createdAt: IsoDateTimeSchema,
150 threadId: IdSchema,
151 turnId: Schema.optional(Schema.String),
152 turnCount: Schema.Number,
153 status: Schema.optional(ProviderRuntimeTurnStatusSchema),
154});
155
156export const ProviderRuntimeErrorEventSchema = Schema.Struct({
157 type: Schema.Literal("runtime.error"),
158 eventId: IdSchema,
159 provider: ProviderRuntimeProviderSchema,
160 sessionId: IdSchema,
161 createdAt: IsoDateTimeSchema,
162 threadId: Schema.optional(Schema.String),
163 turnId: Schema.optional(Schema.String),
164 itemId: Schema.optional(Schema.String),
165 message: Schema.String,
166});
167
168export const ProviderRuntimeEventSchema = Schema.Union([
169 ProviderRuntimeSessionStartedEventSchema,
170 ProviderRuntimeSessionExitedEventSchema,
171 ProviderRuntimeThreadStartedEventSchema,
172 ProviderRuntimeTurnStartedEventSchema,
173 ProviderRuntimeTurnCompletedEventSchema,
174 ProviderRuntimeMessageDeltaEventSchema,
175 ProviderRuntimeToolStartedEventSchema,
176 ProviderRuntimeToolCompletedEventSchema,
177 ProviderRuntimeApprovalRequestedEventSchema,
178 ProviderRuntimeApprovalResolvedEventSchema,
179 ProviderRuntimeCheckpointCapturedEventSchema,
180 ProviderRuntimeErrorEventSchema,
181]);
182
183export type ProviderRuntimeSessionStartedEvent = Schema.Schema.Type<
184 typeof ProviderRuntimeSessionStartedEventSchema
185>;
186export type ProviderRuntimeSessionExitedEvent = Schema.Schema.Type<
187 typeof ProviderRuntimeSessionExitedEventSchema
188>;
189export type ProviderRuntimeThreadStartedEvent = Schema.Schema.Type<
190 typeof ProviderRuntimeThreadStartedEventSchema
191>;
192export type ProviderRuntimeTurnStartedEvent = Schema.Schema.Type<
193 typeof ProviderRuntimeTurnStartedEventSchema
194>;
195export type ProviderRuntimeTurnCompletedEvent = Schema.Schema.Type<
196 typeof ProviderRuntimeTurnCompletedEventSchema
197>;
198export type ProviderRuntimeMessageDeltaEvent = Schema.Schema.Type<
199 typeof ProviderRuntimeMessageDeltaEventSchema
200>;
201export type ProviderRuntimeToolStartedEvent = Schema.Schema.Type<
202 typeof ProviderRuntimeToolStartedEventSchema
203>;
204export type ProviderRuntimeToolCompletedEvent = Schema.Schema.Type<
205 typeof ProviderRuntimeToolCompletedEventSchema
206>;
207export type ProviderRuntimeApprovalRequestedEvent = Schema.Schema.Type<
208 typeof ProviderRuntimeApprovalRequestedEventSchema
209>;
210export type ProviderRuntimeApprovalResolvedEvent = Schema.Schema.Type<
211 typeof ProviderRuntimeApprovalResolvedEventSchema
212>;
213export type ProviderRuntimeCheckpointCapturedEvent = Schema.Schema.Type<
214 typeof ProviderRuntimeCheckpointCapturedEventSchema
215>;
216export type ProviderRuntimeErrorEvent = Schema.Schema.Type<typeof ProviderRuntimeErrorEventSchema>;
217export type ProviderRuntimeEvent = Schema.Schema.Type<typeof ProviderRuntimeEventSchema>;