simplify layer composition

Julius Marminge committed 8 months ago

Commit 9ef01eb1 parent 2fa10f77 files · +760 −691Browse files

Unified · Split

7 files

apps/server/src/orchestration/Layer.test.ts +439 −0
apps/server/src/orchestration/Layer.ts +98 −110
apps/server/src/orchestration/Service.ts +25 −0
apps/server/src/orchestration/engine.test.ts +0 −388
apps/server/src/orchestration/layers.ts +0 −23
apps/server/src/orchestration/services.ts +0 −8
apps/server/src/wsServer.ts +198 −162

apps/server/src/orchestration/Layer.test.ts+439−0

@@ -0,0 +1,439 @@
	1	import fs from "node:fs";
	2	import os from "node:os";
	3	import path from "node:path";
	4
	5	import type { OrchestrationEvent } from "@t3tools/contracts";
	6	import { Effect } from "effect";
	7	import { afterEach, describe, expect, it } from "vitest";
	8
	9	import {
	10	OrchestrationEventRepository,
	11	type OrchestrationEventRepositoryShape,
	12	} from "../persistence/Services/OrchestrationEvents";
	13
	14	import { Layer, ManagedRuntime } from "effect";
	15
	16	import { makeSqlitePersistenceLive } from "../persistence/Layers/Sqlite";
	17
	18	import { OrchestrationEngineLive } from "./Layer";
	19	import { OrchestrationEngineService } from "./Service";
	20	import { OrchestrationEventRepositoryLive } from "../persistence/Layers/OrchestrationEvents";
	21
	22	const tempDirs: string[] = [];
	23
	24	function makeTempDir(prefix: string): string {
	25	const dir = fs.mkdtempSync(path.join(os.tmpdir(), prefix));
	26	tempDirs.push(dir);
	27	return dir;
	28	}
	29
	30	afterEach(() => {
	31	for (const dir of tempDirs.splice(0, tempDirs.length)) {
	32	fs.rmSync(dir, { recursive: true, force: true });
	33	}
	34	});
	35
	36	export async function createOrchestrationSystem(stateDir: string) {
	37	const dbPath = path.join(stateDir, "orchestration.sqlite");
	38	const orchestrationLayer = OrchestrationEngineLive.pipe(
	39	Layer.provide(OrchestrationEventRepositoryLive),
	40	Layer.provide(makeSqlitePersistenceLive(dbPath)),
	41	);
	42	const runtime = ManagedRuntime.make(orchestrationLayer);
	43	const engine = await runtime.runPromise(OrchestrationEngineService);
	44	return {
	45	engine,
	46	run: <A, E>(effect: Effect.Effect<A, E>) => runtime.runPromise(effect),
	47	dispose: () => runtime.dispose(),
	48	};
	49	}
	50
	51	describe("OrchestrationEngine", () => {
	52	it("replays to the same deterministic snapshot", async () => {
	53	const stateDir = makeTempDir("t3code-orchestration-");
	54	const createdAt = new Date().toISOString();
	55	const projectId = "project-1";
	56	const threadId = "thread-1";
	57
	58	const firstSystem = await createOrchestrationSystem(stateDir);
	59	const engineA = firstSystem.engine;
	60	await firstSystem.run(
	61	engineA.dispatch({
	62	type: "thread.create",
	63	commandId: "cmd-1",
	64	threadId,
	65	projectId,
	66	title: "Thread",
	67	model: "gpt-5-codex",
	68	branch: null,
	69	worktreePath: null,
	70	createdAt,
	71	}),
	72	);
	73	await firstSystem.run(
	74	engineA.dispatch({
	75	type: "message.send",
	76	commandId: "cmd-3",
	77	threadId,
	78	messageId: "msg-1",
	79	role: "user",
	80	text: "hello",
	81	streaming: false,
	82	createdAt,
	83	}),
	84	);
	85	const snapshotA = await firstSystem.run(engineA.getSnapshot());
	86	await firstSystem.dispose();
	87
	88	const secondSystem = await createOrchestrationSystem(stateDir);
	89	const engineB = secondSystem.engine;
	90	const snapshotB = await secondSystem.run(engineB.getSnapshot());
	91	expect(snapshotB).toEqual(snapshotA);
	92	await secondSystem.dispose();
	93	});
	94
	95	it("fans out read-model updates to subscribers", async () => {
	96	const stateDir = makeTempDir("t3code-orchestration-fanout-");
	97	const system = await createOrchestrationSystem(stateDir);
	98	const engine = system.engine;
	99	const updates: number[] = [];
	100	const unsubscribe = await system.run(
	101	engine.subscribeToReadModel((snapshot) => {
	102	updates.push(snapshot.sequence);
	103	}),
	104	);
	105	await system.run(
	106	engine.dispatch({
	107	type: "thread.create",
	108	commandId: "cmd-thread",
	109	threadId: "thread-2",
	110	projectId: "project-2",
	111	title: "fanout",
	112	model: "gpt-5-codex",
	113	branch: null,
	114	worktreePath: null,
	115	createdAt: new Date().toISOString(),
	116	}),
	117	);
	118	unsubscribe();
	119	expect(updates.length).toBeGreaterThan(0);
	120	await system.dispose();
	121	});
	122
	123	it("replays append-only events from sequence", async () => {
	124	const stateDir = makeTempDir("t3code-orchestration-replay-");
	125	const system = await createOrchestrationSystem(stateDir);
	126	const engine = system.engine;
	127	await system.run(
	128	engine.dispatch({
	129	type: "thread.create",
	130	commandId: "cmd-a",
	131	threadId: "thread-replay",
	132	projectId: "project-replay",
	133	title: "replay",
	134	model: "gpt-5-codex",
	135	branch: null,
	136	worktreePath: null,
	137	createdAt: new Date().toISOString(),
	138	}),
	139	);
	140	await system.run(
	141	engine.dispatch({
	142	type: "thread.delete",
	143	commandId: "cmd-b",
	144	threadId: "thread-replay",
	145	createdAt: new Date().toISOString(),
	146	}),
	147	);
	148	const events = await system.run(engine.replayEvents(0));
	149	expect(events.length).toBe(2);
	150	expect(events[0]?.type).toBe("thread.created");
	151	expect(events[1]?.type).toBe("thread.deleted");
	152	await system.dispose();
	153	});
	154
	155	it("stores completed turn summaries even when no files changed", async () => {
	156	const stateDir = makeTempDir("t3code-orchestration-turn-diff-");
	157	const firstSystem = await createOrchestrationSystem(stateDir);
	158	const engine = firstSystem.engine;
	159	const completedAt = new Date().toISOString();
	160	await firstSystem.run(
	161	engine.dispatch({
	162	type: "thread.create",
	163	commandId: "cmd-thread-turn-diff",
	164	threadId: "thread-turn-diff",
	165	projectId: "project-turn-diff",
	166	title: "Turn diff thread",
	167	model: "gpt-5-codex",
	168	branch: null,
	169	worktreePath: null,
	170	createdAt: completedAt,
	171	}),
	172	);
	173	await firstSystem.run(
	174	engine.dispatch({
	175	type: "thread.turnDiff.complete",
	176	commandId: "cmd-turn-diff-complete",
	177	threadId: "thread-turn-diff",
	178	turnId: "turn-1",
	179	completedAt,
	180	status: "completed",
	181	files: [],
	182	checkpointTurnCount: 1,
	183	createdAt: completedAt,
	184	}),
	185	);
	186
	187	const firstThread = (await firstSystem.run(engine.getSnapshot())).threads.find(
	188	(thread) => thread.id === "thread-turn-diff",
	189	);
	190	expect(firstThread?.turnDiffSummaries).toEqual([
	191	{
	192	turnId: "turn-1",
	193	completedAt,
	194	status: "completed",
	195	files: [],
	196	checkpointTurnCount: 1,
	197	},
	198	]);
	199	await firstSystem.dispose();
	200
	201	const secondSystem = await createOrchestrationSystem(stateDir);
	202	const restartedEngine = secondSystem.engine;
	203	const restartedThread = (await secondSystem.run(restartedEngine.getSnapshot())).threads.find(
	204	(thread) => thread.id === "thread-turn-diff",
	205	);
	206	expect(restartedThread?.turnDiffSummaries).toEqual([
	207	{
	208	turnId: "turn-1",
	209	completedAt,
	210	status: "completed",
	211	files: [],
	212	checkpointTurnCount: 1,
	213	},
	214	]);
	215	await secondSystem.dispose();
	216	});
	217
	218	it("reverts thread messages and turn summaries to a checkpoint", async () => {
	219	const stateDir = makeTempDir("t3code-orchestration-revert-");
	220	const firstSystem = await createOrchestrationSystem(stateDir);
	221	const engine = firstSystem.engine;
	222	const createdAt = new Date().toISOString();
	223	await firstSystem.run(
	224	engine.dispatch({
	225	type: "thread.create",
	226	commandId: "cmd-thread-revert",
	227	threadId: "thread-revert",
	228	projectId: "project-revert",
	229	title: "Revert thread",
	230	model: "gpt-5-codex",
	231	branch: null,
	232	worktreePath: null,
	233	createdAt,
	234	}),
	235	);
	236	await firstSystem.run(
	237	engine.dispatch({
	238	type: "message.send",
	239	commandId: "cmd-msg-1-user",
	240	threadId: "thread-revert",
	241	messageId: "user-1",
	242	role: "user",
	243	text: "first",
	244	streaming: false,
	245	createdAt,
	246	}),
	247	);
	248	await firstSystem.run(
	249	engine.dispatch({
	250	type: "message.send",
	251	commandId: "cmd-msg-1-assistant",
	252	threadId: "thread-revert",
	253	messageId: "assistant:turn-1",
	254	role: "assistant",
	255	text: "first-response",
	256	streaming: false,
	257	createdAt,
	258	}),
	259	);
	260	await firstSystem.run(
	261	engine.dispatch({
	262	type: "thread.turnDiff.complete",
	263	commandId: "cmd-turn-1-complete",
	264	threadId: "thread-revert",
	265	turnId: "turn-1",
	266	completedAt: createdAt,
	267	status: "completed",
	268	files: [],
	269	assistantMessageId: "assistant:turn-1",
	270	checkpointTurnCount: 1,
	271	createdAt,
	272	}),
	273	);
	274
	275	const createdAtSecond = new Date(Date.now() + 1_000).toISOString();
	276	await firstSystem.run(
	277	engine.dispatch({
	278	type: "message.send",
	279	commandId: "cmd-msg-2-user",
	280	threadId: "thread-revert",
	281	messageId: "user-2",
	282	role: "user",
	283	text: "second",
	284	streaming: false,
	285	createdAt: createdAtSecond,
	286	}),
	287	);
	288	await firstSystem.run(
	289	engine.dispatch({
	290	type: "message.send",
	291	commandId: "cmd-msg-2-assistant",
	292	threadId: "thread-revert",
	293	messageId: "assistant:turn-2",
	294	role: "assistant",
	295	text: "second-response",
	296	streaming: false,
	297	createdAt: createdAtSecond,
	298	}),
	299	);
	300	await firstSystem.run(
	301	engine.dispatch({
	302	type: "thread.turnDiff.complete",
	303	commandId: "cmd-turn-2-complete",
	304	threadId: "thread-revert",
	305	turnId: "turn-2",
	306	completedAt: createdAtSecond,
	307	status: "completed",
	308	files: [],
	309	assistantMessageId: "assistant:turn-2",
	310	checkpointTurnCount: 2,
	311	createdAt: createdAtSecond,
	312	}),
	313	);
	314
	315	await firstSystem.run(
	316	engine.dispatch({
	317	type: "thread.revert",
	318	commandId: "cmd-thread-revert-apply",
	319	threadId: "thread-revert",
	320	turnCount: 1,
	321	messageCount: 2,
	322	createdAt: createdAtSecond,
	323	}),
	324	);
	325
	326	const thread = (await firstSystem.run(engine.getSnapshot())).threads.find(
	327	(entry) => entry.id === "thread-revert",
	328	);
	329	expect(thread?.messages.map((message) => message.id)).toEqual(["user-1", "assistant:turn-1"]);
	330	expect(thread?.turnDiffSummaries.map((summary) => summary.turnId)).toEqual(["turn-1"]);
	331	expect(thread?.latestTurnId).toBe("turn-1");
	332	await firstSystem.dispose();
	333
	334	const secondSystem = await createOrchestrationSystem(stateDir);
	335	const restarted = secondSystem.engine;
	336	const restartedThread = (await secondSystem.run(restarted.getSnapshot())).threads.find(
	337	(entry) => entry.id === "thread-revert",
	338	);
	339	expect(restartedThread?.messages.map((message) => message.id)).toEqual([
	340	"user-1",
	341	"assistant:turn-1",
	342	]);
	343	expect(restartedThread?.turnDiffSummaries.map((summary) => summary.turnId)).toEqual(["turn-1"]);
	344	await secondSystem.dispose();
	345	});
	346
	347	it("allows stop to be called multiple times", async () => {
	348	const inMemoryStore: OrchestrationEventRepositoryShape = {
	349	append(event) {
	350	return Effect.succeed({ ...event, sequence: 1 });
	351	},
	352	readFromSequence() {
	353	return Effect.succeed([]);
	354	},
	355	readAll() {
	356	return Effect.succeed([]);
	357	},
	358	};
	359	const runtime = ManagedRuntime.make(
	360	OrchestrationEngineLive.pipe(
	361	Layer.provide(Layer.succeed(OrchestrationEventRepository, inMemoryStore)),
	362	),
	363	);
	364	await runtime.runPromise(OrchestrationEngineService);
	365	await runtime.dispose();
	366	await expect(runtime.dispose()).resolves.toBeUndefined();
	367	});
	368
	369	it("keeps processing queued commands after a storage failure", async () => {
	370	const events: OrchestrationEvent[] = [];
	371	let nextSequence = 1;
	372	let shouldFailFirstAppend = true;
	373
	374	const flakyStore: OrchestrationEventRepositoryShape = {
	375	append(event) {
	376	if (shouldFailFirstAppend) {
	377	shouldFailFirstAppend = false;
	378	return Effect.die(new Error("append failed"));
	379	}
	380	const savedEvent = {
	381	...event,
	382	sequence: nextSequence,
	383	} satisfies OrchestrationEvent;
	384	nextSequence += 1;
	385	events.push(savedEvent);
	386	return Effect.succeed(savedEvent);
	387	},
	388	readFromSequence(sequenceExclusive) {
	389	return Effect.succeed(events.filter((event) => event.sequence > sequenceExclusive));
	390	},
	391	readAll() {
	392	return Effect.succeed(events);
	393	},
	394	};
	395
	396	const runtime = ManagedRuntime.make(
	397	OrchestrationEngineLive.pipe(
	398	Layer.provide(Layer.succeed(OrchestrationEventRepository, flakyStore)),
	399	),
	400	);
	401	const engine = await runtime.runPromise(OrchestrationEngineService);
	402	const createdAt = new Date().toISOString();
	403
	404	await expect(
	405	runtime.runPromise(
	406	engine.dispatch({
	407	type: "thread.create",
	408	commandId: "cmd-flaky-1",
	409	threadId: "thread-flaky-fail",
	410	projectId: "project-flaky",
	411	title: "flaky-fail",
	412	model: "gpt-5-codex",
	413	branch: null,
	414	worktreePath: null,
	415	createdAt,
	416	}),
	417	),
	418	).rejects.toThrow("append failed");
	419
	420	const result = await runtime.runPromise(
	421	engine.dispatch({
	422	type: "thread.create",
	423	commandId: "cmd-flaky-2",
	424	threadId: "thread-flaky-ok",
	425	projectId: "project-flaky",
	426	title: "flaky-ok",
	427	model: "gpt-5-codex",
	428	branch: null,
	429	worktreePath: null,
	430	createdAt,
	431	}),
	432	);
	433
	434	expect(result.sequence).toBe(1);
	435	expect((await runtime.runPromise(engine.getSnapshot())).sequence).toBe(1);
	436
	437	await runtime.dispose();
	438	});
	439	});

