authorgravatar for julius0216@outlook.comJulius Marminge <julius0216@outlook.com> 2026-02-23 11:14:56-08:00
committergravatar for jmarminge@gmail.comJulius Marminge <jmarminge@gmail.com> 2026-02-26 13:20:36-08:00
logd4b6536a502e34e473cfa4f5d6b54be0f483cc1e
tree343668a922f7a572ee08de292f32f615f820850b
parent7934639aa1379f745af7db7c05c964bf6d7a7cc8

runtime ingestion

6 files changed, 813 insertions(+), 400 deletions(-)

apps/server/integration/TestProviderAdapter.integration.ts+8-3
...@@ -131,7 +131,7 @@ export const makeTestProviderAdapterHarness = Effect.gen(function* () {...@@ -131,7 +131,7 @@ export const makeTestProviderAdapterHarness = Effect.gen(function* () {
131 }131 }
132132
133 const assistantDeltas: string[] = [];133 const assistantDeltas: string[] = [];
134 let hasCompletedEvent = false;134 const deferredTurnCompletedEvents: ProviderRuntimeEvent[] = [];
135 for (const fixtureEvent of response.events) {135 for (const fixtureEvent of response.events) {
136 const rawEvent: Record<string, unknown> = {136 const rawEvent: Record<string, unknown> = {
137 ...(fixtureEvent as Record<string, unknown>),137 ...(fixtureEvent as Record<string, unknown>),
...@@ -152,7 +152,8 @@ export const makeTestProviderAdapterHarness = Effect.gen(function* () {...@@ -152,7 +152,8 @@ export const makeTestProviderAdapterHarness = Effect.gen(function* () {
152 assistantDeltas.push(runtimeEvent.delta);152 assistantDeltas.push(runtimeEvent.delta);
153 }153 }
154 if (runtimeEvent.type === "turn.completed") {154 if (runtimeEvent.type === "turn.completed") {
155 hasCompletedEvent = true;155 deferredTurnCompletedEvents.push(runtimeEvent);
156 continue;
156 }157 }
157158
158 yield* emit(runtimeEvent);159 yield* emit(runtimeEvent);
...@@ -182,7 +183,7 @@ export const makeTestProviderAdapterHarness = Effect.gen(function* () {...@@ -182,7 +183,7 @@ export const makeTestProviderAdapterHarness = Effect.gen(function* () {
182 turns: [...state.snapshot.turns, nextTurn],183 turns: [...state.snapshot.turns, nextTurn],
183 };184 };
184185
185 if (!hasCompletedEvent) {186 if (deferredTurnCompletedEvents.length === 0) {
186 yield* emit({187 yield* emit({
187 type: "turn.completed",188 type: "turn.completed",
188 eventId: randomUUID(),189 eventId: randomUUID(),
...@@ -193,6 +194,10 @@ export const makeTestProviderAdapterHarness = Effect.gen(function* () {...@@ -193,6 +194,10 @@ export const makeTestProviderAdapterHarness = Effect.gen(function* () {
193 turnId,194 turnId,
194 status: "completed",195 status: "completed",
195 });196 });
197 } else {
198 for (const completedEvent of deferredTurnCompletedEvents) {
199 yield* emit(completedEvent);
200 }
196 }201 }
197202
198 return {203 return {
apps/server/src/orchestration/Layers/ProviderRuntimeIngestion.test.ts created+264
...@@ -0,0 +1,264 @@
1import type { ProviderRuntimeEvent } from "@t3tools/contracts";
2import { Effect, Exit, Layer, ManagedRuntime, PubSub, Scope, Stream } from "effect";
3import { afterEach, describe, expect, it, vi } from "vitest";
4
5import { OrchestrationEventStoreLive } from "../../persistence/Layers/OrchestrationEventStore.ts";
6import { SqlitePersistenceMemory } from "../../persistence/Layers/Sqlite.ts";
7import {
8 ProviderService,
9 type ProviderServiceShape,
10} from "../../provider/Services/ProviderService.ts";
11import { OrchestrationEngineLive } from "./OrchestrationEngine.ts";
12import { ProviderRuntimeIngestionLive } from "./ProviderRuntimeIngestion.ts";
13import {
14 OrchestrationEngineService,
15 type OrchestrationEngineShape,
16} from "../Services/OrchestrationEngine.ts";
17import { ProviderRuntimeIngestionService } from "../Services/ProviderRuntimeIngestion.ts";
18
19function createProviderServiceHarness() {
20 const runtimeEventPubSub = Effect.runSync(PubSub.unbounded<ProviderRuntimeEvent>());
21 const getCheckpointDiff = vi.fn(
22 (input: { sessionId: string; fromTurnCount: number; toTurnCount: number }) =>
23 Effect.succeed({
24 threadId: "thread-1",
25 fromTurnCount: input.fromTurnCount,
26 toTurnCount: input.toTurnCount,
27 diff: "",
28 }),
29 );
30
31 const unsupported = () => Effect.die(new Error("Unsupported provider call in test")) as never;
32 const service: ProviderServiceShape = {
33 startSession: () => unsupported(),
34 sendTurn: () => unsupported(),
35 interruptTurn: () => unsupported(),
36 respondToRequest: () => unsupported(),
37 stopSession: () => unsupported(),
38 listSessions: () => Effect.succeed([]),
39 listCheckpoints: () => unsupported(),
40 getCheckpointDiff,
41 revertToCheckpoint: () => unsupported(),
42 stopAll: () => Effect.void,
43 streamEvents: Stream.fromPubSub(runtimeEventPubSub),
44 };
45
46 const emit = (event: ProviderRuntimeEvent): void => {
47 Effect.runSync(PubSub.publish(runtimeEventPubSub, event));
48 };
49
50 return {
51 service,
52 emit,
53 getCheckpointDiff,
54 };
55}
56
57async function waitForThread(
58 engine: OrchestrationEngineShape,
59 predicate: (thread: {
60 session: { status: string; activeTurnId: string | null; lastError: string | null } | null;
61 messages: ReadonlyArray<{ id: string; text: string; streaming: boolean }>;
62 }) => boolean,
63 timeoutMs = 2000,
64) {
65 const deadline = Date.now() + timeoutMs;
66 const poll = async (): Promise<{
67 session: { status: string; activeTurnId: string | null; lastError: string | null } | null;
68 messages: ReadonlyArray<{ id: string; text: string; streaming: boolean }>;
69 }> => {
70 const readModel = await Effect.runPromise(engine.getReadModel());
71 const thread = readModel.threads.find((entry) => entry.id === "thread-1");
72 if (thread && predicate(thread)) {
73 return thread;
74 }
75 if (Date.now() >= deadline) {
76 throw new Error("Timed out waiting for thread state");
77 }
78 await new Promise((resolve) => setTimeout(resolve, 10));
79 return poll();
80 };
81 return poll();
82}
83
84describe("ProviderRuntimeIngestion", () => {
85 let runtime: ManagedRuntime.ManagedRuntime<
86 OrchestrationEngineService | ProviderRuntimeIngestionService,
87 unknown
88 > | null = null;
89 let scope: Scope.Closeable | null = null;
90
91 afterEach(async () => {
92 if (scope) {
93 await Effect.runPromise(Scope.close(scope, Exit.void));
94 }
95 scope = null;
96 if (runtime) {
97 await runtime.dispose();
98 }
99 runtime = null;
100 });
101
102 async function createHarness() {
103 const provider = createProviderServiceHarness();
104 const orchestrationLayer = OrchestrationEngineLive.pipe(
105 Layer.provide(OrchestrationEventStoreLive),
106 Layer.provide(SqlitePersistenceMemory),
107 );
108 const layer = ProviderRuntimeIngestionLive.pipe(
109 Layer.provideMerge(orchestrationLayer),
110 Layer.provideMerge(Layer.succeed(ProviderService, provider.service)),
111 );
112 runtime = ManagedRuntime.make(layer);
113 const engine = await runtime.runPromise(Effect.service(OrchestrationEngineService));
114 const ingestion = await runtime.runPromise(Effect.service(ProviderRuntimeIngestionService));
115 scope = await Effect.runPromise(Scope.make("sequential"));
116 await Effect.runPromise(ingestion.start.pipe(Scope.provide(scope)));
117 await Effect.runPromise(Effect.sleep("10 millis"));
118
119 const createdAt = new Date().toISOString();
120 await Effect.runPromise(
121 engine.dispatch({
122 type: "thread.create",
123 commandId: "cmd-thread-create",
124 threadId: "thread-1",
125 projectId: "project-1",
126 title: "Thread",
127 model: "gpt-5-codex",
128 branch: null,
129 worktreePath: null,
130 createdAt,
131 }),
132 );
133 await Effect.runPromise(
134 engine.dispatch({
135 type: "thread.session",
136 commandId: "cmd-session-seed",
137 threadId: "thread-1",
138 session: {
139 sessionId: "sess-1",
140 provider: "codex",
141 status: "ready",
142 threadId: "thread-1",
143 activeTurnId: null,
144 createdAt,
145 updatedAt: createdAt,
146 lastError: null,
147 },
148 createdAt,
149 }),
150 );
151
152 return {
153 engine,
154 emit: provider.emit,
155 };
156 }
157
158 it("maps turn started/completed events into thread session updates", async () => {
159 const harness = await createHarness();
160 const now = new Date().toISOString();
161
162 harness.emit({
163 type: "turn.started",
164 eventId: "evt-turn-started",
165 provider: "codex",
166 sessionId: "sess-1",
167 createdAt: now,
168 turnId: "turn-1",
169 });
170
171 await waitForThread(
172 harness.engine,
173 (thread) => thread.session?.status === "running" && thread.session?.activeTurnId === "turn-1",
174 );
175
176 harness.emit({
177 type: "turn.completed",
178 eventId: "evt-turn-completed",
179 provider: "codex",
180 sessionId: "sess-1",
181 createdAt: new Date().toISOString(),
182 turnId: "turn-1",
183 status: "failed",
184 errorMessage: "turn failed",
185 });
186
187 const thread = await waitForThread(
188 harness.engine,
189 (entry) =>
190 entry.session?.status === "ready" &&
191 entry.session?.activeTurnId === null &&
192 entry.session?.lastError === "turn failed",
193 );
194 expect(thread.session?.status).toBe("ready");
195 expect(thread.session?.lastError).toBe("turn failed");
196 });
197
198 it("maps message delta/completed into finalized assistant messages", async () => {
199 const harness = await createHarness();
200 const now = new Date().toISOString();
201
202 harness.emit({
203 type: "message.delta",
204 eventId: "evt-message-delta-1",
205 provider: "codex",
206 sessionId: "sess-1",
207 createdAt: now,
208 turnId: "turn-2",
209 itemId: "item-1",
210 delta: "hello",
211 });
212 harness.emit({
213 type: "message.delta",
214 eventId: "evt-message-delta-2",
215 provider: "codex",
216 sessionId: "sess-1",
217 createdAt: now,
218 turnId: "turn-2",
219 itemId: "item-1",
220 delta: " world",
221 });
222 harness.emit({
223 type: "message.completed",
224 eventId: "evt-message-completed",
225 provider: "codex",
226 sessionId: "sess-1",
227 createdAt: now,
228 turnId: "turn-2",
229 itemId: "item-1",
230 });
231
232 const thread = await waitForThread(harness.engine, (entry) =>
233 entry.messages.some((message) => message.id === "assistant:item-1" && !message.streaming),
234 );
235 const message = thread.messages.find((entry) => entry.id === "assistant:item-1");
236 expect(message?.text).toBe("hello world");
237 expect(message?.streaming).toBe(false);
238 });
239
240 it("maps runtime.error into errored session state", async () => {
241 const harness = await createHarness();
242 const now = new Date().toISOString();
243
244 harness.emit({
245 type: "runtime.error",
246 eventId: "evt-runtime-error",
247 provider: "codex",
248 sessionId: "sess-1",
249 createdAt: now,
250 turnId: "turn-3",
251 message: "runtime exploded",
252 });
253
254 const thread = await waitForThread(
255 harness.engine,
256 (entry) =>
257 entry.session?.status === "error" &&
258 entry.session?.activeTurnId === "turn-3" &&
259 entry.session?.lastError === "runtime exploded",
260 );
261 expect(thread.session?.status).toBe("error");
262 expect(thread.session?.lastError).toBe("runtime exploded");
263 });
264});
apps/server/src/orchestration/Layers/ProviderRuntimeIngestion.ts created+326
...@@ -0,0 +1,326 @@
1import type { ProviderRuntimeEvent } from "@t3tools/contracts";
2import { Effect, Layer, Queue, Stream } from "effect";
3
4import {
5 parseTurnDiffFilesFromUnifiedDiff,
6 type TurnDiffFileSummary,
7} from "../../checkpointing/Diffs.ts";
8import { createLogger } from "../../logger.ts";
9import { ProviderService } from "../../provider/Services/ProviderService.ts";
10import { OrchestrationEngineService } from "../Services/OrchestrationEngine.ts";
11import {
12 ProviderRuntimeIngestionService,
13 type ProviderRuntimeIngestionShape,
14} from "../Services/ProviderRuntimeIngestion.ts";
15
16const providerTurnKey = (sessionId: string, turnId: string) => `${sessionId}:${turnId}`;
17
18type ActivityTone = "thinking" | "tool" | "info" | "error";
19
20interface ThreadActivityInput {
21 readonly id: string;
22 readonly createdAt: string;
23 readonly label: string;
24 readonly detail?: string;
25 readonly tone: ActivityTone;
26 readonly turnId?: string;
27 readonly requestId?: string;
28 readonly requestKind?: "command" | "file-change";
29}
30
31function truncateDetail(value: string, limit = 180): string {
32 return value.length > limit ? `${value.slice(0, limit - 1)}...` : value;
33}
34
35function runtimeEventToActivities(event: ProviderRuntimeEvent): ReadonlyArray<ThreadActivityInput> {
36 switch (event.type) {
37 case "approval.requested":
38 return [
39 {
40 id: event.eventId,
41 createdAt: event.createdAt,
42 label:
43 event.requestKind === "command"
44 ? "Command approval requested"
45 : "File-change approval requested",
46 ...(event.detail ? { detail: truncateDetail(event.detail) } : {}),
47 tone: "tool",
48 ...(event.turnId ? { turnId: event.turnId } : {}),
49 requestId: event.requestId,
50 requestKind: event.requestKind,
51 },
52 ];
53 case "approval.resolved":
54 return [
55 {
56 id: event.eventId,
57 createdAt: event.createdAt,
58 label: "Approval resolved",
59 tone: "info",
60 ...(event.turnId ? { turnId: event.turnId } : {}),
61 requestId: event.requestId,
62 ...(event.requestKind ? { requestKind: event.requestKind } : {}),
63 },
64 ];
65 case "runtime.error":
66 return [
67 {
68 id: event.eventId,
69 createdAt: event.createdAt,
70 label: "Runtime error",
71 detail: truncateDetail(event.message),
72 tone: "error",
73 ...(event.turnId ? { turnId: event.turnId } : {}),
74 },
75 ];
76 case "checkpoint.captured":
77 return [
78 {
79 id: event.eventId,
80 createdAt: event.createdAt,
81 label: "Checkpoint captured",
82 tone: "info",
83 ...(event.turnId ? { turnId: event.turnId } : {}),
84 },
85 ];
86 case "tool.completed":
87 return [
88 {
89 id: event.eventId,
90 createdAt: event.createdAt,
91 label: `${event.title} complete`,
92 ...(event.detail ? { detail: truncateDetail(event.detail) } : {}),
93 tone: "tool",
94 ...(event.turnId ? { turnId: event.turnId } : {}),
95 },
96 ];
97 default:
98 return [];
99 }
100}
101
102const make = Effect.gen(function* () {
103 const orchestrationEngine = yield* OrchestrationEngineService;
104 const providerService = yield* ProviderService;
105 const logger = createLogger("provider-runtime-ingestion");
106
107 const turnMessageIdsByTurnKey = new Map<string, Set<string>>();
108 const latestMessageIdByTurnKey = new Map<string, string>();
109
110 const rememberAssistantMessageId = (sessionId: string, turnId: string, messageId: string) => {
111 const key = providerTurnKey(sessionId, turnId);
112 const existingIds = turnMessageIdsByTurnKey.get(key);
113 if (existingIds) {
114 existingIds.add(messageId);
115 } else {
116 turnMessageIdsByTurnKey.set(key, new Set([messageId]));
117 }
118 latestMessageIdByTurnKey.set(key, messageId);
119 };
120
121 const getAssistantMessageIdsForTurn = (sessionId: string, turnId: string) => {
122 return turnMessageIdsByTurnKey.get(providerTurnKey(sessionId, turnId)) ?? new Set<string>();
123 };
124
125 const clearAssistantMessageIdsForTurn = (sessionId: string, turnId: string) => {
126 turnMessageIdsByTurnKey.delete(providerTurnKey(sessionId, turnId));
127 };
128
129 const getLatestAssistantMessageIdForTurn = (sessionId: string, turnId: string) => {
130 return latestMessageIdByTurnKey.get(providerTurnKey(sessionId, turnId));
131 };
132
133 const clearTurnStateForSession = (sessionId: string) => {
134 const prefix = `${sessionId}:`;
135 for (const key of turnMessageIdsByTurnKey.keys()) {
136 if (key.startsWith(prefix)) {
137 turnMessageIdsByTurnKey.delete(key);
138 }
139 }
140 for (const key of latestMessageIdByTurnKey.keys()) {
141 if (key.startsWith(prefix)) {
142 latestMessageIdByTurnKey.delete(key);
143 }
144 }
145 };
146
147 const processEvent = (event: ProviderRuntimeEvent) =>
148 Effect.gen(function* () {
149 const readModel = yield* orchestrationEngine.getReadModel();
150 const thread = readModel.threads.find(
151 (entry) => entry.session?.sessionId === event.sessionId,
152 );
153 if (!thread) return;
154
155 const now = event.createdAt;
156 if (event.type === "turn.started" || event.type === "turn.completed") {
157 yield* orchestrationEngine.dispatch({
158 type: "thread.session",
159 commandId: crypto.randomUUID(),
160 threadId: thread.id,
161 session: {
162 sessionId: event.sessionId,
163 provider: event.provider,
164 status: event.type === "turn.started" ? "running" : "ready",
165 threadId: thread.id,
166 activeTurnId: event.type === "turn.started" ? event.turnId : null,
167 createdAt: thread.session?.createdAt ?? now,
168 updatedAt: now,
169 lastError:
170 event.type === "turn.completed" && event.status === "failed"
171 ? (event.errorMessage ?? null)
172 : null,
173 },
174 createdAt: now,
175 });
176 }
177
178 if (event.type === "message.delta" && event.delta.length > 0) {
179 const assistantMessageId = `assistant:${event.itemId ?? event.turnId ?? event.sessionId}`;
180 if (event.turnId) {
181 rememberAssistantMessageId(event.sessionId, event.turnId, assistantMessageId);
182 }
183 yield* orchestrationEngine.dispatch({
184 type: "message.send",
185 commandId: crypto.randomUUID(),
186 threadId: thread.id,
187 messageId: assistantMessageId,
188 role: "assistant",
189 text: event.delta,
190 streaming: true,
191 createdAt: now,
192 });
193 }
194
195 if (event.type === "message.completed") {
196 const assistantMessageId = `assistant:${event.itemId}`;
197 if (event.turnId) {
198 rememberAssistantMessageId(event.sessionId, event.turnId, assistantMessageId);
199 }
200 yield* orchestrationEngine.dispatch({
201 type: "message.send",
202 commandId: crypto.randomUUID(),
203 threadId: thread.id,
204 messageId: assistantMessageId,
205 role: "assistant",
206 text: "",
207 streaming: false,
208 createdAt: now,
209 });
210 }
211
212 if (event.type === "turn.completed" && event.turnId) {
213 const assistantMessageIds = getAssistantMessageIdsForTurn(event.sessionId, event.turnId);
214 yield* Effect.forEach(assistantMessageIds, (assistantMessageId) =>
215 orchestrationEngine.dispatch({
216 type: "message.send",
217 commandId: crypto.randomUUID(),
218 threadId: thread.id,
219 messageId: assistantMessageId,
220 role: "assistant",
221 text: "",
222 streaming: false,
223 createdAt: now,
224 }),
225 ).pipe(Effect.asVoid);
226 clearAssistantMessageIdsForTurn(event.sessionId, event.turnId);
227 }
228
229 if (event.type === "session.exited") {
230 clearTurnStateForSession(event.sessionId);
231 }
232
233 if (event.type === "checkpoint.captured" && event.turnId) {
234 const files: ReadonlyArray<TurnDiffFileSummary> = yield* providerService
235 .getCheckpointDiff({
236 sessionId: event.sessionId,
237 fromTurnCount: Math.max(0, event.turnCount - 1),
238 toTurnCount: event.turnCount,
239 })
240 .pipe(
241 Effect.map((result) => parseTurnDiffFilesFromUnifiedDiff(result.diff)),
242 Effect.catch((error) =>
243 Effect.sync(() => {
244 logger.warn("failed to derive checkpoint file summary", {
245 sessionId: event.sessionId,
246 turnId: event.turnId,
247 turnCount: event.turnCount,
248 error:
249 error instanceof Error
250 ? error.message
251 : typeof error === "string"
252 ? error
253 : "unknown error",
254 });
255 }).pipe(Effect.as([] as ReadonlyArray<TurnDiffFileSummary>)),
256 ),
257 );
258 const assistantMessageId =
259 getLatestAssistantMessageIdForTurn(event.sessionId, event.turnId) ??
260 `assistant:${event.turnId}`;
261 yield* orchestrationEngine.dispatch({
262 type: "thread.turnDiff.complete",
263 commandId: crypto.randomUUID(),
264 threadId: thread.id,
265 turnId: event.turnId,
266 completedAt: now,
267 ...(event.status ? { status: event.status } : {}),
268 files,
269 assistantMessageId,
270 checkpointTurnCount: event.turnCount,
271 createdAt: now,
272 });
273 }
274
275 if (event.type === "runtime.error") {
276 yield* orchestrationEngine.dispatch({
277 type: "thread.session",
278 commandId: crypto.randomUUID(),
279 threadId: thread.id,
280 session: {
281 sessionId: event.sessionId,
282 provider: event.provider,
283 status: "error",
284 threadId: thread.id,
285 activeTurnId: event.turnId ?? null,
286 createdAt: thread.session?.createdAt ?? now,
287 updatedAt: now,
288 lastError: event.message,
289 },
290 createdAt: now,
291 });
292 }
293
294 const activities = runtimeEventToActivities(event);
295 yield* Effect.forEach(activities, (activity) =>
296 orchestrationEngine.dispatch({
297 type: "thread.activity.append",
298 commandId: crypto.randomUUID(),
299 threadId: thread.id,
300 activity,
301 createdAt: activity.createdAt,
302 }),
303 ).pipe(Effect.asVoid);
304 });
305
306 const start: ProviderRuntimeIngestionShape["start"] = Effect.gen(function* () {
307 const providerEventQueue = yield* Queue.unbounded<ProviderRuntimeEvent>();
308
309 yield* Effect.addFinalizer(() => Queue.shutdown(providerEventQueue).pipe(Effect.asVoid));
310
311 yield* Effect.forkScoped(
312 Effect.forever(Queue.take(providerEventQueue).pipe(Effect.flatMap(processEvent))),
313 );
314 yield* Effect.forkScoped(
315 Stream.runForEach(providerService.streamEvents, (event) =>
316 Queue.offer(providerEventQueue, event).pipe(Effect.asVoid),
317 ),
318 );
319 });
320
321 return {
322 start,
323 } satisfies ProviderRuntimeIngestionShape;
324});
325
326export const ProviderRuntimeIngestionLive = Layer.effect(ProviderRuntimeIngestionService, make);
apps/server/src/orchestration/Services/ProviderRuntimeIngestion.ts created+17
...@@ -0,0 +1,17 @@
1import { ServiceMap } from "effect";
2import type { Effect, Scope } from "effect";
3
4export interface ProviderRuntimeIngestionShape {
5 /**
6 * Start ingesting provider runtime events into orchestration commands.
7 *
8 * The returned effect must be run in a scope so all worker fibers can be
9 * finalized on shutdown.
10 */
11 readonly start: Effect.Effect<void, never, Scope.Scope>;
12}
13
14export class ProviderRuntimeIngestionService extends ServiceMap.Service<
15 ProviderRuntimeIngestionService,
16 ProviderRuntimeIngestionShape
17>()("orchestration/ProviderRuntimeIngestion") {}
apps/server/src/wsServer.test.ts+122-1
...@@ -3,15 +3,18 @@ import os from "node:os";...@@ -3,15 +3,18 @@ import os from "node:os";
3import path from "node:path";3import path from "node:path";
4import { EventEmitter } from "node:events";4import { EventEmitter } from "node:events";
55
6import { Effect, Layer, ManagedRuntime } from "effect";6import { Effect, Layer, ManagedRuntime, PubSub, Stream } from "effect";
7import { describe, expect, it, afterEach, vi } from "vitest";7import { describe, expect, it, afterEach, vi } from "vitest";
8import { createServer } from "./wsServer";8import { createServer } from "./wsServer";
9import WebSocket from "ws";9import WebSocket from "ws";
1010
11import {11import {
12 DEFAULT_TERMINAL_ID,12 DEFAULT_TERMINAL_ID,
13 ORCHESTRATION_WS_CHANNELS,
14 ORCHESTRATION_WS_METHODS,
13 WS_CHANNELS,15 WS_CHANNELS,
14 WS_METHODS,16 WS_METHODS,
17 type ProviderRuntimeEvent,
15 type KeybindingsConfig,18 type KeybindingsConfig,
16 type ResolvedKeybindingsConfig,19 type ResolvedKeybindingsConfig,
17 type WsPush,20 type WsPush,
...@@ -33,6 +36,7 @@ import { ProjectRepositoryLive } from "./persistence/Layers/Projects";...@@ -33,6 +36,7 @@ import { ProjectRepositoryLive } from "./persistence/Layers/Projects";
33import { makeSqlitePersistenceLive, SqlitePersistenceMemory } from "./persistence/Layers/Sqlite";36import { makeSqlitePersistenceLive, SqlitePersistenceMemory } from "./persistence/Layers/Sqlite";
34import { NodeServices } from "@effect/platform-node";37import { NodeServices } from "@effect/platform-node";
35import { SqlClient } from "effect/unstable/sql";38import { SqlClient } from "effect/unstable/sql";
39import { ProviderService, type ProviderServiceShape } from "./provider/Services/ProviderService";
3640
37interface PendingMessages {41interface PendingMessages {
38 queue: unknown[];42 queue: unknown[];
...@@ -200,6 +204,28 @@ async function sendRequest(ws: WebSocket, method: string, params?: unknown): Pro...@@ -200,6 +204,28 @@ async function sendRequest(ws: WebSocket, method: string, params?: unknown): Pro
200 }204 }
201}205}
202206
207async function waitForPush(
208 ws: WebSocket,
209 channel: string,
210 predicate?: (push: WsPush) => boolean,
211 maxMessages = 120,
212): Promise<WsPush> {
213 const take = async (remaining: number): Promise<WsPush> => {
214 if (remaining <= 0) {
215 throw new Error(`Timed out waiting for push on ${channel}`);
216 }
217 const message = (await waitForMessage(ws)) as WsPush;
218 if (message.type !== "push" || message.channel !== channel) {
219 return take(remaining - 1);
220 }
221 if (!predicate || predicate(message)) {
222 return message;
223 }
224 return take(remaining - 1);
225 };
226 return take(maxMessages);
227}
228
203function compileKeybindings(bindings: KeybindingsConfig): ResolvedKeybindingsConfig {229function compileKeybindings(bindings: KeybindingsConfig): ResolvedKeybindingsConfig {
204 const resolved: ResolvedKeybindingsConfig = [];230 const resolved: ResolvedKeybindingsConfig = [];
205 for (const binding of bindings) {231 for (const binding of bindings) {
...@@ -244,6 +270,7 @@ describe("WebSocket Server", () => {...@@ -244,6 +270,7 @@ describe("WebSocket Server", () => {
244 devUrl?: string;270 devUrl?: string;
245 authToken?: string;271 authToken?: string;
246 stateDir?: string;272 stateDir?: string;
273 providerLayer?: Layer.Layer<ProviderService, unknown>;
247 gitManager?: {274 gitManager?: {
248 status: (input: { cwd: string }) => Promise<unknown>;275 status: (input: { cwd: string }) => Promise<unknown>;
249 runStackedAction: (input: { cwd: string; action: string }) => Promise<unknown>;276 runStackedAction: (input: { cwd: string; action: string }) => Promise<unknown>;
...@@ -259,6 +286,7 @@ describe("WebSocket Server", () => {...@@ -259,6 +286,7 @@ describe("WebSocket Server", () => {
259 persistenceLayer: options.persistenceLayer ?? SqlitePersistenceMemory,286 persistenceLayer: options.persistenceLayer ?? SqlitePersistenceMemory,
260 ...(options.devUrl ? { devUrl: options.devUrl } : {}),287 ...(options.devUrl ? { devUrl: options.devUrl } : {}),
261 ...(options.authToken ? { authToken: options.authToken } : {}),288 ...(options.authToken ? { authToken: options.authToken } : {}),
289 ...(options.providerLayer ? { providerLayer: options.providerLayer } : {}),
262 ...(options.gitManager ? { gitManager: options.gitManager as never } : {}),290 ...(options.gitManager ? { gitManager: options.gitManager as never } : {}),
263 ...(options.terminalManager ? { terminalManager: options.terminalManager } : {}),291 ...(options.terminalManager ? { terminalManager: options.terminalManager } : {}),
264 });292 });
...@@ -668,6 +696,99 @@ describe("WebSocket Server", () => {...@@ -668,6 +696,99 @@ describe("WebSocket Server", () => {
668 expect(diffResponse.error?.message).toContain("Unknown provider session");696 expect(diffResponse.error?.message).toContain("Unknown provider session");
669 });697 });
670698
699 it("keeps orchestration domain push behavior for provider runtime events", async () => {
700 const runtimeEventPubSub = Effect.runSync(PubSub.unbounded<ProviderRuntimeEvent>());
701 const emitRuntimeEvent = (event: ProviderRuntimeEvent) => {
702 Effect.runSync(PubSub.publish(runtimeEventPubSub, event));
703 };
704 const unsupported = () => Effect.die(new Error("Unsupported provider call in test")) as never;
705 const providerService: ProviderServiceShape = {
706 startSession: () => unsupported(),
707 sendTurn: () => unsupported(),
708 interruptTurn: () => unsupported(),
709 respondToRequest: () => unsupported(),
710 stopSession: () => unsupported(),
711 listSessions: () => Effect.succeed([]),
712 listCheckpoints: () => unsupported(),
713 getCheckpointDiff: (input) =>
714 Effect.succeed({
715 threadId: "thread-1",
716 fromTurnCount: input.fromTurnCount,
717 toTurnCount: input.toTurnCount,
718 diff: "",
719 }),
720 revertToCheckpoint: () => unsupported(),
721 stopAll: () => Effect.void,
722 streamEvents: Stream.fromPubSub(runtimeEventPubSub),
723 };
724 const providerLayer = Layer.succeed(ProviderService, providerService);
725
726 server = createTestServer({
727 cwd: "/test",
728 providerLayer,
729 });
730 await server.start();
731 const addr = server.httpServer.address();
732 const port = typeof addr === "object" && addr !== null ? addr.port : 0;
733
734 const ws = await connectWs(port);
735 connections.push(ws);
736 await waitForMessage(ws);
737
738 const createdAt = new Date().toISOString();
739 const createThreadResponse = await sendRequest(ws, ORCHESTRATION_WS_METHODS.dispatchCommand, {
740 type: "thread.create",
741 commandId: "cmd-ws-runtime-thread-create",
742 threadId: "thread-1",
743 projectId: "project-1",
744 title: "Thread 1",
745 model: "gpt-5-codex",
746 branch: null,
747 worktreePath: null,
748 createdAt,
749 });
750 expect(createThreadResponse.error).toBeUndefined();
751
752 const setSessionResponse = await sendRequest(ws, ORCHESTRATION_WS_METHODS.dispatchCommand, {
753 type: "thread.session",
754 commandId: "cmd-ws-runtime-session-set",
755 threadId: "thread-1",
756 session: {
757 sessionId: "sess-test",
758 provider: "codex",
759 status: "ready",
760 threadId: "thread-1",
761 activeTurnId: null,
762 createdAt,
763 updatedAt: createdAt,
764 lastError: null,
765 },
766 createdAt,
767 });
768 expect(setSessionResponse.error).toBeUndefined();
769
770 emitRuntimeEvent({
771 type: "message.delta",
772 eventId: "evt-ws-runtime-message-delta",
773 provider: "codex",
774 sessionId: "sess-test",
775 createdAt: new Date().toISOString(),
776 turnId: "turn-1",
777 itemId: "item-1",
778 delta: "hello from runtime",
779 });
780
781 const domainPush = await waitForPush(ws, ORCHESTRATION_WS_CHANNELS.domainEvent, (push) => {
782 const event = push.data as { type?: string; payload?: { id?: string; text?: string } };
783 return event.type === "message.sent" && event.payload?.id === "assistant:item-1";
784 });
785
786 const domainEvent = domainPush.data as { type: string; payload: { id: string; text: string } };
787 expect(domainEvent.type).toBe("message.sent");
788 expect(domainEvent.payload.id).toBe("assistant:item-1");
789 expect(domainEvent.payload.text).toBe("hello from runtime");
790 });
791
671 it("routes terminal RPC methods and broadcasts terminal events", async () => {792 it("routes terminal RPC methods and broadcasts terminal events", async () => {
672 const cwd = makeTempDir("t3code-ws-terminal-cwd-");793 const cwd = makeTempDir("t3code-ws-terminal-cwd-");
673 const terminalManager = new MockTerminalManager();794 const terminalManager = new MockTerminalManager();
apps/server/src/wsServer.ts+76-396
...@@ -11,7 +11,6 @@ import {...@@ -11,7 +11,6 @@ import {
11 ORCHESTRATION_WS_METHODS,11 ORCHESTRATION_WS_METHODS,
12 WS_CHANNELS,12 WS_CHANNELS,
13 WS_METHODS,13 WS_METHODS,
14 type ProviderRuntimeEvent,
15 type TerminalEvent,14 type TerminalEvent,
16 type WsPush,15 type WsPush,
17 type WsRequest,16 type WsRequest,
...@@ -19,7 +18,7 @@ import {...@@ -19,7 +18,7 @@ import {
19 wsRequestSchema,18 wsRequestSchema,
20} from "@t3tools/contracts";19} from "@t3tools/contracts";
21import { NodeServices } from "@effect/platform-node";20import { NodeServices } from "@effect/platform-node";
22import { Effect, Exit, Layer, ManagedRuntime, Queue, Scope, Stream } from "effect";21import { Effect, Exit, Layer, ManagedRuntime, Scope, Stream } from "effect";
23import { WebSocketServer, type WebSocket } from "ws";22import { WebSocketServer, type WebSocket } from "ws";
2423
25import { createLogger } from "./logger";24import { createLogger } from "./logger";
...@@ -37,10 +36,12 @@ import { TerminalManager } from "./terminalManager";...@@ -37,10 +36,12 @@ import { TerminalManager } from "./terminalManager";
37import { loadResolvedKeybindingsConfig, upsertKeybindingRule } from "./keybindings";36import { loadResolvedKeybindingsConfig, upsertKeybindingRule } from "./keybindings";
38import { searchWorkspaceEntries } from "./workspaceEntries";37import { searchWorkspaceEntries } from "./workspaceEntries";
39import { OrchestrationEngineLive } from "./orchestration/Layers/OrchestrationEngine";38import { OrchestrationEngineLive } from "./orchestration/Layers/OrchestrationEngine";
39import { ProviderRuntimeIngestionLive } from "./orchestration/Layers/ProviderRuntimeIngestion";
40import {40import {
41 OrchestrationEngineService,41 OrchestrationEngineService,
42 type OrchestrationEngineShape,42 type OrchestrationEngineShape,
43} from "./orchestration/Services/OrchestrationEngine";43} from "./orchestration/Services/OrchestrationEngine";
44import { ProviderRuntimeIngestionService } from "./orchestration/Services/ProviderRuntimeIngestion";
44import { ProjectRepositoryLive } from "./persistence/Layers/Projects";45import { ProjectRepositoryLive } from "./persistence/Layers/Projects";
45import { makeSqlitePersistenceLive } from "./persistence/Layers/Sqlite";46import { makeSqlitePersistenceLive } from "./persistence/Layers/Sqlite";
46import { ProjectRepository, type ProjectRepositoryShape } from "./persistence/Services/Projects";47import { ProjectRepository, type ProjectRepositoryShape } from "./persistence/Services/Projects";
...@@ -53,7 +54,6 @@ import { ProviderSessionDirectoryLive } from "./provider/Layers/ProviderSessionD...@@ -53,7 +54,6 @@ import { ProviderSessionDirectoryLive } from "./provider/Layers/ProviderSessionD
53import { ProviderService, type ProviderServiceShape } from "./provider/Services/ProviderService";54import { ProviderService, type ProviderServiceShape } from "./provider/Services/ProviderService";
54import { CheckpointStoreLive } from "./checkpointing/Layers/CheckpointStore";55import { CheckpointStoreLive } from "./checkpointing/Layers/CheckpointStore";
55import { CheckpointServiceLive } from "./checkpointing/Layers/CheckpointService";56import { CheckpointServiceLive } from "./checkpointing/Layers/CheckpointService";
56import { parseTurnDiffFilesFromUnifiedDiff, type TurnDiffFileSummary } from "./checkpointing/Diffs";
57import { CheckpointRepositoryLive } from "./persistence/Layers/Checkpoints";57import { CheckpointRepositoryLive } from "./persistence/Layers/Checkpoints";
58import { makeEventNdjsonLogger, type EventNdjsonLogger } from "./provider/Layers/EventNdjsonLogger";58import { makeEventNdjsonLogger, type EventNdjsonLogger } from "./provider/Layers/EventNdjsonLogger";
59import * as SqlClient from "effect/unstable/sql/SqlClient";59import * as SqlClient from "effect/unstable/sql/SqlClient";
...@@ -88,6 +88,7 @@ export interface ServerOptions {...@@ -88,6 +88,7 @@ export interface ServerOptions {
88 logWebSocketEvents?: boolean | undefined;88 logWebSocketEvents?: boolean | undefined;
89 gitManager?: GitManager | undefined;89 gitManager?: GitManager | undefined;
90 terminalManager?: TerminalManager | undefined;90 terminalManager?: TerminalManager | undefined;
91 providerLayer?: Layer.Layer<ProviderService, unknown> | undefined;
91 authToken?: string | undefined;92 authToken?: string | undefined;
92}93}
9394
...@@ -119,93 +120,13 @@ function parseBooleanEnv(value: string | undefined): boolean | undefined {...@@ -119,93 +120,13 @@ function parseBooleanEnv(value: string | undefined): boolean | undefined {
119 return undefined;120 return undefined;
120}121}
121122
122const providerTurnKey = (sessionId: string, turnId: string) => `${sessionId}:${turnId}`;123type EffectRuntime = ManagedRuntime.ManagedRuntime<
123124 | ProjectRepository
124type ActivityTone = "thinking" | "tool" | "info" | "error";125 | OrchestrationEngineService
125126 | ProviderService
126interface ThreadActivityInput {127 | ProviderRuntimeIngestionService,
127 readonly id: string;128 unknown
128 readonly createdAt: string;129>;
129 readonly label: string;
130 readonly detail?: string;
131 readonly tone: ActivityTone;
132 readonly turnId?: string;
133 readonly requestId?: string;
134 readonly requestKind?: "command" | "file-change";
135}
136
137function truncateDetail(value: string, limit = 180): string {
138 return value.length > limit ? `${value.slice(0, limit - 1)}...` : value;
139}
140
141function runtimeEventToActivities(event: ProviderRuntimeEvent): ReadonlyArray<ThreadActivityInput> {
142 switch (event.type) {
143 case "approval.requested":
144 return [
145 {
146 id: event.eventId,
147 createdAt: event.createdAt,
148 label:
149 event.requestKind === "command"
150 ? "Command approval requested"
151 : "File-change approval requested",
152 ...(event.detail ? { detail: truncateDetail(event.detail) } : {}),
153 tone: "tool",
154 ...(event.turnId ? { turnId: event.turnId } : {}),
155 requestId: event.requestId,
156 requestKind: event.requestKind,
157 },
158 ];
159 case "approval.resolved":
160 return [
161 {
162 id: event.eventId,
163 createdAt: event.createdAt,
164 label: "Approval resolved",
165 tone: "info",
166 ...(event.turnId ? { turnId: event.turnId } : {}),
167 requestId: event.requestId,
168 ...(event.requestKind ? { requestKind: event.requestKind } : {}),
169 },
170 ];
171 case "runtime.error":
172 return [
173 {
174 id: event.eventId,
175 createdAt: event.createdAt,
176 label: "Runtime error",
177 detail: truncateDetail(event.message),
178 tone: "error",
179 ...(event.turnId ? { turnId: event.turnId } : {}),
180 },
181 ];
182 case "checkpoint.captured":
183 return [
184 {
185 id: event.eventId,
186 createdAt: event.createdAt,
187 label: "Checkpoint captured",
188 tone: "info",
189 ...(event.turnId ? { turnId: event.turnId } : {}),
190 },
191 ];
192 case "tool.completed":
193 return [
194 {
195 id: event.eventId,
196 createdAt: event.createdAt,
197 label: `${event.title} complete`,
198 ...(event.detail ? { detail: truncateDetail(event.detail) } : {}),
199 tone: "tool",
200 ...(event.turnId ? { turnId: event.turnId } : {}),
201 },
202 ];
203 default:
204 return [];
205 }
206}
207
208const noop = () => {};
209130
210export function createServer(options: ServerOptions) {131export function createServer(options: ServerOptions) {
211 const {132 const {
...@@ -219,14 +140,12 @@ export function createServer(options: ServerOptions) {...@@ -219,14 +140,12 @@ export function createServer(options: ServerOptions) {
219 logWebSocketEvents: explicitLogWsEvents,140 logWebSocketEvents: explicitLogWsEvents,
220 gitManager = new GitManager(),141 gitManager = new GitManager(),
221 terminalManager = new TerminalManager(),142 terminalManager = new TerminalManager(),
143 providerLayer: customProviderLayer,
222 authToken,144 authToken,
223 } = options;145 } = options;
224146
225 const resolvedStateDir = stateDir ?? path.join(os.homedir(), ".t3", "userdata");147 const resolvedStateDir = stateDir ?? path.join(os.homedir(), ".t3", "userdata");
226 let effectRuntime: ManagedRuntime.ManagedRuntime<148 let effectRuntime: EffectRuntime | null = null;
227 ProjectRepository | OrchestrationEngineService | ProviderService,
228 unknown
229 > | null = null;
230 let projectRepository: ProjectRepositoryShape | null = null;149 let projectRepository: ProjectRepositoryShape | null = null;
231 let orchestrationEngine: OrchestrationEngineShape | null = null;150 let orchestrationEngine: OrchestrationEngineShape | null = null;
232 let providerService: ProviderServiceShape | null = null;151 let providerService: ProviderServiceShape | null = null;
...@@ -237,8 +156,7 @@ export function createServer(options: ServerOptions) {...@@ -237,8 +156,7 @@ export function createServer(options: ServerOptions) {
237 let keybindingsConfig = loadResolvedKeybindingsConfig(logger);156 let keybindingsConfig = loadResolvedKeybindingsConfig(logger);
238 let orchestrationDomainEventLogger: EventNdjsonLogger | undefined;157 let orchestrationDomainEventLogger: EventNdjsonLogger | undefined;
239 let orchestrationCommandLogger: EventNdjsonLogger | undefined;158 let orchestrationCommandLogger: EventNdjsonLogger | undefined;
240 let unsubscribeDomainEvents = noop;159 let subscriptionsScope: Scope.Closeable | null = null;
241 let unsubscribeProviderEvents = noop;
242160
243 function logOutgoingPush(push: WsPush, recipients: number) {161 function logOutgoingPush(push: WsPush, recipients: number) {
244 if (!logWebSocketEvents) return;162 if (!logWebSocketEvents) return;
...@@ -269,282 +187,36 @@ export function createServer(options: ServerOptions) {...@@ -269,282 +187,36 @@ export function createServer(options: ServerOptions) {
269 return providerService;187 return providerService;
270 };188 };
271189
272 async function attachOrchestrationSubscriptions(190 const attachSubscriptionss = Effect.gen(function* () {
273 runtime: ManagedRuntime.ManagedRuntime<191 const orchestrationEngine = yield* OrchestrationEngineService;
274 ProjectRepository | OrchestrationEngineService | ProviderService,192 const runtimeIngestion = yield* ProviderRuntimeIngestionService;
275 unknown193 subscriptionsScope = yield* Scope.make("sequential");
276 >,
277 orchestrationEngine: OrchestrationEngineShape,
278 ) {
279 const subscriptionScope = await runtime.runPromise(Scope.make("sequential"));
280 await runtime.runPromise(
281 Effect.gen(function* () {
282 yield* Effect.forkScoped(
283 Stream.runForEach(orchestrationEngine.streamDomainEvents, (event) =>
284 Effect.sync(() => {
285 orchestrationDomainEventLogger?.write({
286 observedAt: new Date().toISOString(),
287 event,
288 });
289 const push: WsPush = {
290 type: "push",
291 channel: ORCHESTRATION_WS_CHANNELS.domainEvent,
292 data: event,
293 };
294 const message = JSON.stringify(push);
295 let recipients = 0;
296 for (const client of clients) {
297 if (client.readyState === client.OPEN) {
298 client.send(message);
299 recipients += 1;
300 }
301 }
302 logOutgoingPush(push, recipients);
303 }),
304 ),
305 );
306 }).pipe(Scope.provide(subscriptionScope)),
307 );
308
309 unsubscribeDomainEvents = () => {
310 runtime.runFork(Scope.close(subscriptionScope, Exit.void));
311 };
312 }
313
314 async function attachProviderSubscriptions(
315 runtime: ManagedRuntime.ManagedRuntime<
316 ProjectRepository | OrchestrationEngineService | ProviderService,
317 unknown
318 >,
319 providerService: ProviderServiceShape,
320 orchestrationEngine: OrchestrationEngineShape,
321 ) {
322 const turnMessageIdsByTurnKey = new Map<string, Set<string>>();
323 const latestMessageIdByTurnKey = new Map<string, string>();
324
325 const rememberAssistantMessageId = (sessionId: string, turnId: string, messageId: string) => {
326 const key = providerTurnKey(sessionId, turnId);
327 const existingIds = turnMessageIdsByTurnKey.get(key);
328 if (existingIds) {
329 existingIds.add(messageId);
330 } else {
331 turnMessageIdsByTurnKey.set(key, new Set([messageId]));
332 }
333 latestMessageIdByTurnKey.set(key, messageId);
334 };
335
336 const getAssistantMessageIdsForTurn = (sessionId: string, turnId: string) => {
337 return turnMessageIdsByTurnKey.get(providerTurnKey(sessionId, turnId)) ?? new Set<string>();
338 };
339
340 const clearAssistantMessageIdsForTurn = (sessionId: string, turnId: string) => {
341 turnMessageIdsByTurnKey.delete(providerTurnKey(sessionId, turnId));
342 };
343
344 const getLatestAssistantMessageIdForTurn = (sessionId: string, turnId: string) => {
345 return latestMessageIdByTurnKey.get(providerTurnKey(sessionId, turnId));
346 };
347
348 const clearTurnStateForSession = (sessionId: string) => {
349 const prefix = `${sessionId}:`;
350 for (const key of turnMessageIdsByTurnKey.keys()) {
351 if (key.startsWith(prefix)) {
352 turnMessageIdsByTurnKey.delete(key);
353 }
354 }
355 for (const key of latestMessageIdByTurnKey.keys()) {
356 if (key.startsWith(prefix)) {
357 latestMessageIdByTurnKey.delete(key);
358 }
359 }
360 };
361
362 const processEvent = (event: ProviderRuntimeEvent) =>
363 Effect.gen(function* () {
364 const readModel = yield* orchestrationEngine.getReadModel();
365 const thread = readModel.threads.find(
366 (entry) => entry.session?.sessionId === event.sessionId,
367 );
368 if (!thread) return;
369
370 const now = event.createdAt;
371 if (event.type === "turn.started" || event.type === "turn.completed") {
372 yield* orchestrationEngine.dispatch({
373 type: "thread.session",
374 commandId: crypto.randomUUID(),
375 threadId: thread.id,
376 session: {
377 sessionId: event.sessionId,
378 provider: event.provider,
379 status: event.type === "turn.started" ? "running" : "ready",
380 threadId: thread.id,
381 activeTurnId: event.type === "turn.started" ? event.turnId : null,
382 createdAt: thread.session?.createdAt ?? now,
383 updatedAt: now,
384 lastError:
385 event.type === "turn.completed" && event.status === "failed"
386 ? (event.errorMessage ?? null)
387 : null,
388 },
389 createdAt: now,
390 });
391 }
392
393 if (event.type === "message.delta" && event.delta.length > 0) {
394 const assistantMessageId = `assistant:${event.itemId ?? event.turnId ?? event.sessionId}`;
395 if (event.turnId) {
396 rememberAssistantMessageId(event.sessionId, event.turnId, assistantMessageId);
397 }
398 yield* orchestrationEngine.dispatch({
399 type: "message.send",
400 commandId: crypto.randomUUID(),
401 threadId: thread.id,
402 messageId: assistantMessageId,
403 role: "assistant",
404 text: event.delta,
405 streaming: true,
406 createdAt: now,
407 });
408 }
409
410 if (event.type === "message.completed") {
411 const assistantMessageId = `assistant:${event.itemId}`;
412 if (event.turnId) {
413 rememberAssistantMessageId(event.sessionId, event.turnId, assistantMessageId);
414 }
415 yield* orchestrationEngine.dispatch({
416 type: "message.send",
417 commandId: crypto.randomUUID(),
418 threadId: thread.id,
419 messageId: assistantMessageId,
420 role: "assistant",
421 text: "",
422 streaming: false,
423 createdAt: now,
424 });
425 }
426194
427 if (event.type === "turn.completed") {195 yield* Stream.runForEach(orchestrationEngine.streamDomainEvents, (event) =>
428 if (event.turnId) {196 Effect.sync(() => {
429 const assistantMessageIds = getAssistantMessageIdsForTurn(197 orchestrationDomainEventLogger?.write({
430 event.sessionId,198 observedAt: new Date().toISOString(),
431 event.turnId,199 event,
432 );200 });
433 yield* Effect.forEach(assistantMessageIds, (assistantMessageId) =>201 const push: WsPush = {
434 orchestrationEngine.dispatch({202 type: "push",
435 type: "message.send",203 channel: ORCHESTRATION_WS_CHANNELS.domainEvent,
436 commandId: crypto.randomUUID(),204 data: event,
437 threadId: thread.id,205 };
438 messageId: assistantMessageId,206 const message = JSON.stringify(push);
439 role: "assistant",207 let recipients = 0;
440 text: "",208 for (const client of clients) {
441 streaming: false,209 if (client.readyState === client.OPEN) {
442 createdAt: now,210 client.send(message);
443 }),211 recipients += 1;
444 ).pipe(Effect.asVoid);
445 clearAssistantMessageIdsForTurn(event.sessionId, event.turnId);
446 }212 }
447 }213 }
214 logOutgoingPush(push, recipients);
215 }),
216 ).pipe(Effect.forkIn(subscriptionsScope));
448217
449 if (event.type === "session.exited") {218 yield* runtimeIngestion.start.pipe(Scope.provide(subscriptionsScope));
450 clearTurnStateForSession(event.sessionId);219 });
451 }
452
453 if (event.type === "checkpoint.captured" && event.turnId) {
454 const files: ReadonlyArray<TurnDiffFileSummary> = yield* providerService
455 .getCheckpointDiff({
456 sessionId: event.sessionId,
457 fromTurnCount: Math.max(0, event.turnCount - 1),
458 toTurnCount: event.turnCount,
459 })
460 .pipe(
461 Effect.map((result) => parseTurnDiffFilesFromUnifiedDiff(result.diff)),
462 Effect.catch((error) =>
463 Effect.sync(() => {
464 logger.warn("failed to derive checkpoint file summary", {
465 sessionId: event.sessionId,
466 turnId: event.turnId,
467 turnCount: event.turnCount,
468 error:
469 error instanceof Error
470 ? error.message
471 : typeof error === "string"
472 ? error
473 : "unknown error",
474 });
475 }).pipe(Effect.as([] as ReadonlyArray<TurnDiffFileSummary>)),
476 ),
477 );
478 const assistantMessageId =
479 getLatestAssistantMessageIdForTurn(event.sessionId, event.turnId) ??
480 `assistant:${event.turnId}`;
481 yield* orchestrationEngine.dispatch({
482 type: "thread.turnDiff.complete",
483 commandId: crypto.randomUUID(),
484 threadId: thread.id,
485 turnId: event.turnId,
486 completedAt: now,
487 ...(event.status ? { status: event.status } : {}),
488 files,
489 assistantMessageId,
490 checkpointTurnCount: event.turnCount,
491 createdAt: now,
492 });
493 }
494
495 if (event.type === "runtime.error") {
496 yield* orchestrationEngine.dispatch({
497 type: "thread.session",
498 commandId: crypto.randomUUID(),
499 threadId: thread.id,
500 session: {
501 sessionId: event.sessionId,
502 provider: event.provider,
503 status: "error",
504 threadId: thread.id,
505 activeTurnId: event.turnId ?? null,
506 createdAt: thread.session?.createdAt ?? now,
507 updatedAt: now,
508 lastError: event.message,
509 },
510 createdAt: now,
511 });
512 }
513
514 const activities = runtimeEventToActivities(event);
515 yield* Effect.forEach(activities, (activity) =>
516 orchestrationEngine.dispatch({
517 type: "thread.activity.append",
518 commandId: crypto.randomUUID(),
519 threadId: thread.id,
520 activity,
521 createdAt: activity.createdAt,
522 }),
523 ).pipe(Effect.asVoid);
524 });
525
526 const subscriptionScope = await runtime.runPromise(Scope.make("sequential"));
527 await runtime.runPromise(
528 Effect.gen(function* () {
529 const providerEventQueue = yield* Queue.unbounded<ProviderRuntimeEvent>();
530
531 yield* Effect.addFinalizer(() => Queue.shutdown(providerEventQueue).pipe(Effect.asVoid));
532
533 yield* Effect.forkScoped(
534 Effect.forever(Queue.take(providerEventQueue).pipe(Effect.flatMap(processEvent))),
535 );
536 yield* Effect.forkScoped(
537 Stream.runForEach(providerService.streamEvents, (event) =>
538 Queue.offer(providerEventQueue, event).pipe(Effect.asVoid),
539 ),
540 );
541 }).pipe(Scope.provide(subscriptionScope)),
542 );
543
544 unsubscribeProviderEvents = () => {
545 runtime.runFork(Scope.close(subscriptionScope, Exit.void));
546 };
547 }
548220
549 const onTerminalEvent = (event: TerminalEvent) => {221 const onTerminalEvent = (event: TerminalEvent) => {
550 const push: WsPush = {222 const push: WsPush = {
...@@ -955,29 +627,38 @@ export function createServer(options: ServerOptions) {...@@ -955,29 +627,38 @@ export function createServer(options: ServerOptions) {
955 path.join(orchestrationLogsDir, "orchestration-command.ndjson"),627 path.join(orchestrationLogsDir, "orchestration-command.ndjson"),
956 );628 );
957 const persistenceLayer = customPersistenceLayer ?? makeSqlitePersistenceLive(dbPath);629 const persistenceLayer = customPersistenceLayer ?? makeSqlitePersistenceLive(dbPath);
958 const orchestrationLayer = Layer.provide(630 const orchestrationLayer = Layer.provide(OrchestrationEngineLive, OrchestrationEventStoreLive);
959 OrchestrationEngineLive,631 const providerLayer =
960 OrchestrationEventStoreLive,632 customProviderLayer ??
961 );633 (() => {
962 const codexAdapterLayer = makeCodexAdapterLive({634 const codexAdapterLayer = makeCodexAdapterLive({
963 nativeEventLogPath: path.join(providerLogsDir, "provider-native.ndjson"),635 nativeEventLogPath: path.join(providerLogsDir, "provider-native.ndjson"),
964 });636 });
965 const adapterRegistryLayer = ProviderAdapterRegistryLive.pipe(Layer.provide(codexAdapterLayer));637 const adapterRegistryLayer = ProviderAdapterRegistryLive.pipe(
966 const checkpointStoreLayer = CheckpointStoreLive.pipe(Layer.provide(NodeServices.layer));638 Layer.provide(codexAdapterLayer),
967 const checkpointDependenciesLayer = Layer.mergeAll(639 );
968 checkpointStoreLayer,640 const checkpointStoreLayer = CheckpointStoreLive.pipe(Layer.provide(NodeServices.layer));
969 CheckpointRepositoryLive,641 const checkpointDependenciesLayer = Layer.mergeAll(
970 adapterRegistryLayer,642 checkpointStoreLayer,
971 ProviderSessionDirectoryLive,643 CheckpointRepositoryLive,
972 );644 adapterRegistryLayer,
973 const checkpointServiceLayer = CheckpointServiceLive.pipe(645 ProviderSessionDirectoryLive,
974 Layer.provideMerge(checkpointDependenciesLayer),646 );
647 const checkpointServiceLayer = CheckpointServiceLive.pipe(
648 Layer.provideMerge(checkpointDependenciesLayer),
649 );
650 return makeProviderServiceLive({
651 canonicalEventLogPath: path.join(providerLogsDir, "provider-canonical.ndjson"),
652 }).pipe(Layer.provideMerge(checkpointServiceLayer));
653 })();
654
655 const runtimeServicesLayer = Layer.mergeAll(
656 orchestrationLayer,
657 ProjectRepositoryLive,
658 providerLayer,
975 );659 );
976 const providerLayer = makeProviderServiceLive({660 const layer = ProviderRuntimeIngestionLive.pipe(
977 canonicalEventLogPath: path.join(providerLogsDir, "provider-canonical.ndjson"),661 Layer.provideMerge(runtimeServicesLayer),
978 }).pipe(Layer.provideMerge(checkpointServiceLayer));
979
980 const layer = Layer.mergeAll(orchestrationLayer, ProjectRepositoryLive, providerLayer).pipe(
981 Layer.provide(persistenceLayer),662 Layer.provide(persistenceLayer),
982 Layer.provideMerge(NodeServices.layer),663 Layer.provideMerge(NodeServices.layer),
983 );664 );
...@@ -993,8 +674,7 @@ export function createServer(options: ServerOptions) {...@@ -993,8 +674,7 @@ export function createServer(options: ServerOptions) {
993 projectRepository = repository;674 projectRepository = repository;
994 providerService = nextProviderService;675 providerService = nextProviderService;
995 await runtime.runPromise(repository.pruneMissing());676 await runtime.runPromise(repository.pruneMissing());
996 await attachOrchestrationSubscriptions(runtime, nextOrchestrationEngine);677 await runtime.runPromise(attachSubscriptionss);
997 await attachProviderSubscriptions(runtime, nextProviderService, nextOrchestrationEngine);
998 } catch (error) {678 } catch (error) {
999 await runtime.dispose().catch(() => undefined);679 await runtime.dispose().catch(() => undefined);
1000 throw error;680 throw error;
...@@ -1004,10 +684,10 @@ export function createServer(options: ServerOptions) {...@@ -1004,10 +684,10 @@ export function createServer(options: ServerOptions) {
1004 }684 }
1005685
1006 async function disposeEffectRuntime() {686 async function disposeEffectRuntime() {
1007 unsubscribeDomainEvents();687 if (subscriptionsScope && effectRuntime) {
1008 unsubscribeProviderEvents();688 await effectRuntime.runPromise(Scope.close(subscriptionsScope, Exit.void));
1009 unsubscribeDomainEvents = noop;689 subscriptionsScope = null;
1010 unsubscribeProviderEvents = noop;690 }
1011 orchestrationDomainEventLogger?.close();691 orchestrationDomainEventLogger?.close();
1012 orchestrationDomainEventLogger = undefined;692 orchestrationDomainEventLogger = undefined;
1013 orchestrationCommandLogger?.close();693 orchestrationCommandLogger?.close();