History

Add provider orchestrator and Codex adapter

- ProviderOrchestrator manages provider sessions and message dispatch - CodexProviderAdapter wraps Codex app-server integration - Refactor wsServer to use new provider abstraction

Julius Marminge committed 8 months ago

Commit dbbc2251 parent 4ce3c3b4 files · +667 −14Browse files

Unified · Split

4 files
  1. apps/server/src/effect/providers/codexProviderAdapter.ts +97 −0
  2. apps/server/src/effect/providers/providerAdapter.ts +37 −0
  3. apps/server/src/effect/providers/providerOrchestrator.ts +518 −0
  4. apps/server/src/wsServer.ts +15 −14
apps/server/src/effect/providers/codexProviderAdapter.ts+97−0
@@ -0,0 +1,97 @@
1import {
2 providerInterruptTurnInputSchema,
3 providerRespondToRequestInputSchema,
4 providerSendTurnInputSchema,
5 providerSessionStartInputSchema,
6 providerStopSessionInputSchema,
7} from "@t3tools/contracts";
8import type {
9 ProviderEvent,
10 ProviderInterruptTurnInput,
11 ProviderRespondToRequestInput,
12 ProviderSendTurnInput,
13 ProviderSession,
14 ProviderSessionStartInput,
15 ProviderStopSessionInput,
16 ProviderTurnStartResult,
17} from "@t3tools/contracts";
18
19import { CodexAppServerManager } from "../../codexAppServerManager";
20import type { ProviderAdapter, ProviderThreadSnapshot } from "./providerAdapter";
21
22export class CodexProviderAdapter implements ProviderAdapter {
23 readonly kind = "codex" as const;
24 private readonly manager = new CodexAppServerManager();
25
26 onEvent(listener: (event: ProviderEvent) => void): () => void {
27 this.manager.on("event", listener);
28 return () => {
29 this.manager.off("event", listener);
30 };
31 }
32
33 hasSession(sessionId: string): boolean {
34 return this.manager.hasSession(sessionId);
35 }
36
37 startSession(input: ProviderSessionStartInput): Promise<ProviderSession> {
38 const parsed = providerSessionStartInputSchema.parse(input);
39 if (parsed.provider !== this.kind) {
40 throw new Error(`Provider '${parsed.provider}' is not implemented by CodexProviderAdapter.`);
41 }
42 return this.manager.startSession(parsed);
43 }
44
45 sendTurn(input: ProviderSendTurnInput): Promise<ProviderTurnStartResult> {
46 return this.manager.sendTurn(providerSendTurnInputSchema.parse(input));
47 }
48
49 interruptTurn(input: ProviderInterruptTurnInput): Promise<void> {
50 const parsed = providerInterruptTurnInputSchema.parse(input);
51 return this.manager.interruptTurn(parsed.sessionId, parsed.turnId);
52 }
53
54 respondToRequest(input: ProviderRespondToRequestInput): Promise<void> {
55 const parsed = providerRespondToRequestInputSchema.parse(input);
56 return this.manager.respondToRequest(parsed.sessionId, parsed.requestId, parsed.decision);
57 }
58
59 stopSession(input: ProviderStopSessionInput): void {
60 const parsed = providerStopSessionInputSchema.parse(input);
61 this.manager.stopSession(parsed.sessionId);
62 }
63
64 listSessions(): ProviderSession[] {
65 return this.manager.listSessions();
66 }
67
68 async readThread(sessionId: string): Promise<ProviderThreadSnapshot> {
69 const snapshot = await this.manager.readThread(sessionId);
70 return {
71 threadId: snapshot.threadId,
72 turns: snapshot.turns.map((turn) => ({
73 id: turn.id,
74 items: turn.items,
75 })),
76 };
77 }
78
79 async rollbackThread(sessionId: string, numTurns: number): Promise<ProviderThreadSnapshot> {
80 const snapshot = await this.manager.rollbackThread(sessionId, numTurns);
81 return {
82 threadId: snapshot.threadId,
83 turns: snapshot.turns.map((turn) => ({
84 id: turn.id,
85 items: turn.items,
86 })),
87 };
88 }
89
90 stopAll(): void {
91 this.manager.stopAll();
92 }
93
94 dispose(): void {
95 this.manager.stopAll();
96 }
97}
apps/server/src/effect/providers/providerAdapter.ts+37−0
@@ -0,0 +1,37 @@
1import type {
2 ProviderEvent,
3 ProviderInterruptTurnInput,
4 ProviderKind,
5 ProviderRespondToRequestInput,
6 ProviderSendTurnInput,
7 ProviderSession,
8 ProviderSessionStartInput,
9 ProviderStopSessionInput,
10 ProviderTurnStartResult,
11} from "@t3tools/contracts";
12
13export interface ProviderThreadTurnSnapshot {
14 id: string;
15 items: unknown[];
16}
17
18export interface ProviderThreadSnapshot {
19 threadId: string;
20 turns: ProviderThreadTurnSnapshot[];
21}
22
23export interface ProviderAdapter {
24 readonly kind: ProviderKind;
25 onEvent(listener: (event: ProviderEvent) => void): () => void;
26 hasSession(sessionId: string): boolean;
27 startSession(input: ProviderSessionStartInput): Promise<ProviderSession>;
28 sendTurn(input: ProviderSendTurnInput): Promise<ProviderTurnStartResult>;
29 interruptTurn(input: ProviderInterruptTurnInput): Promise<void>;
30 respondToRequest(input: ProviderRespondToRequestInput): Promise<void>;
31 stopSession(input: ProviderStopSessionInput): void;
32 listSessions(): ProviderSession[];
33 readThread(sessionId: string): Promise<ProviderThreadSnapshot>;
34 rollbackThread(sessionId: string, numTurns: number): Promise<ProviderThreadSnapshot>;
35 stopAll(): void;
36 dispose(): void;
37}
apps/server/src/effect/providers/providerOrchestrator.ts+518−0
@@ -0,0 +1,518 @@
1import { EventEmitter } from "node:events";
2import { randomUUID } from "node:crypto";
3
4import type {
5 ProviderCheckpoint,
6 ProviderEvent,
7 ProviderGetCheckpointDiffInput,
8 ProviderGetCheckpointDiffResult,
9 ProviderInterruptTurnInput,
10 ProviderKind,
11 ProviderListCheckpointsInput,
12 ProviderListCheckpointsResult,
13 ProviderRevertToCheckpointInput,
14 ProviderRevertToCheckpointResult,
15 ProviderRespondToRequestInput,
16 ProviderSendTurnInput,
17 ProviderSession,
18 ProviderSessionStartInput,
19 ProviderStopSessionInput,
20 ProviderTurnStartResult,
21} from "@t3tools/contracts";
22import {
23 providerGetCheckpointDiffInputSchema,
24 providerInterruptTurnInputSchema,
25 providerListCheckpointsInputSchema,
26 providerRevertToCheckpointInputSchema,
27 providerRespondToRequestInputSchema,
28 providerSendTurnInputSchema,
29 providerSessionStartInputSchema,
30 providerStopSessionInputSchema,
31} from "@t3tools/contracts";
32
33import { FilesystemCheckpointStore } from "../../filesystemCheckpointStore";
34import type { ProviderAdapter, ProviderThreadTurnSnapshot } from "./providerAdapter";
35
36export interface ProviderOrchestratorEvents {
37 event: [event: ProviderEvent];
38}
39
40export class ProviderOrchestrator extends EventEmitter<ProviderOrchestratorEvents> {
41 private readonly adapters: Map<ProviderKind, ProviderAdapter>;
42 private readonly filesystemCheckpointStore = new FilesystemCheckpointStore();
43 private readonly sessionToProvider = new Map<string, ProviderKind>();
44 private readonly sessionCheckpointCwds = new Map<string, string>();
45 private readonly filesystemLocks = new Map<string, Promise<void>>();
46 private readonly unsubscribers: Array<() => void>;
47
48 constructor(adapters: ProviderAdapter[]) {
49 super();
50 this.adapters = new Map(adapters.map((adapter) => [adapter.kind, adapter] as const));
51 this.unsubscribers = adapters.map((adapter) =>
52 adapter.onEvent((event) => {
53 this.sessionToProvider.set(event.sessionId, event.provider);
54 this.emit("event", event);
55 this.maybeCaptureFilesystemCheckpoint(event);
56 }),
57 );
58 }
59
60 async startSession(raw: ProviderSessionStartInput): Promise<ProviderSession> {
61 const input = providerSessionStartInputSchema.parse(raw);
62 const provider = input.provider ?? "codex";
63 const adapter = this.requireAdapter(provider);
64 const session = await adapter.startSession(input);
65 this.sessionToProvider.set(session.sessionId, session.provider);
66 await this.initializeFilesystemCheckpointing(adapter, session, input.cwd).catch((error) => {
67 const message =
68 error instanceof Error ? error.message : "Failed to initialize filesystem checkpoints.";
69 this.emitFilesystemCheckpointError(session.sessionId, message, session.threadId);
70 });
71 return session;
72 }
73
74 async sendTurn(raw: ProviderSendTurnInput): Promise<ProviderTurnStartResult> {
75 const input = providerSendTurnInputSchema.parse(raw);
76 return this.resolveAdapterForSession(input.sessionId).sendTurn(input);
77 }
78
79 async interruptTurn(raw: ProviderInterruptTurnInput): Promise<void> {
80 const input = providerInterruptTurnInputSchema.parse(raw);
81 return this.resolveAdapterForSession(input.sessionId).interruptTurn(input);
82 }
83
84 async respondToRequest(raw: ProviderRespondToRequestInput): Promise<void> {
85 const input = providerRespondToRequestInputSchema.parse(raw);
86 return this.resolveAdapterForSession(input.sessionId).respondToRequest(input);
87 }
88
89 stopSession(raw: ProviderStopSessionInput): void {
90 const input = providerStopSessionInputSchema.parse(raw);
91 this.resolveAdapterForSession(input.sessionId).stopSession(input);
92 this.sessionToProvider.delete(input.sessionId);
93 this.sessionCheckpointCwds.delete(input.sessionId);
94 this.filesystemLocks.delete(input.sessionId);
95 }
96
97 listSessions(): ProviderSession[] {
98 const sessions = [...this.adapters.values()].flatMap((adapter) => adapter.listSessions());
99 for (const session of sessions) {
100 this.sessionToProvider.set(session.sessionId, session.provider);
101 }
102 return sessions;
103 }
104
105 async listCheckpoints(raw: ProviderListCheckpointsInput): Promise<ProviderListCheckpointsResult> {
106 const input = providerListCheckpointsInputSchema.parse(raw);
107 const adapter = this.resolveAdapterForSession(input.sessionId);
108 const snapshot = await adapter.readThread(input.sessionId);
109 return {
110 threadId: snapshot.threadId,
111 checkpoints: buildCheckpoints(snapshot.turns),
112 };
113 }
114
115 async getCheckpointDiff(
116 raw: ProviderGetCheckpointDiffInput,
117 ): Promise<ProviderGetCheckpointDiffResult> {
118 const input = providerGetCheckpointDiffInputSchema.parse(raw);
119 const adapter = this.resolveAdapterForSession(input.sessionId);
120 const checkpointCwd = await this.getOrInitializeFilesystemCheckpointCwd(input.sessionId, adapter);
121 if (!checkpointCwd) {
122 throw new Error("Filesystem checkpoints are unavailable for this session.");
123 }
124
125 return this.withFilesystemLock(input.sessionId, async () => {
126 const snapshot = await adapter.readThread(input.sessionId);
127 if (input.toTurnCount > snapshot.turns.length) {
128 throw new Error(
129 `Checkpoint turn count ${input.toTurnCount} exceeds current turn count ${snapshot.turns.length}.`,
130 );
131 }
132
133 const diff = await this.filesystemCheckpointStore.diffCheckpoints({
134 cwd: checkpointCwd,
135 threadId: snapshot.threadId,
136 fromTurnCount: input.fromTurnCount,
137 toTurnCount: input.toTurnCount,
138 });
139 return {
140 threadId: snapshot.threadId,
141 fromTurnCount: input.fromTurnCount,
142 toTurnCount: input.toTurnCount,
143 diff,
144 };
145 });
146 }
147
148 async revertToCheckpoint(
149 raw: ProviderRevertToCheckpointInput,
150 ): Promise<ProviderRevertToCheckpointResult> {
151 const input = providerRevertToCheckpointInputSchema.parse(raw);
152 const adapter = this.resolveAdapterForSession(input.sessionId);
153 const checkpointCwd = await this.getOrInitializeFilesystemCheckpointCwd(input.sessionId, adapter);
154 if (!checkpointCwd) {
155 throw new Error("Filesystem checkpoints are unavailable for this session.");
156 }
157
158 return this.withFilesystemLock(input.sessionId, async () => {
159 const beforeSnapshot = await adapter.readThread(input.sessionId);
160 const currentTurnCount = beforeSnapshot.turns.length;
161 if (input.turnCount > currentTurnCount) {
162 throw new Error(
163 `Checkpoint turn count ${input.turnCount} exceeds current turn count ${currentTurnCount}.`,
164 );
165 }
166
167 if (input.turnCount > 0) {
168 const hasCheckpoint = await this.filesystemCheckpointStore.hasCheckpoint({
169 cwd: checkpointCwd,
170 threadId: beforeSnapshot.threadId,
171 turnCount: input.turnCount,
172 });
173 if (!hasCheckpoint) {
174 throw new Error(
175 `Filesystem checkpoint is unavailable for turn ${input.turnCount} in thread ${beforeSnapshot.threadId}.`,
176 );
177 }
178 }
179
180 const restored = await this.filesystemCheckpointStore.restoreCheckpoint({
181 cwd: checkpointCwd,
182 threadId: beforeSnapshot.threadId,
183 turnCount: input.turnCount,
184 });
185 if (!restored) {
186 throw new Error(
187 `Filesystem checkpoint is unavailable for turn ${input.turnCount} in thread ${beforeSnapshot.threadId}.`,
188 );
189 }
190
191 const rollbackTurns = currentTurnCount - input.turnCount;
192 const afterSnapshot =
193 rollbackTurns > 0
194 ? await adapter.rollbackThread(input.sessionId, rollbackTurns)
195 : beforeSnapshot;
196
197 await this.filesystemCheckpointStore.pruneAfterTurn({
198 cwd: checkpointCwd,
199 threadId: afterSnapshot.threadId,
200 maxTurnCount: afterSnapshot.turns.length,
201 });
202
203 const checkpoints = buildCheckpoints(afterSnapshot.turns);
204 const currentCheckpoint =
205 checkpoints.find((checkpoint) => checkpoint.isCurrent) ??
206 checkpoints[checkpoints.length - 1] ??
207 checkpoints[0];
208 const rolledBackTurns = Math.max(0, currentTurnCount - afterSnapshot.turns.length);
209
210 return {
211 threadId: afterSnapshot.threadId,
212 turnCount: currentCheckpoint?.turnCount ?? 0,
213 messageCount: currentCheckpoint?.messageCount ?? 0,
214 rolledBackTurns,
215 checkpoints,
216 };
217 });
218 }
219
220 stopAll(): void {
221 for (const adapter of this.adapters.values()) {
222 adapter.stopAll();
223 }
224 this.sessionToProvider.clear();
225 this.sessionCheckpointCwds.clear();
226 this.filesystemLocks.clear();
227 }
228
229 dispose(): void {
230 for (const unsubscribe of this.unsubscribers) {
231 unsubscribe();
232 }
233 for (const adapter of this.adapters.values()) {
234 adapter.dispose();
235 }
236 this.unsubscribers.length = 0;
237 this.sessionToProvider.clear();
238 this.sessionCheckpointCwds.clear();
239 this.filesystemLocks.clear();
240 }
241
242 private async initializeFilesystemCheckpointing(
243 adapter: ProviderAdapter,
244 session: ProviderSession,
245 preferredCwd?: string,
246 ): Promise<void> {
247 const cwd = preferredCwd ?? session.cwd ?? process.cwd();
248 await this.withFilesystemLock(session.sessionId, async () => {
249 const supportsGit = await this.filesystemCheckpointStore.isGitRepository(cwd);
250 if (!supportsGit) {
251 this.sessionCheckpointCwds.delete(session.sessionId);
252 return;
253 }
254
255 const snapshot = await adapter.readThread(session.sessionId);
256 await this.filesystemCheckpointStore.ensureRootCheckpoint({
257 cwd,
258 threadId: snapshot.threadId,
259 });
260 await this.filesystemCheckpointStore.captureCheckpoint({
261 cwd,
262 threadId: snapshot.threadId,
263 turnCount: snapshot.turns.length,
264 });
265 this.sessionCheckpointCwds.set(session.sessionId, cwd);
266 });
267 }
268
269 private async getOrInitializeFilesystemCheckpointCwd(
270 sessionId: string,
271 adapter: ProviderAdapter,
272 ): Promise<string | null> {
273 const existingCwd = this.sessionCheckpointCwds.get(sessionId);
274 if (existingCwd) return existingCwd;
275
276 const session = adapter.listSessions().find((entry) => entry.sessionId === sessionId);
277 const candidateCwds = session?.cwd ? [session.cwd] : [process.cwd()];
278
279 await this.withFilesystemLock(sessionId, async () => {
280 const currentCwd = this.sessionCheckpointCwds.get(sessionId);
281 if (currentCwd) return;
282
283 const cwdSupport = await Promise.all(
284 candidateCwds.map(async (cwd) => ({
285 cwd,
286 supportsGit: await this.filesystemCheckpointStore.isGitRepository(cwd),
287 })),
288 );
289 const supportedCwd = cwdSupport.find((entry) => entry.supportsGit)?.cwd;
290 if (!supportedCwd) {
291 this.sessionCheckpointCwds.delete(sessionId);
292 return;
293 }
294
295 const snapshot = await adapter.readThread(sessionId);
296 await this.filesystemCheckpointStore.ensureRootCheckpoint({
297 cwd: supportedCwd,
298 threadId: snapshot.threadId,
299 });
300 await this.filesystemCheckpointStore.captureCheckpoint({
301 cwd: supportedCwd,
302 threadId: snapshot.threadId,
303 turnCount: snapshot.turns.length,
304 });
305 this.sessionCheckpointCwds.set(sessionId, supportedCwd);
306 });
307
308 return this.sessionCheckpointCwds.get(sessionId) ?? null;
309 }
310
311 private maybeCaptureFilesystemCheckpoint(event: ProviderEvent): void {
312 if (event.kind !== "notification" || event.method !== "turn/completed") {
313 return;
314 }
315 const adapter = this.resolveAdapterForSession(event.sessionId);
316 const checkpointCwd = this.sessionCheckpointCwds.get(event.sessionId);
317 if (!checkpointCwd) {
318 void this.getOrInitializeFilesystemCheckpointCwd(event.sessionId, adapter)
319 .then(async (initializedCwd) => {
320 if (!initializedCwd) return;
321 const snapshot = await adapter.readThread(event.sessionId);
322 this.emitCheckpointCaptured(event.sessionId, snapshot.threadId, snapshot.turns.length);
323 })
324 .catch((error) => {
325 const message =
326 error instanceof Error ? error.message : "Failed to initialize filesystem checkpoints.";
327 this.emitFilesystemCheckpointError(event.sessionId, message, event.threadId);
328 });
329 return;
330 }
331
332 void this.withFilesystemLock(event.sessionId, async () => {
333 const snapshot = await adapter.readThread(event.sessionId);
334 await this.filesystemCheckpointStore.captureCheckpoint({
335 cwd: checkpointCwd,
336 threadId: snapshot.threadId,
337 turnCount: snapshot.turns.length,
338 });
339 this.emitCheckpointCaptured(event.sessionId, snapshot.threadId, snapshot.turns.length);
340 }).catch((error) => {
341 const message = error instanceof Error ? error.message : "Failed to capture checkpoint.";
342 this.emitFilesystemCheckpointError(event.sessionId, message, event.threadId);
343 });
344 }
345
346 private emitCheckpointCaptured(sessionId: string, threadId: string, turnCount: number): void {
347 this.emit("event", {
348 id: randomUUID(),
349 kind: "notification",
350 provider: this.sessionToProvider.get(sessionId) ?? "codex",
351 sessionId,
352 createdAt: new Date().toISOString(),
353 method: "checkpoint/captured",
354 threadId,
355 payload: {
356 threadId,
357 turnCount,
358 },
359 });
360 }
361
362 private emitFilesystemCheckpointError(
363 sessionId: string,
364 message: string,
365 threadId?: string,
366 ): void {
367 this.emit("event", {
368 id: randomUUID(),
369 kind: "error",
370 provider: this.sessionToProvider.get(sessionId) ?? "codex",
371 sessionId,
372 createdAt: new Date().toISOString(),
373 method: "checkpoint/filesystemError",
374 message,
375 ...(threadId ? { threadId } : {}),
376 });
377 }
378
379 private withFilesystemLock<T>(sessionId: string, fn: () => Promise<T>): Promise<T> {
380 const previous = this.filesystemLocks.get(sessionId) ?? Promise.resolve();
381 const next = previous.catch(() => undefined).then(fn);
382 const completion = next.then(
383 () => undefined,
384 () => undefined,
385 );
386 this.filesystemLocks.set(sessionId, completion);
387 return next.finally(() => {
388 const tracked = this.filesystemLocks.get(sessionId);
389 if (tracked === completion) {
390 this.filesystemLocks.delete(sessionId);
391 }
392 });
393 }
394
395 private resolveAdapterForSession(sessionId: string): ProviderAdapter {
396 const mappedProvider = this.sessionToProvider.get(sessionId);
397 if (mappedProvider) {
398 return this.requireAdapter(mappedProvider);
399 }
400 for (const adapter of this.adapters.values()) {
401 if (adapter.hasSession(sessionId)) {
402 this.sessionToProvider.set(sessionId, adapter.kind);
403 return adapter;
404 }
405 }
406 throw new Error(`Unknown provider session: ${sessionId}`);
407 }
408
409 private requireAdapter(provider: ProviderKind): ProviderAdapter {
410 const adapter = this.adapters.get(provider);
411 if (!adapter) {
412 throw new Error(`Provider '${provider}' is not implemented yet.`);
413 }
414 return adapter;
415 }
416}
417
418function asObject(value: unknown): Record<string, unknown> | undefined {
419 if (!value || typeof value !== "object") return undefined;
420 return value as Record<string, unknown>;
421}
422
423function asString(value: unknown): string | undefined {
424 return typeof value === "string" ? value : undefined;
425}
426
427function asArray(value: unknown): unknown[] {
428 return Array.isArray(value) ? value : [];
429}
430
431function trimToPreview(value: string): string {
432 const normalized = value.replace(/\s+/g, " ").trim();
433 if (normalized.length <= 120) return normalized;
434 return `${normalized.slice(0, 117)}...`;
435}
436
437function summarizeUserMessageContent(content: unknown[]): string | undefined {
438 const segments: string[] = [];
439 for (const part of content) {
440 const record = asObject(part);
441 const type = asString(record?.type);
442 if (!type) continue;
443 if (type === "text") {
444 const text = asString(record?.text);
445 if (text && text.trim().length > 0) {
446 segments.push(text.trim());
447 }
448 continue;
449 }
450 if (type === "image") {
451 segments.push("[Image attachment]");
452 continue;
453 }
454 if (type === "localImage") {
455 segments.push("[Local image attachment]");
456 }
457 }
458 if (segments.length === 0) return undefined;
459 return trimToPreview(segments.join(" "));
460}
461
462function summarizeTurn(turn: ProviderThreadTurnSnapshot): { messageCountDelta: number; preview?: string } {
463 let messageCountDelta = 0;
464 let preview: string | undefined;
465 for (const item of turn.items) {
466 const record = asObject(item);
467 const type = asString(record?.type);
468 if (!type) continue;
469 if (type === "userMessage") {
470 messageCountDelta += 1;
471 if (!preview) {
472 preview = summarizeUserMessageContent(asArray(record?.content));
473 }
474 continue;
475 }
476 if (type === "agentMessage") {
477 messageCountDelta += 1;
478 if (!preview) {
479 const text = asString(record?.text);
480 if (text && text.trim().length > 0) {
481 preview = trimToPreview(text);
482 }
483 }
484 }
485 }
486 return {
487 messageCountDelta,
488 ...(preview ? { preview } : {}),
489 };
490}
491
492function buildCheckpoints(turns: ProviderThreadTurnSnapshot[]): ProviderCheckpoint[] {
493 const checkpoints: ProviderCheckpoint[] = [];
494 let messageCount = 0;
495 const isEmpty = turns.length === 0;
496 checkpoints.push({
497 id: "root",
498 turnCount: 0,
499 messageCount: 0,
500 label: "Start of conversation",
501 isCurrent: isEmpty,
502 });
503 for (let index = 0; index < turns.length; index += 1) {
504 const turn = turns[index];
505 if (!turn) continue;
506 const turnSummary = summarizeTurn(turn);
507 messageCount += turnSummary.messageCountDelta;
508 checkpoints.push({
509 id: turn.id,
510 turnCount: index + 1,
511 messageCount,
512 label: `Turn ${index + 1}`,
513 ...(turnSummary.preview ? { preview: turnSummary.preview } : {}),
514 isCurrent: index === turns.length - 1,
515 });
516 }
517 return checkpoints;
518}
apps/server/src/wsServer.ts+15−14
@@ -24,7 +24,6 @@ import { WebSocketServer, type WebSocket } from "ws"; · 23 unmodified lines
24 24
25import { createLogger } from "./logger";25import { createLogger } from "./logger";
26import { ProjectRegistry } from "./projectRegistry";26import { ProjectRegistry } from "./projectRegistry";
27import { ProviderManager } from "./providerManager";
28import { GitManager } from "./gitManager";27import { GitManager } from "./gitManager";
29import {28import {
30 checkoutGitBranch,29 checkoutGitBranch,
@@ -40,6 +39,8 @@ import { loadResolvedKeybindingsConfig, upsertKeybindingRule } from "./keybindin · 9 unmodified lines
40import { searchWorkspaceEntries } from "./workspaceEntries";39import { searchWorkspaceEntries } from "./workspaceEntries";
41import { CoreEngine } from "./effect/coreEngine";40import { CoreEngine } from "./effect/coreEngine";
42import { SqliteEventStore } from "./effect/eventStore";41import { SqliteEventStore } from "./effect/eventStore";
42import { CodexProviderAdapter } from "./effect/providers/codexProviderAdapter";
43import { ProviderOrchestrator } from "./effect/providers/providerOrchestrator";
43 44
44const MIME_TYPES: Record<string, string> = {45const MIME_TYPES: Record<string, string> = {
45 ".html": "text/html; charset=utf-8",46 ".html": "text/html; charset=utf-8",
@@ -92,7 +93,7 @@ export function createServer(options: ServerOptions) { · 46 unmodified lines
92 terminalManager: providedTerminalManager,93 terminalManager: providedTerminalManager,
93 authToken,94 authToken,
94 } = options;95 } = options;
95 const providerManager = new ProviderManager();96 const providerOrchestrator = new ProviderOrchestrator([new CodexProviderAdapter()]);
96 const terminalManager = providedTerminalManager ?? new TerminalManager();97 const terminalManager = providedTerminalManager ?? new TerminalManager();
97 const projectRegistry =98 const projectRegistry =
98 providedRegistry ?? new ProjectRegistry(path.join(os.homedir(), ".t3", "userdata"));99 providedRegistry ?? new ProjectRegistry(path.join(os.homedir(), ".t3", "userdata"));
@@ -143,7 +144,7 @@ export function createServer(options: ServerOptions) { · 44 unmodified lines
143 }144 }
144 145
145 // Forward provider events to all connected WebSocket clients146 // Forward provider events to all connected WebSocket clients
146 providerManager.on("event", (event) => {147 providerOrchestrator.on("event", (event) => {
147 const push: WsPush = {148 const push: WsPush = {
148 type: "push",149 type: "push",
149 channel: WS_CHANNELS.providerEvent,150 channel: WS_CHANNELS.providerEvent,
@@ -336,33 +337,33 @@ export function createServer(options: ServerOptions) { · 186 unmodified lines
336 async function routeRequest(request: WsRequest): Promise<unknown> {337 async function routeRequest(request: WsRequest): Promise<unknown> {
337 switch (request.method) {338 switch (request.method) {
338 case WS_METHODS.providersStartSession:339 case WS_METHODS.providersStartSession:
339 return providerManager.startSession(request.params as never);340 return providerOrchestrator.startSession(request.params as never);
340 341
341 case WS_METHODS.providersSendTurn:342 case WS_METHODS.providersSendTurn:
342 return providerManager.sendTurn(request.params as never);343 return providerOrchestrator.sendTurn(request.params as never);
343 344
344 case WS_METHODS.providersInterruptTurn:345 case WS_METHODS.providersInterruptTurn:
345 return providerManager.interruptTurn(request.params as never);346 return providerOrchestrator.interruptTurn(request.params as never);
346 347
347 case WS_METHODS.providersRespondToRequest:348 case WS_METHODS.providersRespondToRequest:
348 return providerManager.respondToRequest(request.params as never);349 return providerOrchestrator.respondToRequest(request.params as never);
349 350
350 case WS_METHODS.providersStopSession: {351 case WS_METHODS.providersStopSession: {
351 providerManager.stopSession(request.params as never);352 providerOrchestrator.stopSession(request.params as never);
352 return undefined;353 return undefined;
353 }354 }
354 355
355 case WS_METHODS.providersListSessions:356 case WS_METHODS.providersListSessions:
356 return providerManager.listSessions();357 return providerOrchestrator.listSessions();
357 358
358 case WS_METHODS.providersListCheckpoints:359 case WS_METHODS.providersListCheckpoints:
359 return providerManager.listCheckpoints(request.params as never);360 return providerOrchestrator.listCheckpoints(request.params as never);
360 361
361 case WS_METHODS.providersGetCheckpointDiff:362 case WS_METHODS.providersGetCheckpointDiff:
362 return providerManager.getCheckpointDiff(request.params as never);363 return providerOrchestrator.getCheckpointDiff(request.params as never);
363 364
364 case WS_METHODS.providersRevertToCheckpoint:365 case WS_METHODS.providersRevertToCheckpoint:
365 return providerManager.revertToCheckpoint(request.params as never);366 return providerOrchestrator.revertToCheckpoint(request.params as never);
366 367
367 case WS_METHODS.projectsList:368 case WS_METHODS.projectsList:
368 return projectRegistry.list();369 return projectRegistry.list();
@@ -651,8 +652,8 @@ export function createServer(options: ServerOptions) { · 282 unmodified lines
651 unsubscribeCoreDelta = null;652 unsubscribeCoreDelta = null;
652 }653 }
653 terminalManager.off("event", onTerminalEvent);654 terminalManager.off("event", onTerminalEvent);
654 providerManager.stopAll();655 providerOrchestrator.stopAll();
655 providerManager.dispose();656 providerOrchestrator.dispose();
656 terminalManager.dispose();657 terminalManager.dispose();
657 await coreEngine.stop();658 await coreEngine.stop();
658 eventStore.close();659 eventStore.close();