apps/server/src/orchestration/Layer.tsrenamed from apps/server/src/orchestration/engine.ts+98−110

@@ -5,21 +5,22 @@ import type { · 4 unmodified lines
5	} from "@t3tools/contracts";	5	} from "@t3tools/contracts";
6	import { OrchestrationCommandSchema } from "@t3tools/contracts";	6	import { OrchestrationCommandSchema } from "@t3tools/contracts";
7	import {	7	import {
8	PubSub,	8	Cause,
9	Queue,	9	Deferred,
10	Schema,
11	Stream,
12	Effect,	10	Effect,
		11	Either,
13	Fiber,	12	Fiber,
		13	Layer,
		14	PubSub,
		15	Queue,
14	Runtime,	16	Runtime,
15	Either,	17	Schema,
16	Deferred,
17	Cause,
18	} from "effect";	18	} from "effect";
19		19
20	import { createLogger } from "../logger";	20	import { createLogger } from "../logger";
21	import type { OrchestrationEventRepositoryShape } from "../persistence/Services/OrchestrationEvents";	21	import { OrchestrationEventRepository } from "../persistence/Services/OrchestrationEvents";
22	import { createEmptyReadModel, reduceEvent } from "./reducer";	22	import { createEmptyReadModel, reduceEvent } from "./reducer";
		23	import { OrchestrationEngineService, type OrchestrationEngineShape } from "./Service";
23		24
24	type CommandEnvelope = {	25	type CommandEnvelope = {
25	command: OrchestrationCommand;	26	command: OrchestrationCommand;
@@ -168,45 +169,38 @@ function mapCommandToEvent(command: OrchestrationCommand): Omit<OrchestrationEve · 142 unmodified lines
168	}	169	}
169	}	170	}
170		171
171	export class OrchestrationEngine {	172	const decodeUnknownCommand = Schema.decodeUnknownEither(OrchestrationCommandSchema);
172	private readonly logger = createLogger("orchestration");
173	private readonly runtime = Runtime.defaultRuntime;
174	private readonly eventStore: OrchestrationEventRepositoryShape;
175		173
176	private readModel: OrchestrationReadModel;	174	export const makeOrchestrationEngine = Effect.gen(function* () {
177	private commandQueue: Queue.Queue<CommandEnvelope>;	175	const logger = createLogger("orchestration");
178	private readModelPubSub: PubSub.PubSub<OrchestrationReadModel>;	176	const eventStore = yield* OrchestrationEventRepository;
179	private eventPubSub: PubSub.PubSub<OrchestrationEvent>;	177	const runtime = Runtime.defaultRuntime;
180	private workerFiber: Fiber.RuntimeFiber<void, unknown> \| null = null;
181	private readonly readModelListeners = new Set<(snapshot: OrchestrationReadModel) => void>();
182	private readonly domainEventListeners = new Set<(event: OrchestrationEvent) => void>();
183		178
184	constructor(eventStore: OrchestrationEventRepositoryShape) {	179	let readModel = createEmptyReadModel(new Date().toISOString());
185	this.eventStore = eventStore;	180
186	this.readModel = createEmptyReadModel(new Date().toISOString());	181	const commandQueue = yield* Queue.unbounded<CommandEnvelope>();
187	this.commandQueue = Runtime.runSync(this.runtime)(Queue.unbounded<CommandEnvelope>());	182	const readModelPubSub = yield* PubSub.unbounded<OrchestrationReadModel>();
188	this.readModelPubSub = Runtime.runSync(this.runtime)(	183	const eventPubSub = yield* PubSub.unbounded<OrchestrationEvent>();
189	PubSub.unbounded<OrchestrationReadModel>(),
190	);
191	this.eventPubSub = Runtime.runSync(this.runtime)(PubSub.unbounded<OrchestrationEvent>());
192	}
193		184
194	private processEnvelope(envelope: CommandEnvelope): Effect.Effect<void> {	185	const readModelListeners = new Set<(snapshot: OrchestrationReadModel) => void>();
195	return Effect.gen(this, function* () {	186	const domainEventListeners = new Set<(event: OrchestrationEvent) => void>();
		187
		188	const processEnvelope = (envelope: CommandEnvelope): Effect.Effect<void> =>
		189	Effect.gen(function* () {
196	const eventBase = mapCommandToEvent(envelope.command);	190	const eventBase = mapCommandToEvent(envelope.command);
197	const savedEvent = yield* this.eventStore.append(eventBase);	191	const savedEvent = yield* eventStore.append(eventBase);
198	this.readModel = yield* reduceEvent(this.readModel, savedEvent);	192	readModel = yield* reduceEvent(readModel, savedEvent);
199		193
200	const snapshot = this.readModel;	194	const snapshot = readModel;
201	yield* Effect.all([	195	yield* Effect.all([
202	PubSub.publish(this.eventPubSub, savedEvent),	196	PubSub.publish(eventPubSub, savedEvent),
203	PubSub.publish(this.readModelPubSub, snapshot),	197	PubSub.publish(readModelPubSub, snapshot),
204	]);	198	]);
205		199
206	for (const listener of this.domainEventListeners) {	200	for (const listener of domainEventListeners) {
207	listener(savedEvent);	201	listener(savedEvent);
208	}	202	}
209	for (const listener of this.readModelListeners) {	203	for (const listener of readModelListeners) {
210	listener(snapshot);	204	listener(snapshot);
211	}	205	}
212		206
@@ -219,91 +213,85 @@ export class OrchestrationEngine { · 6 unmodified lines
219	).pipe(Effect.asVoid),	213	).pipe(Effect.asVoid),
220	),	214	),
221	);	215	);
222	}
223		216
224	private bootstrapReadModel(): Effect.Effect<void, unknown> {	217	const bootstrapReadModel: Effect.Effect<void, unknown> = Effect.gen(function* () {
225	return Effect.gen(this, function* () {	218	const existingEvents = yield* eventStore.readAll();
226	const existingEvents = yield* this.eventStore.readAll();	219	for (const event of existingEvents) {
227	for (const event of existingEvents) {	220	readModel = yield* reduceEvent(readModel, event);
228	this.readModel = yield* reduceEvent(this.readModel, event);
229	}
230	});
231	}
232
233	async start(): Promise<void> {
234	if (this.workerFiber) {
235	return;
236	}	221	}
		222	});
237		223
238	await Runtime.runPromise(this.runtime)(this.bootstrapReadModel());	224	yield* bootstrapReadModel;
239		225
240	const worker = Stream.fromQueue(this.commandQueue).pipe(	226	const worker = Effect.forever(Queue.take(commandQueue).pipe(Effect.flatMap(processEnvelope)));
241	Stream.runForEach((envelope) => this.processEnvelope(envelope)),	227	const workerFiber: Fiber.RuntimeFiber<void, unknown> = Runtime.runFork(runtime)(worker);
242	);	228	logger.info("orchestration engine started", {
		229	sequence: readModel.sequence,
		230	});
243		231
244	this.workerFiber = Runtime.runFork(this.runtime)(worker);	232	yield* Effect.addFinalizer(() => Fiber.interrupt(workerFiber).pipe(Effect.orDie));
245	this.logger.info("orchestration engine started", {
246	sequence: this.readModel.sequence,
247	});
248	}
249		233
250	async stop(): Promise<void> {	234	const getSnapshot: OrchestrationEngineShape["getSnapshot"] = () =>
251	if (this.workerFiber) {	235	Effect.sync((): OrchestrationReadModel => readModel);
252	await Runtime.runPromise(this.runtime)(Fiber.interrupt(this.workerFiber));
253	this.workerFiber = null;
254	}
255	}
256		236
257	getSnapshot(): OrchestrationReadModel {	237	const replayEvents: OrchestrationEngineShape["replayEvents"] = (fromSequenceExclusive) =>
258	return this.readModel;	238	eventStore.readFromSequence(fromSequenceExclusive);
259	}
260		239
261	async replayEvents(fromSequenceExclusive: number): Promise<OrchestrationEvent[]> {	240	const dispatch: OrchestrationEngineShape["dispatch"] = (command) =>
262	return Runtime.runPromise(this.runtime)(	241	Effect.gen(function* () {
263	this.eventStore.readFromSequence(fromSequenceExclusive),	242	const result = yield* Deferred.make<{ sequence: number }, Error>();
264	);	243	yield* Queue.offer(commandQueue, { command, result });
265	}	244	return yield* Deferred.await(result);
		245	}).pipe(Effect.mapError((error) => asError(error, "Queue offer failed")));
266		246
267	async dispatchUnknown(command: unknown): Promise<{ sequence: number }> {	247	const dispatchUnknown: OrchestrationEngineShape["dispatchUnknown"] = (command) =>
268	const decode = Schema.decodeUnknownEither(OrchestrationCommandSchema);	248	Effect.gen(function* () {
269	let payload: unknown = command;	249	let payload: unknown = command;
270	if (typeof command === "string") {	250	if (typeof command === "string") {
271	try {	251	try {
272	payload = JSON.parse(command) as unknown;	252	payload = JSON.parse(command) as unknown;
273	} catch {	253	} catch {
274	throw new Error("Invalid orchestration command: payload is not valid JSON");	254	return yield* Effect.fail(
		255	new Error("Invalid orchestration command: payload is not valid JSON"),
		256	);
		257	}
275	}	258	}
276	}	259	const decoded = decodeUnknownCommand(payload);
277	const decoded = decode(payload);	260	if (Either.isLeft(decoded)) {
278	if (Either.isLeft(decoded)) {	261	return yield* Effect.fail(
279	const issues = decoded.left.toString();	262	new Error(`Invalid orchestration command: ${decoded.left.toString()}`),
280	throw new Error(`Invalid orchestration command: ${issues}`);	263	);
281	}	264	}
282	return this.dispatch(decoded.right);	265	return yield* dispatch(decoded.right);
283	}	266	});
284		267
285	async dispatch(command: OrchestrationCommand): Promise<{ sequence: number }> {	268	const subscribeToReadModel: OrchestrationEngineShape["subscribeToReadModel"] = (callback) =>
286	const program = Effect.gen(this, function* () {	269	Effect.sync(() => {
287	const result = yield* Deferred.make<{ sequence: number }, Error>();	270	readModelListeners.add(callback);
288	yield* Queue.offer(this.commandQueue, { command, result });	271	return () => {
289	return yield* Deferred.await(result);	272	readModelListeners.delete(callback);
		273	};
290	});	274	});
291	return Runtime.runPromise(this.runtime)(program).catch((error) => {	275
292	throw asError(error, "Queue offer failed");	276	const subscribeToDomainEvents: OrchestrationEngineShape["subscribeToDomainEvents"] = (callback) =>
		277	Effect.sync(() => {
		278	domainEventListeners.add(callback);
		279	return () => {
		280	domainEventListeners.delete(callback);
		281	};
293	});	282	});
294	}
295		283
296	subscribeToReadModel(callback: (snapshot: OrchestrationReadModel) => void): () => void {	284	return {
297	this.readModelListeners.add(callback);	285	getSnapshot,
298	return () => {	286	replayEvents,
299	this.readModelListeners.delete(callback);	287	dispatchUnknown,
300	};	288	dispatch,
301	}	289	subscribeToReadModel,
		290	subscribeToDomainEvents,
		291	} satisfies OrchestrationEngineShape;
		292	});
302		293
303	subscribeToDomainEvents(callback: (event: OrchestrationEvent) => void): () => void {	294	export const OrchestrationEngineLive = Layer.scoped(
304	this.domainEventListeners.add(callback);	295	OrchestrationEngineService,
305	return () => {	296	makeOrchestrationEngine,
306	this.domainEventListeners.delete(callback);	297	);
307	};
308	}
309	}

apps/server/src/orchestration/Service.ts+25−0

@@ -0,0 +1,25 @@
	1	import type {
	2	OrchestrationCommand,
	3	OrchestrationEvent,
	4	OrchestrationReadModel,
	5	} from "@t3tools/contracts";
	6	import { Context } from "effect";
	7	import type { Effect } from "effect";
	8
	9	export interface OrchestrationEngineShape {
	10	readonly getSnapshot: () => Effect.Effect<OrchestrationReadModel>;
	11	readonly replayEvents: (fromSequenceExclusive: number) => Effect.Effect<OrchestrationEvent[]>;
	12	readonly dispatchUnknown: (command: unknown) => Effect.Effect<{ sequence: number }, Error>;
	13	readonly dispatch: (command: OrchestrationCommand) => Effect.Effect<{ sequence: number }, Error>;
	14	readonly subscribeToReadModel: (
	15	callback: (snapshot: OrchestrationReadModel) => void,
	16	) => Effect.Effect<() => void>;
	17	readonly subscribeToDomainEvents: (
	18	callback: (event: OrchestrationEvent) => void,
	19	) => Effect.Effect<() => void>;
	20	}
	21
	22	export class OrchestrationEngineService extends Context.Tag("orchestration/Engine")<
	23	OrchestrationEngineService,
	24	OrchestrationEngineShape
	25	>() {}

apps/server/src/orchestration/engine.test.ts+0−388

@@ -1,388 +0,0 @@
1	import fs from "node:fs";
2	import os from "node:os";
3	import path from "node:path";
4
5	import type { OrchestrationEvent } from "@t3tools/contracts";
6	import { Effect } from "effect";
7	import { afterEach, describe, expect, it } from "vitest";
8
9	import type { OrchestrationEventRepositoryShape } from "../persistence/Services/OrchestrationEvents";
10	import { OrchestrationEngine } from "./engine";
11
12	import { Layer, ManagedRuntime } from "effect";
13
14	import { makeSqlitePersistenceLive } from "../persistence/Layers/Sqlite";
15
16	import { OrchestrationLive } from "./layers";
17	import { OrchestrationEngineService } from "./services";
18
19	const tempDirs: string[] = [];
20
21	function makeTempDir(prefix: string): string {
22	const dir = fs.mkdtempSync(path.join(os.tmpdir(), prefix));
23	tempDirs.push(dir);
24	return dir;
25	}
26
27	afterEach(() => {
28	for (const dir of tempDirs.splice(0, tempDirs.length)) {
29	fs.rmSync(dir, { recursive: true, force: true });
30	}
31	});
32
33	export async function createOrchestrationSystem(stateDir: string) {
34	const dbPath = path.join(stateDir, "orchestration.sqlite");
35	const orchestrationLayer = OrchestrationLive.pipe(
36	Layer.provide(makeSqlitePersistenceLive(dbPath)),
37	);
38	const runtime = ManagedRuntime.make(orchestrationLayer);
39	const engine = await runtime.runPromise(OrchestrationEngineService);
40	return {
41	engine,
42	dispose: () => runtime.dispose(),
43	};
44	}
45
46	describe("OrchestrationEngine", () => {
47	it("replays to the same deterministic snapshot", async () => {
48	const stateDir = makeTempDir("t3code-orchestration-");
49	const createdAt = new Date().toISOString();
50	const projectId = "project-1";
51	const threadId = "thread-1";
52
53	const firstSystem = await createOrchestrationSystem(stateDir);
54	const engineA = firstSystem.engine;
55	await engineA.dispatch({
56	type: "thread.create",
57	commandId: "cmd-1",
58	threadId,
59	projectId,
60	title: "Thread",
61	model: "gpt-5-codex",
62	branch: null,
63	worktreePath: null,
64	createdAt,
65	});
66	await engineA.dispatch({
67	type: "message.send",
68	commandId: "cmd-3",
69	threadId,
70	messageId: "msg-1",
71	role: "user",
72	text: "hello",
73	streaming: false,
74	createdAt,
75	});
76	const snapshotA = engineA.getSnapshot();
77	await firstSystem.dispose();
78
79	const secondSystem = await createOrchestrationSystem(stateDir);
80	const engineB = secondSystem.engine;
81	const snapshotB = engineB.getSnapshot();
82	expect(snapshotB).toEqual(snapshotA);
83	await secondSystem.dispose();
84	});
85
86	it("fans out read-model updates to subscribers", async () => {
87	const stateDir = makeTempDir("t3code-orchestration-fanout-");
88	const system = await createOrchestrationSystem(stateDir);
89	const engine = system.engine;
90	const updates: number[] = [];
91	const unsubscribe = engine.subscribeToReadModel((snapshot) => {
92	updates.push(snapshot.sequence);
93	});
94	await engine.dispatch({
95	type: "thread.create",
96	commandId: "cmd-thread",
97	threadId: "thread-2",
98	projectId: "project-2",
99	title: "fanout",
100	model: "gpt-5-codex",
101	branch: null,
102	worktreePath: null,
103	createdAt: new Date().toISOString(),
104	});
105	unsubscribe();
106	expect(updates.length).toBeGreaterThan(0);
107	await system.dispose();
108	});
109
110	it("replays append-only events from sequence", async () => {
111	const stateDir = makeTempDir("t3code-orchestration-replay-");
112	const system = await createOrchestrationSystem(stateDir);
113	const engine = system.engine;
114	await engine.dispatch({
115	type: "thread.create",
116	commandId: "cmd-a",
117	threadId: "thread-replay",
118	projectId: "project-replay",
119	title: "replay",
120	model: "gpt-5-codex",
121	branch: null,
122	worktreePath: null,
123	createdAt: new Date().toISOString(),
124	});
125	await engine.dispatch({
126	type: "thread.delete",
127	commandId: "cmd-b",
128	threadId: "thread-replay",
129	createdAt: new Date().toISOString(),
130	});
131	const events = await engine.replayEvents(0);
132	expect(events.length).toBe(2);
133	expect(events[0]?.type).toBe("thread.created");
134	expect(events[1]?.type).toBe("thread.deleted");
135	await system.dispose();
136	});
137
138	it("stores completed turn summaries even when no files changed", async () => {
139	const stateDir = makeTempDir("t3code-orchestration-turn-diff-");
140	const firstSystem = await createOrchestrationSystem(stateDir);
141	const engine = firstSystem.engine;
142	const completedAt = new Date().toISOString();
143	await engine.dispatch({
144	type: "thread.create",
145	commandId: "cmd-thread-turn-diff",
146	threadId: "thread-turn-diff",
147	projectId: "project-turn-diff",
148	title: "Turn diff thread",
149	model: "gpt-5-codex",
150	branch: null,
151	worktreePath: null,
152	createdAt: completedAt,
153	});
154	await engine.dispatch({
155	type: "thread.turnDiff.complete",
156	commandId: "cmd-turn-diff-complete",
157	threadId: "thread-turn-diff",
158	turnId: "turn-1",
159	completedAt,
160	status: "completed",
161	files: [],
162	checkpointTurnCount: 1,
163	createdAt: completedAt,
164	});
165
166	const firstThread = engine
167	.getSnapshot()
168	.threads.find((thread) => thread.id === "thread-turn-diff");
169	expect(firstThread?.turnDiffSummaries).toEqual([
170	{
171	turnId: "turn-1",
172	completedAt,
173	status: "completed",
174	files: [],
175	checkpointTurnCount: 1,
176	},
177	]);
178	await firstSystem.dispose();
179
180	const secondSystem = await createOrchestrationSystem(stateDir);
181	const restartedEngine = secondSystem.engine;
182	const restartedThread = restartedEngine
183	.getSnapshot()
184	.threads.find((thread) => thread.id === "thread-turn-diff");
185	expect(restartedThread?.turnDiffSummaries).toEqual([
186	{
187	turnId: "turn-1",
188	completedAt,
189	status: "completed",
190	files: [],
191	checkpointTurnCount: 1,
192	},
193	]);
194	await secondSystem.dispose();
195	});
196
197	it("reverts thread messages and turn summaries to a checkpoint", async () => {
198	const stateDir = makeTempDir("t3code-orchestration-revert-");
199	const firstSystem = await createOrchestrationSystem(stateDir);
200	const engine = firstSystem.engine;
201	const createdAt = new Date().toISOString();
202	await engine.dispatch({
203	type: "thread.create",
204	commandId: "cmd-thread-revert",
205	threadId: "thread-revert",
206	projectId: "project-revert",
207	title: "Revert thread",
208	model: "gpt-5-codex",
209	branch: null,
210	worktreePath: null,
211	createdAt,
212	});
213	await engine.dispatch({
214	type: "message.send",
215	commandId: "cmd-msg-1-user",
216	threadId: "thread-revert",
217	messageId: "user-1",
218	role: "user",
219	text: "first",
220	streaming: false,
221	createdAt,
222	});
223	await engine.dispatch({
224	type: "message.send",
225	commandId: "cmd-msg-1-assistant",
226	threadId: "thread-revert",
227	messageId: "assistant:turn-1",
228	role: "assistant",
229	text: "first-response",
230	streaming: false,
231	createdAt,
232	});
233	await engine.dispatch({
234	type: "thread.turnDiff.complete",
235	commandId: "cmd-turn-1-complete",
236	threadId: "thread-revert",
237	turnId: "turn-1",
238	completedAt: createdAt,
239	status: "completed",
240	files: [],
241	assistantMessageId: "assistant:turn-1",
242	checkpointTurnCount: 1,
243	createdAt,
244	});
245
246	const createdAtSecond = new Date(Date.now() + 1_000).toISOString();
247	await engine.dispatch({
248	type: "message.send",
249	commandId: "cmd-msg-2-user",
250	threadId: "thread-revert",
251	messageId: "user-2",
252	role: "user",
253	text: "second",
254	streaming: false,
255	createdAt: createdAtSecond,
256	});
257	await engine.dispatch({
258	type: "message.send",
259	commandId: "cmd-msg-2-assistant",
260	threadId: "thread-revert",
261	messageId: "assistant:turn-2",
262	role: "assistant",
263	text: "second-response",
264	streaming: false,
265	createdAt: createdAtSecond,
266	});
267	await engine.dispatch({
268	type: "thread.turnDiff.complete",
269	commandId: "cmd-turn-2-complete",
270	threadId: "thread-revert",
271	turnId: "turn-2",
272	completedAt: createdAtSecond,
273	status: "completed",
274	files: [],
275	assistantMessageId: "assistant:turn-2",
276	checkpointTurnCount: 2,
277	createdAt: createdAtSecond,
278	});
279
280	await engine.dispatch({
281	type: "thread.revert",
282	commandId: "cmd-thread-revert-apply",
283	threadId: "thread-revert",
284	turnCount: 1,
285	messageCount: 2,
286	createdAt: createdAtSecond,
287	});
288
289	const thread = engine.getSnapshot().threads.find((entry) => entry.id === "thread-revert");
290	expect(thread?.messages.map((message) => message.id)).toEqual(["user-1", "assistant:turn-1"]);
291	expect(thread?.turnDiffSummaries.map((summary) => summary.turnId)).toEqual(["turn-1"]);
292	expect(thread?.latestTurnId).toBe("turn-1");
293	await firstSystem.dispose();
294
295	const secondSystem = await createOrchestrationSystem(stateDir);
296	const restarted = secondSystem.engine;
297	const restartedThread = restarted
298	.getSnapshot()
299	.threads.find((entry) => entry.id === "thread-revert");
300	expect(restartedThread?.messages.map((message) => message.id)).toEqual([
301	"user-1",
302	"assistant:turn-1",
303	]);
304	expect(restartedThread?.turnDiffSummaries.map((summary) => summary.turnId)).toEqual(["turn-1"]);
305	await secondSystem.dispose();
306	});
307
308	it("allows stop to be called multiple times", async () => {
309	const inMemoryStore: OrchestrationEventRepositoryShape = {
310	append(event) {
311	return Effect.succeed({ ...event, sequence: 1 });
312	},
313	readFromSequence() {
314	return Effect.succeed([]);
315	},
316	readAll() {
317	return Effect.succeed([]);
318	},
319	};
320	const engine = new OrchestrationEngine(inMemoryStore);
321	await engine.start();
322	await engine.stop();
323	await expect(engine.stop()).resolves.toBeUndefined();
324	});
325
326	it("keeps processing queued commands after a storage failure", async () => {
327	const events: OrchestrationEvent[] = [];
328	let nextSequence = 1;
329	let shouldFailFirstAppend = true;
330
331	const flakyStore: OrchestrationEventRepositoryShape = {
332	append(event) {
333	if (shouldFailFirstAppend) {
334	shouldFailFirstAppend = false;
335	return Effect.die(new Error("append failed"));
336	}
337	const savedEvent = {
338	...event,
339	sequence: nextSequence,
340	} satisfies OrchestrationEvent;
341	nextSequence += 1;
342	events.push(savedEvent);
343	return Effect.succeed(savedEvent);
344	},
345	readFromSequence(sequenceExclusive) {
346	return Effect.succeed(events.filter((event) => event.sequence > sequenceExclusive));
347	},
348	readAll() {
349	return Effect.succeed(events);
350	},
351	};
352
353	const engine = new OrchestrationEngine(flakyStore);
354	const createdAt = new Date().toISOString();
355	await engine.start();
356
357	await expect(
358	engine.dispatch({
359	type: "thread.create",
360	commandId: "cmd-flaky-1",
361	threadId: "thread-flaky-fail",
362	projectId: "project-flaky",
363	title: "flaky-fail",
364	model: "gpt-5-codex",
365	branch: null,
366	worktreePath: null,
367	createdAt,
368	}),
369	).rejects.toThrow("append failed");
370
371	const result = await engine.dispatch({
372	type: "thread.create",
373	commandId: "cmd-flaky-2",
374	threadId: "thread-flaky-ok",
375	projectId: "project-flaky",
376	title: "flaky-ok",
377	model: "gpt-5-codex",
378	branch: null,
379	worktreePath: null,
380	createdAt,
381	});
382
383	expect(result.sequence).toBe(1);
384	expect(engine.getSnapshot().sequence).toBe(1);
385
386	await engine.stop();
387	});
388	});

apps/server/src/orchestration/layers.ts+0−23

@@ -1,23 +0,0 @@
1	import { Effect, Layer } from "effect";
2
3	import { OrchestrationEventRepository } from "../persistence/Services/OrchestrationEvents";
4	import { OrchestrationEventRepositoryLive } from "../persistence/Layers/OrchestrationEvents";
5	import { OrchestrationEngine } from "./engine";
6	import { OrchestrationEngineService } from "./services";
7
8	export const OrchestrationEngineLive = Layer.scoped(
9	OrchestrationEngineService,
10	Effect.acquireRelease(
11	Effect.gen(function* () {
12	const eventRepository = yield* OrchestrationEventRepository;
13	const engine = new OrchestrationEngine(eventRepository);
14	yield* Effect.promise(() => engine.start());
15	return engine;
16	}),
17	(engine) => Effect.promise(() => engine.stop()),
18	),
19	);
20
21	export const OrchestrationLive = OrchestrationEngineLive.pipe(
22	Layer.provide(OrchestrationEventRepositoryLive),
23	);

apps/server/src/orchestration/services.ts+0−8

@@ -1,8 +0,0 @@
1	import { Context } from "effect";
2
3	import { OrchestrationEngine } from "./engine";
4
5	export class OrchestrationEngineService extends Context.Tag("orchestration/Engine")<
6	OrchestrationEngineService,
7	OrchestrationEngine
8	>() {}

apps/server/src/wsServer.ts+198−162

@@ -36,13 +36,13 @@ import { · 35 unmodified lines
36	import { TerminalManager } from "./terminalManager";	36	import { TerminalManager } from "./terminalManager";
37	import { loadResolvedKeybindingsConfig, upsertKeybindingRule } from "./keybindings";	37	import { loadResolvedKeybindingsConfig, upsertKeybindingRule } from "./keybindings";
38	import { searchWorkspaceEntries } from "./workspaceEntries";	38	import { searchWorkspaceEntries } from "./workspaceEntries";
39	import type { OrchestrationEngine } from "./orchestration/engine";	39	import { OrchestrationEngineLive } from "./orchestration/Layer";
40	import { OrchestrationLive } from "./orchestration/layers";	40	import { OrchestrationEngineService, type OrchestrationEngineShape } from "./orchestration/Service";
41	import { OrchestrationEngineService } from "./orchestration/services";
42	import { ProjectRepositoryLive } from "./persistence/Layers/Projects";	41	import { ProjectRepositoryLive } from "./persistence/Layers/Projects";
43	import { makeSqlitePersistenceLive } from "./persistence/Layers/Sqlite";	42	import { makeSqlitePersistenceLive } from "./persistence/Layers/Sqlite";
44	import { ProjectRepository, type ProjectRepositoryShape } from "./persistence/Services/Projects";	43	import { ProjectRepository, type ProjectRepositoryShape } from "./persistence/Services/Projects";
45	import assert from "node:assert";	44	import assert from "node:assert";
		45	import { OrchestrationEventRepositoryLive } from "./persistence/Layers/OrchestrationEvents";
46		46
47	const MIME_TYPES: Record<string, string> = {	47	const MIME_TYPES: Record<string, string> = {
48	".html": "text/html; charset=utf-8",	48	".html": "text/html; charset=utf-8",
@@ -136,7 +136,7 @@ export function createServer(options: ServerOptions) { · 87 unmodified lines
136	unknown	136	unknown
137	> \| null = null;	137	> \| null = null;
138	let projectRepository: ProjectRepositoryShape \| null = null;	138	let projectRepository: ProjectRepositoryShape \| null = null;
139	let orchestrationEngine: OrchestrationEngine \| null = null;	139	let orchestrationEngine: OrchestrationEngineShape \| null = null;
140	const clients = new Set<WebSocket>();	140	const clients = new Set<WebSocket>();
141	const logger = createLogger("ws");	141	const logger = createLogger("ws");
142	const logWebSocketEvents =	142	const logWebSocketEvents =
@@ -169,141 +169,157 @@ export function createServer(options: ServerOptions) { · 26 unmodified lines
169	return effectRuntime;	169	return effectRuntime;
170	};	170	};
171		171
172	function attachOrchestrationSubscriptions() {	172	async function attachOrchestrationSubscriptions(
173	const orchestrationEngine = getOrchestrationEngine();	173	runtime: ManagedRuntime.ManagedRuntime<ProjectRepository \| OrchestrationEngineService, unknown>,
174		174	orchestrationEngine: OrchestrationEngineShape,
175	unsubscribeReadModel = orchestrationEngine.subscribeToReadModel((snapshot) => {	175	) {
176	const push: WsPush = {	176	unsubscribeReadModel = await runtime.runPromise(
177	type: "push",	177	orchestrationEngine.subscribeToReadModel((snapshot) => {
178	channel: ORCHESTRATION_WS_CHANNELS.readModel,	178	const push: WsPush = {
179	data: {	179	type: "push",
180	sequence: snapshot.sequence,	180	channel: ORCHESTRATION_WS_CHANNELS.readModel,
181	snapshot,	181	data: {
182	} satisfies OrchestrationReadModelPush,	182	sequence: snapshot.sequence,
183	};	183	snapshot,
184	const message = JSON.stringify(push);	184	} satisfies OrchestrationReadModelPush,
185	let recipients = 0;	185	};
186	for (const client of clients) {	186	const message = JSON.stringify(push);
187	if (client.readyState === client.OPEN) {	187	let recipients = 0;
188	client.send(message);	188	for (const client of clients) {
189	recipients += 1;	189	if (client.readyState === client.OPEN) {
		190	client.send(message);
		191	recipients += 1;
		192	}
190	}	193	}
191	}	194	logOutgoingPush(push, recipients);
192	logOutgoingPush(push, recipients);	195	}),
193	});	196	);
194		197
195	unsubscribeDomainEvents = orchestrationEngine.subscribeToDomainEvents((event) => {	198	unsubscribeDomainEvents = await runtime.runPromise(
196	const push: WsPush = {	199	orchestrationEngine.subscribeToDomainEvents((event) => {
197	type: "push",	200	const push: WsPush = {
198	channel: ORCHESTRATION_WS_CHANNELS.domainEvent,	201	type: "push",
199	data: event,	202	channel: ORCHESTRATION_WS_CHANNELS.domainEvent,
200	};	203	data: event,
201	const message = JSON.stringify(push);	204	};
202	let recipients = 0;	205	const message = JSON.stringify(push);
203	for (const client of clients) {	206	let recipients = 0;
204	if (client.readyState === client.OPEN) {	207	for (const client of clients) {
205	client.send(message);	208	if (client.readyState === client.OPEN) {
206	recipients += 1;	209	client.send(message);
		210	recipients += 1;
		211	}
207	}	212	}
208	}	213	logOutgoingPush(push, recipients);
209	logOutgoingPush(push, recipients);	214	}),
210	});	215	);
211	}	216	}
212		217
213	// Forward provider events to all connected WebSocket clients	218	// Forward provider events to all connected WebSocket clients
214	providerManager.on("event", (event) => {	219	providerManager.on("event", (event) => {
215	void (async () => {	220	void (async () => {
216	const liveOrchestrationEngine = orchestrationEngine;	221	const liveOrchestrationEngine = orchestrationEngine;
217	if (!liveOrchestrationEngine) {	222	const runtime = effectRuntime;
		223	if (!liveOrchestrationEngine \|\| !runtime) {
218	return;	224	return;
219	}	225	}
220	const snapshot = liveOrchestrationEngine.getSnapshot();	226	const snapshot = await runtime.runPromise(liveOrchestrationEngine.getSnapshot());
221	const thread = snapshot.threads.find((entry) => entry.session?.sessionId === event.sessionId);	227	const thread = snapshot.threads.find((entry) => entry.session?.sessionId === event.sessionId);
222	if (!thread) return;	228	if (!thread) return;
223	const now = event.createdAt;	229	const now = event.createdAt;
224	if (event.method === "turn/started" \|\| event.method === "turn/completed") {	230	if (event.method === "turn/started" \|\| event.method === "turn/completed") {
225	await liveOrchestrationEngine.dispatch({	231	await runtime.runPromise(
226	type: "thread.session",	232	liveOrchestrationEngine.dispatch({
227	commandId: crypto.randomUUID(),	233	type: "thread.session",
228	threadId: thread.id,	234	commandId: crypto.randomUUID(),
229	session: {
230	sessionId: event.sessionId,
231	provider: event.provider,
232	status: event.method === "turn/started" ? "running" : "ready",
233	threadId: thread.id,	235	threadId: thread.id,
234	activeTurnId: event.method === "turn/started" ? (event.turnId ?? null) : null,	236	session: {
235	createdAt: thread.session?.createdAt ?? now,	237	sessionId: event.sessionId,
236	updatedAt: now,	238	provider: event.provider,
237	lastError: null,	239	status: event.method === "turn/started" ? "running" : "ready",
238	},	240	threadId: thread.id,
239	createdAt: now,	241	activeTurnId: event.method === "turn/started" ? (event.turnId ?? null) : null,
240	});	242	createdAt: thread.session?.createdAt ?? now,
		243	updatedAt: now,
		244	lastError: null,
		245	},
		246	createdAt: now,
		247	}),
		248	);
241	}	249	}
242	if (event.textDelta && event.textDelta.length > 0) {	250	if (event.textDelta && event.textDelta.length > 0) {
243	const assistantMessageId = `assistant:${event.turnId ?? event.itemId ?? event.sessionId}`;	251	const assistantMessageId = `assistant:${event.turnId ?? event.itemId ?? event.sessionId}`;
244	await liveOrchestrationEngine.dispatch({	252	await runtime.runPromise(
245	type: "message.send",	253	liveOrchestrationEngine.dispatch({
246	commandId: crypto.randomUUID(),	254	type: "message.send",
247	threadId: thread.id,	255	commandId: crypto.randomUUID(),
248	messageId: assistantMessageId,	256	threadId: thread.id,
249	role: "assistant",	257	messageId: assistantMessageId,
250	text: event.textDelta,	258	role: "assistant",
251	streaming: true,	259	text: event.textDelta,
252	createdAt: now,	260	streaming: true,
253	});	261	createdAt: now,
		262	}),
		263	);
254	}	264	}
255	if (event.method === "turn/completed") {	265	if (event.method === "turn/completed") {
256	const assistantMessageId = `assistant:${event.turnId ?? event.itemId ?? event.sessionId}`;	266	const assistantMessageId = `assistant:${event.turnId ?? event.itemId ?? event.sessionId}`;
257	await liveOrchestrationEngine.dispatch({	267	await runtime.runPromise(
258	type: "message.send",	268	liveOrchestrationEngine.dispatch({
259	commandId: crypto.randomUUID(),	269	type: "message.send",
260	threadId: thread.id,	270	commandId: crypto.randomUUID(),
261	messageId: assistantMessageId,	271	threadId: thread.id,
262	role: "assistant",	272	messageId: assistantMessageId,
263	text: "",	273	role: "assistant",
264	streaming: false,	274	text: "",
265	createdAt: now,	275	streaming: false,
266	});	276	createdAt: now,
		277	}),
		278	);
267	}	279	}
268	if (event.method === "checkpoint/captured") {	280	if (event.method === "checkpoint/captured") {
269	const payload = asObject(event.payload);	281	const payload = asObject(event.payload);
270	const turnId = event.turnId ?? asString(payload?.turnId);	282	const turnId = event.turnId ?? asString(payload?.turnId);
271	if (turnId) {	283	if (turnId) {
272	await liveOrchestrationEngine.dispatch({	284	await runtime.runPromise(
273	type: "thread.turnDiff.complete",	285	liveOrchestrationEngine.dispatch({
274	commandId: crypto.randomUUID(),	286	type: "thread.turnDiff.complete",
275	threadId: thread.id,	287	commandId: crypto.randomUUID(),
276	turnId,	288	threadId: thread.id,
277	completedAt: now,	289	turnId,
278	...(asString(payload?.status) !== null	290	completedAt: now,
279	? { status: asString(payload?.status) ?? undefined }	291	...(asString(payload?.status) !== null
280	: {}),	292	? { status: asString(payload?.status) ?? undefined }
281	files: [],	293	: {}),
282	assistantMessageId: `assistant:${turnId}`,	294	files: [],
283	...(asNumber(payload?.turnCount) !== null	295	assistantMessageId: `assistant:${turnId}`,
284	? { checkpointTurnCount: asNumber(payload?.turnCount) ?? undefined }	296	...(asNumber(payload?.turnCount) !== null
285	: {}),	297	? { checkpointTurnCount: asNumber(payload?.turnCount) ?? undefined }
286	createdAt: now,	298	: {}),
287	});	299	createdAt: now,
		300	}),
		301	);
288	}	302	}
289	}	303	}
290	if (event.kind === "error" && event.message) {	304	if (event.kind === "error" && event.message) {
291	await liveOrchestrationEngine.dispatch({	305	await runtime.runPromise(
292	type: "thread.session",	306	liveOrchestrationEngine.dispatch({
293	commandId: crypto.randomUUID(),	307	type: "thread.session",
294	threadId: thread.id,	308	commandId: crypto.randomUUID(),
295	session: {
296	sessionId: event.sessionId,
297	provider: event.provider,
298	status: "error",
299	threadId: thread.id,	309	threadId: thread.id,
300	activeTurnId: event.turnId ?? null,	310	session: {
301	createdAt: thread.session?.createdAt ?? now,	311	sessionId: event.sessionId,
302	updatedAt: now,	312	provider: event.provider,
303	lastError: event.message,	313	status: "error",
304	},	314	threadId: thread.id,
305	createdAt: now,	315	activeTurnId: event.turnId ?? null,
306	});	316	createdAt: thread.session?.createdAt ?? now,
		317	updatedAt: now,
		318	lastError: event.message,
		319	},
		320	createdAt: now,
		321	}),
		322	);
307	}	323	}
308	})().catch(() => undefined);	324	})().catch(() => undefined);
309	});	325	});
@@ -410,6 +426,7 @@ export function createServer(options: ServerOptions) { · 100 unmodified lines
410	wss.on("connection", (ws) => {	426	wss.on("connection", (ws) => {
411	clients.add(ws);	427	clients.add(ws);
412	const orchestrationEngine = getOrchestrationEngine();	428	const orchestrationEngine = getOrchestrationEngine();
		429	const runtime = getEffectRuntime();
413		430
414	// Send welcome message with project info	431	// Send welcome message with project info
415	const segments = cwd.split(/[/\\]/).filter(Boolean);	432	const segments = cwd.split(/[/\\]/).filter(Boolean);
@@ -423,16 +440,24 @@ export function createServer(options: ServerOptions) { · 7 unmodified lines
423	logOutgoingPush(welcome, 1);	440	logOutgoingPush(welcome, 1);
424	ws.send(JSON.stringify(welcome));	441	ws.send(JSON.stringify(welcome));
425		442
426	const snapshotPush: WsPush = {	443	void runtime
427	type: "push",	444	.runPromise(orchestrationEngine.getSnapshot())
428	channel: ORCHESTRATION_WS_CHANNELS.readModel,	445	.then((snapshot) => {
429	data: {	446	if (ws.readyState !== ws.OPEN) {
430	sequence: orchestrationEngine.getSnapshot().sequence,	447	return;
431	snapshot: orchestrationEngine.getSnapshot(),	448	}
432	} satisfies OrchestrationReadModelPush,	449	const snapshotPush: WsPush = {
433	};	450	type: "push",
434	logOutgoingPush(snapshotPush, 1);	451	channel: ORCHESTRATION_WS_CHANNELS.readModel,
435	ws.send(JSON.stringify(snapshotPush));	452	data: {
		453	sequence: snapshot.sequence,
		454	snapshot,
		455	} satisfies OrchestrationReadModelPush,
		456	};
		457	logOutgoingPush(snapshotPush, 1);
		458	ws.send(JSON.stringify(snapshotPush));
		459	})
		460	.catch(() => undefined);
436		461
437	ws.on("message", (raw) => {	462	ws.on("message", (raw) => {
438	void handleMessage(ws, raw);	463	void handleMessage(ws, raw);
@@ -509,32 +534,36 @@ export function createServer(options: ServerOptions) { · 70 unmodified lines
509	const params = request.params as { sessionId?: string };	534	const params = request.params as { sessionId?: string };
510	const sessionId = params.sessionId;	535	const sessionId = params.sessionId;
511	if (typeof sessionId === "string") {	536	if (typeof sessionId === "string") {
512	const snapshot = orchestrationEngine.getSnapshot();	537	const snapshot = await effectRuntime.runPromise(orchestrationEngine.getSnapshot());
513	const thread = snapshot.threads.find((entry) => entry.session?.sessionId === sessionId);	538	const thread = snapshot.threads.find((entry) => entry.session?.sessionId === sessionId);
514	if (thread) {	539	if (thread) {
515	const now = new Date().toISOString();	540	const now = new Date().toISOString();
516	await orchestrationEngine.dispatch({	541	await effectRuntime.runPromise(
517	type: "thread.revert",	542	orchestrationEngine.dispatch({
518	commandId: crypto.randomUUID(),	543	type: "thread.revert",
519	threadId: thread.id,
520	turnCount: result.turnCount,
521	messageCount: result.messageCount,
522	createdAt: now,
523	});
524	if (thread.session) {
525	await orchestrationEngine.dispatch({
526	type: "thread.session",
527	commandId: crypto.randomUUID(),	544	commandId: crypto.randomUUID(),
528	threadId: thread.id,	545	threadId: thread.id,
529	session: {	546	turnCount: result.turnCount,
530	...thread.session,	547	messageCount: result.messageCount,
531	status: "ready",
532	activeTurnId: null,
533	updatedAt: now,
534	lastError: null,
535	},
536	createdAt: now,	548	createdAt: now,
537	});	549	}),
		550	);
		551	if (thread.session) {
		552	await effectRuntime.runPromise(
		553	orchestrationEngine.dispatch({
		554	type: "thread.session",
		555	commandId: crypto.randomUUID(),
		556	threadId: thread.id,
		557	session: {
		558	...thread.session,
		559	status: "ready",
		560	activeTurnId: null,
		561	updatedAt: now,
		562	lastError: null,
		563	},
		564	createdAt: now,
		565	}),
		566	);
538	}	567	}
539	}	568	}
540	}	569	}
@@ -605,16 +634,18 @@ export function createServer(options: ServerOptions) { · 64 unmodified lines
605	const projects = await effectRuntime.runPromise(projectRepository.list());	634	const projects = await effectRuntime.runPromise(projectRepository.list());
606	const project = projects.find((entry) => entry.cwd === params.cwd);	635	const project = projects.find((entry) => entry.cwd === params.cwd);
607	if (project) {	636	if (project) {
608	await orchestrationEngine.dispatch({	637	await effectRuntime.runPromise(
609	type: "git.readModel.upsert",	638	orchestrationEngine.dispatch({
610	commandId: crypto.randomUUID(),	639	type: "git.readModel.upsert",
611	projectId: project.id,	640	commandId: crypto.randomUUID(),
612	branch: status.branch,	641	projectId: project.id,
613	hasWorkingTreeChanges: status.hasWorkingTreeChanges,	642	branch: status.branch,
614	aheadCount: status.aheadCount,	643	hasWorkingTreeChanges: status.hasWorkingTreeChanges,
615	behindCount: status.behindCount,	644	aheadCount: status.aheadCount,
616	createdAt: new Date().toISOString(),	645	behindCount: status.behindCount,
617	});	646	createdAt: new Date().toISOString(),
		647	}),
		648	);
618	}	649	}
619	return status;	650	return status;
620	}	651	}
@@ -677,19 +708,21 @@ export function createServer(options: ServerOptions) { · 56 unmodified lines
677	};	708	};
678		709
679	case ORCHESTRATION_WS_METHODS.getSnapshot:	710	case ORCHESTRATION_WS_METHODS.getSnapshot:
680	return orchestrationEngine.getSnapshot();	711	return effectRuntime.runPromise(orchestrationEngine.getSnapshot());
681		712
682	case ORCHESTRATION_WS_METHODS.dispatchCommand:	713	case ORCHESTRATION_WS_METHODS.dispatchCommand:
683	return orchestrationEngine.dispatchUnknown(request.params);	714	return effectRuntime.runPromise(orchestrationEngine.dispatchUnknown(request.params));
684		715
685	case ORCHESTRATION_WS_METHODS.replayEvents:	716	case ORCHESTRATION_WS_METHODS.replayEvents:
686	return orchestrationEngine.replayEvents(	717	return effectRuntime.runPromise(
687	Math.max(	718	orchestrationEngine.replayEvents(
688	0,	719	Math.max(
689	Math.floor(	720	0,
690	Number(	721	Math.floor(
691	(request.params as { fromSequenceExclusive?: number } \| undefined)	722	Number(
692	?.fromSequenceExclusive ?? 0,	723	(request.params as { fromSequenceExclusive?: number } \| undefined)
		724	?.fromSequenceExclusive ?? 0,
		725	),
693	),	726	),
694	),	727	),
695	),	728	),
@@ -703,11 +736,14 @@ export function createServer(options: ServerOptions) { · 7 unmodified lines
703	async function createEffectRuntime() {	736	async function createEffectRuntime() {
704	const dbPath = path.join(resolvedStateDir, "orchestration.sqlite");	737	const dbPath = path.join(resolvedStateDir, "orchestration.sqlite");
705	const persistenceLayer = makeSqlitePersistenceLive(dbPath);	738	const persistenceLayer = makeSqlitePersistenceLive(dbPath);
706	const runtime = ManagedRuntime.make(	739	const orchestrationLayer = Layer.provide(
707	Layer.mergeAll(OrchestrationLive, ProjectRepositoryLive).pipe(	740	OrchestrationEngineLive,
708	Layer.provide(persistenceLayer),	741	OrchestrationEventRepositoryLive,
709	),	742	);
		743	const layer = Layer.mergeAll(orchestrationLayer, ProjectRepositoryLive).pipe(
		744	Layer.provide(persistenceLayer),
710	);	745	);
		746	const runtime = ManagedRuntime.make(layer);
711		747
712	try {	748	try {
713	const [nextOrchestrationEngine, repository] = await Promise.all([	749	const [nextOrchestrationEngine, repository] = await Promise.all([
@@ -717,7 +753,7 @@ export function createServer(options: ServerOptions) { · 3 unmodified lines
717	orchestrationEngine = nextOrchestrationEngine;	753	orchestrationEngine = nextOrchestrationEngine;
718	projectRepository = repository;	754	projectRepository = repository;
719	await runtime.runPromise(repository.pruneMissing());	755	await runtime.runPromise(repository.pruneMissing());
720	attachOrchestrationSubscriptions();	756	await attachOrchestrationSubscriptions(runtime, nextOrchestrationEngine);
721	} catch (error) {	757	} catch (error) {
722	await runtime.dispose().catch(() => undefined);	758	await runtime.dispose().catch(() => undefined);
723	throw error;	759	throw error;