diff --git a/.plans/13-provider-service-integration-tests.md b/.plans/13-provider-service-integration-tests.md new file mode 100644 index 0000000000000000000000000000000000000000..0cc916429d634aac392a6f6ffe7b332180eb737a --- /dev/null +++ b/.plans/13-provider-service-integration-tests.md @@ -0,0 +1,110 @@ +# ProviderService Integration Test Plan + +Goal: +- Validate end-to-end `ProviderService` behavior with real layers: + - `ProviderServiceLive` + - `CheckpointServiceLive` + - `CheckpointStoreLive` + - `CheckpointRepositoryLive` (sqlite in-memory) + - `ProviderSessionDirectoryLive` +- Only fake the adapter event source (deterministic Codex-like stream). +- Avoid mocking checkpointing/persistence orchestration logic. + +## Test Harness + +Build a deterministic `TestProviderAdapterLive` in `apps/server/src/provider/Layers/TestProviderAdapter.integration.ts`: +- Service contract: `ProviderAdapterShape`. +- Internal state: + - session registry (session + cwd + threadId) + - thread snapshot store (`threadId`, `turns`) + - event subscribers +- Behavior: + - `startSession`: creates session with threadId. + - `sendTurn`: appends a deterministic turn snapshot and emits ordered events: + - `turn/started` + - `item/started` / `item/completed` (tool + approval variants depending on scenario) + - `item/agentMessage/delta` chunks + - `turn/completed` + - optional "mutator" callback per turn to change workspace files before completion. + - `readThread`, `rollbackThread`, `stopSession`, `stopAll`. + +Use real git-backed temporary workspaces in integration tests: +- initialize repo with baseline commit +- run provider turn in workspace +- assert checkpoint diffs against real git refs + +## Core Integration Specs + +1) `startSession` initializes checkpoint root exactly once +- Arrange: + - start provider session in git repo. +- Assert: + - `provider_checkpoints` contains root row (turn 0). + - checkpoint ref exists in git. + - second `startSession` for new session creates a new independent root. + +2) Turn without filesystem change +- Arrange: + - emit normal turn events, no file mutation. +- Assert: + - provider subscribers receive: + - `turn/started` + - `turn/completed` + - synthetic `checkpoint/captured` + - `listCheckpoints` returns root + turn 1. + - `getCheckpointDiff(0 -> 1)` returns empty/no-op diff. + +3) Turn with filesystem change +- Arrange: + - mutate `README.md` during turn. +- Assert: + - `listCheckpoints` returns root + turn 1. + - `getCheckpointDiff(0 -> 1)` contains file path and hunk. + - persisted checkpoint metadata includes non-empty `checkpointRef`. + +4) Multi-turn sequencing and checkpoint monotonicity +- Arrange: + - turn 1: no file change + - turn 2: file change + - turn 3: file change +- Assert: + - turn counts are monotonic and contiguous in DB (0,1,2,3). + - latest checkpoint is marked current. + - diffs for adjacent turns map to expected filesystem deltas. + +5) Revert to checkpoint +- Arrange: + - execute 3 turns with at least one file-changing turn. + - call `revertToCheckpoint(turnCount=1)`. +- Assert: + - workspace content matches turn 1 state. + - adapter `rollbackThread` called with `numTurns=2`. + - DB rows for turns >1 are removed. + - later refs are deleted from git. + +6) Capture failure surface +- Arrange: + - adapter emits `turn/completed`, but file mutation leaves invalid repo state or store capture fails. +- Assert: + - `ProviderService` emits `checkpoint/captureError`. + - no partial metadata/ref divergence is left behind. + +## WebSocket Coverage (Thin Integration) + +Add one ws server integration spec: +- Subscribe to `providers.event`. +- Run a deterministic provider turn through ws methods. +- Assert push stream includes: + - `turn/started`, tool events, `turn/completed`, `checkpoint/captured`. +- Assert orchestration projection still updates assistant message and turn diff summary. + +## Proposed PR Split + +PR A: +- Test adapter harness + shared integration fixtures (repo setup, runtime/layer setup). + +PR B: +- Core ProviderService integration specs (cases 1-4). + +PR C: +- Revert + failure-path specs (cases 5-6) + ws thin integration spec. diff --git a/apps/server/integration/TestProviderAdapter.integration.ts b/apps/server/integration/TestProviderAdapter.integration.ts new file mode 100644 index 0000000000000000000000000000000000000000..b9828f881c1344882ba7248fb4978d2928bb1324 --- /dev/null +++ b/apps/server/integration/TestProviderAdapter.integration.ts @@ -0,0 +1,310 @@ +import { randomUUID } from "node:crypto"; + +import type { + ProviderRuntimeEvent, + ProviderSession, + ProviderTurnStartResult, +} from "@t3tools/contracts"; +import { Effect, Queue, Stream } from "effect"; + +import { + ProviderAdapterSessionNotFoundError, + ProviderAdapterValidationError, + type ProviderAdapterError, +} from "../src/provider/Errors.ts"; +import type { + ProviderAdapterShape, + ProviderThreadSnapshot, + ProviderThreadTurnSnapshot, +} from "../src/provider/Services/ProviderAdapter.ts"; + +export interface TestTurnResponse { + readonly events: ReadonlyArray; + readonly mutateWorkspace?: (input: { + readonly cwd: string; + readonly turnCount: number; + }) => Effect.Effect; +} + +interface SessionState { + readonly session: ProviderSession; + snapshot: ProviderThreadSnapshot; + turnCount: number; + readonly queuedResponses: Array; + readonly rollbackCalls: Array; +} + +export interface TestProviderAdapterHarness { + readonly adapter: ProviderAdapterShape; + readonly queueTurnResponse: ( + sessionId: string, + response: TestTurnResponse, + ) => Effect.Effect; + readonly getRollbackCalls: (sessionId: string) => ReadonlyArray; +} + +const PROVIDER = "codex" as const; + +function nowIso(): string { + return new Date().toISOString(); +} + +function sessionNotFound(sessionId: string): ProviderAdapterSessionNotFoundError { + return new ProviderAdapterSessionNotFoundError({ + provider: PROVIDER, + sessionId, + }); +} + +function missingSessionEffect(sessionId: string): Effect.Effect { + return Effect.fail(sessionNotFound(sessionId)); +} + +export const makeTestProviderAdapterHarness = Effect.gen(function* () { + const runtimeEvents = yield* Queue.unbounded(); + let sessionCount = 0; + const sessions = new Map(); + + const emit = (event: ProviderRuntimeEvent) => Queue.offer(runtimeEvents, event); + + const startSession: ProviderAdapterShape["startSession"] = (input) => + Effect.gen(function* () { + if (input.provider !== undefined && input.provider !== PROVIDER) { + return yield* Effect.fail( + new ProviderAdapterValidationError({ + provider: PROVIDER, + operation: "startSession", + issue: `Expected provider '${PROVIDER}' but received '${input.provider}'.`, + }), + ); + } + + sessionCount += 1; + const sessionId = `test-session-${sessionCount}`; + const threadId = `test-thread-${sessionCount}`; + const createdAt = nowIso(); + + const session: ProviderSession = { + sessionId, + provider: PROVIDER, + status: "ready", + threadId, + cwd: input.cwd, + createdAt, + updatedAt: createdAt, + }; + + sessions.set(sessionId, { + session, + snapshot: { + threadId, + turns: [], + }, + turnCount: 0, + queuedResponses: [], + rollbackCalls: [], + }); + + return session; + }); + + const sendTurn: ProviderAdapterShape["sendTurn"] = (input) => + Effect.gen(function* () { + const state = sessions.get(input.sessionId); + if (!state) { + return yield* missingSessionEffect(input.sessionId); + } + + state.turnCount += 1; + const turnCount = state.turnCount; + const turnId = `turn-${turnCount}`; + + const response = state.queuedResponses.shift(); + if (!response) { + return yield* Effect.fail( + new ProviderAdapterValidationError({ + provider: PROVIDER, + operation: "sendTurn", + issue: `No queued turn response for session ${input.sessionId}.`, + }), + ); + } + + const assistantDeltas: string[] = []; + let hasCompletedEvent = false; + for (const fixtureEvent of response.events) { + const rawEvent: Record = { + ...(fixtureEvent as Record), + eventId: randomUUID(), + provider: PROVIDER, + sessionId: input.sessionId, + createdAt: nowIso(), + }; + if (Object.hasOwn(rawEvent, "threadId")) { + rawEvent.threadId = state.snapshot.threadId; + } + if (Object.hasOwn(rawEvent, "turnId")) { + rawEvent.turnId = turnId; + } + + const runtimeEvent = rawEvent as ProviderRuntimeEvent; + if (runtimeEvent.type === "message.delta") { + assistantDeltas.push(runtimeEvent.delta); + } + if (runtimeEvent.type === "turn.completed") { + hasCompletedEvent = true; + } + + yield* emit(runtimeEvent); + } + + if (response.mutateWorkspace && state.session.cwd) { + yield* response.mutateWorkspace({ cwd: state.session.cwd!, turnCount }); + } + + const userItem = { + type: "userMessage", + content: [{ type: "text", text: input.input }], + } as const; + const assistantText = assistantDeltas.join(""); + const nextItems: Array = + assistantText.length > 0 + ? [userItem, { type: "agentMessage", text: assistantText }] + : [userItem]; + + const nextTurn: ProviderThreadTurnSnapshot = { + id: turnId, + items: nextItems, + }; + + state.snapshot = { + threadId: state.snapshot.threadId, + turns: [...state.snapshot.turns, nextTurn], + }; + + if (!hasCompletedEvent) { + yield* emit({ + type: "turn.completed", + eventId: randomUUID(), + provider: PROVIDER, + sessionId: input.sessionId, + createdAt: nowIso(), + threadId: state.snapshot.threadId, + turnId, + status: "completed", + }); + } + + return { + threadId: state.snapshot.threadId, + turnId, + } satisfies ProviderTurnStartResult; + }); + + const interruptTurn: ProviderAdapterShape["interruptTurn"] = ( + sessionId, + _turnId, + ) => (sessions.has(sessionId) ? Effect.void : missingSessionEffect(sessionId)); + + const respondToRequest: ProviderAdapterShape["respondToRequest"] = ( + sessionId, + _requestId, + _decision, + ) => (sessions.has(sessionId) ? Effect.void : missingSessionEffect(sessionId)); + + const stopSession: ProviderAdapterShape["stopSession"] = (sessionId) => + Effect.sync(() => { + sessions.delete(sessionId); + }); + + const listSessions: ProviderAdapterShape["listSessions"] = () => + Effect.sync(() => Array.from(sessions.values(), (state) => state.session)); + + const hasSession: ProviderAdapterShape["hasSession"] = (sessionId) => + Effect.succeed(sessions.has(sessionId)); + + const readThread: ProviderAdapterShape["readThread"] = (sessionId) => { + const state = sessions.get(sessionId); + if (!state) { + return missingSessionEffect(sessionId); + } + return Effect.succeed(state.snapshot); + }; + + const rollbackThread: ProviderAdapterShape["rollbackThread"] = ( + sessionId, + numTurns, + ) => { + const state = sessions.get(sessionId); + if (!state) { + return missingSessionEffect(sessionId); + } + if (!Number.isInteger(numTurns) || numTurns < 0 || numTurns > state.snapshot.turns.length) { + return Effect.fail( + new ProviderAdapterValidationError({ + provider: PROVIDER, + operation: "rollbackThread", + issue: "numTurns must be an integer between 0 and current turn count.", + }), + ); + } + + return Effect.sync(() => { + state.rollbackCalls.push(numTurns); + state.snapshot = { + threadId: state.snapshot.threadId, + turns: state.snapshot.turns.slice(0, state.snapshot.turns.length - numTurns), + }; + state.turnCount = state.snapshot.turns.length; + return state.snapshot; + }); + }; + + const stopAll: ProviderAdapterShape["stopAll"] = () => + Effect.sync(() => { + sessions.clear(); + }); + + const adapter: ProviderAdapterShape = { + provider: PROVIDER, + startSession, + sendTurn, + interruptTurn, + respondToRequest, + stopSession, + listSessions, + hasSession, + readThread, + rollbackThread, + stopAll, + streamEvents: Stream.fromQueue(runtimeEvents), + }; + + const queueTurnResponse = ( + sessionId: string, + response: TestTurnResponse, + ): Effect.Effect => + Effect.sync(() => sessions.get(sessionId)).pipe( + Effect.flatMap((state) => + state + ? Effect.sync(() => { + state.queuedResponses.push(response); + }) + : Effect.fail(sessionNotFound(sessionId)), + ), + ); + + const getRollbackCalls = (sessionId: string): ReadonlyArray => { + const state = sessions.get(sessionId); + if (!state) { + return []; + } + return [...state.rollbackCalls]; + }; + + return { + adapter, + queueTurnResponse, + getRollbackCalls, + } satisfies TestProviderAdapterHarness; +}); diff --git a/apps/server/integration/fixtures/providerRuntime.ts b/apps/server/integration/fixtures/providerRuntime.ts new file mode 100644 index 0000000000000000000000000000000000000000..6ea6587da70c32003ef1246547907b6634e650df --- /dev/null +++ b/apps/server/integration/fixtures/providerRuntime.ts @@ -0,0 +1,130 @@ +import type { ProviderRuntimeEvent } from "@t3tools/contracts"; + +const PROVIDER = "codex" as const; +const SESSION_ID = "fixture-session"; +const THREAD_ID = "fixture-thread"; +const TURN_ID = "fixture-turn"; + +function baseEvent(eventId: string, createdAt: string): Pick< + ProviderRuntimeEvent, + "eventId" | "provider" | "sessionId" | "createdAt" +> { + return { + eventId, + provider: PROVIDER, + sessionId: SESSION_ID, + createdAt, + }; +} + +export const codexTurnTextFixture = [ + { + type: "turn.started", + ...baseEvent("evt-1", "2026-02-23T00:00:00.000Z"), + threadId: THREAD_ID, + turnId: TURN_ID, + }, + { + type: "message.delta", + ...baseEvent("evt-2", "2026-02-23T00:00:00.100Z"), + threadId: THREAD_ID, + turnId: TURN_ID, + delta: "I will make a small update.\n", + }, + { + type: "message.delta", + ...baseEvent("evt-3", "2026-02-23T00:00:00.200Z"), + threadId: THREAD_ID, + turnId: TURN_ID, + delta: "Done.\n", + }, + { + type: "turn.completed", + ...baseEvent("evt-4", "2026-02-23T00:00:00.300Z"), + threadId: THREAD_ID, + turnId: TURN_ID, + status: "completed", + }, +] satisfies ReadonlyArray; + +export const codexTurnToolFixture = [ + { + type: "turn.started", + ...baseEvent("evt-11", "2026-02-23T00:01:00.000Z"), + threadId: THREAD_ID, + turnId: TURN_ID, + }, + { + type: "tool.started", + ...baseEvent("evt-12", "2026-02-23T00:01:00.100Z"), + threadId: THREAD_ID, + turnId: TURN_ID, + toolKind: "command", + title: "Command run", + detail: "echo integration", + }, + { + type: "tool.completed", + ...baseEvent("evt-13", "2026-02-23T00:01:00.200Z"), + threadId: THREAD_ID, + turnId: TURN_ID, + toolKind: "command", + title: "Command run", + detail: "echo integration", + }, + { + type: "message.delta", + ...baseEvent("evt-14", "2026-02-23T00:01:00.300Z"), + threadId: THREAD_ID, + turnId: TURN_ID, + delta: "Applied the requested edit.\n", + }, + { + type: "turn.completed", + ...baseEvent("evt-15", "2026-02-23T00:01:00.400Z"), + threadId: THREAD_ID, + turnId: TURN_ID, + status: "completed", + }, +] satisfies ReadonlyArray; + +export const codexTurnApprovalFixture = [ + { + type: "turn.started", + ...baseEvent("evt-21", "2026-02-23T00:02:00.000Z"), + threadId: THREAD_ID, + turnId: TURN_ID, + }, + { + type: "approval.requested", + ...baseEvent("evt-22", "2026-02-23T00:02:00.100Z"), + threadId: THREAD_ID, + turnId: TURN_ID, + requestId: "req-1", + requestKind: "command", + detail: "Please approve command", + }, + { + type: "approval.resolved", + ...baseEvent("evt-23", "2026-02-23T00:02:00.200Z"), + threadId: THREAD_ID, + turnId: TURN_ID, + requestId: "req-1", + requestKind: "command", + decision: "accept", + }, + { + type: "message.delta", + ...baseEvent("evt-24", "2026-02-23T00:02:00.300Z"), + threadId: THREAD_ID, + turnId: TURN_ID, + delta: "Approval received and command executed.\n", + }, + { + type: "turn.completed", + ...baseEvent("evt-25", "2026-02-23T00:02:00.400Z"), + threadId: THREAD_ID, + turnId: TURN_ID, + status: "completed", + }, +] satisfies ReadonlyArray; diff --git a/apps/server/integration/providerService.integration.test.ts b/apps/server/integration/providerService.integration.test.ts new file mode 100644 index 0000000000000000000000000000000000000000..af4326bc6acd901f839668946cd6208799571d83 --- /dev/null +++ b/apps/server/integration/providerService.integration.test.ts @@ -0,0 +1,361 @@ +import type { ProviderRuntimeEvent } from "@t3tools/contracts"; +import { NodeServices } from "@effect/platform-node"; +import { it, assert } from "@effect/vitest"; +import { Effect, FileSystem, Layer, Path, Queue, Stream } from "effect"; + +import { runGit as runGitProcess } from "../src/git/Process.ts"; +import { CheckpointServiceLive } from "../src/checkpointing/Layers/CheckpointService.ts"; +import { CheckpointStoreLive } from "../src/checkpointing/Layers/CheckpointStore.ts"; +import { ProviderUnsupportedError } from "../src/provider/Errors.ts"; +import { ProviderAdapterRegistry } from "../src/provider/Services/ProviderAdapterRegistry.ts"; +import { ProviderSessionDirectoryLive } from "../src/provider/Layers/ProviderSessionDirectory.ts"; +import { ProviderServiceLive } from "../src/provider/Layers/ProviderService.ts"; +import { + ProviderService, + type ProviderServiceShape, +} from "../src/provider/Services/ProviderService.ts"; +import { SqlitePersistenceMemory } from "../src/persistence/Layers/Sqlite.ts"; +import { CheckpointRepositoryLive } from "../src/persistence/Layers/Checkpoints.ts"; + +import { + makeTestProviderAdapterHarness, + type TestProviderAdapterHarness, + type TestTurnResponse, +} from "./TestProviderAdapter.integration.ts"; +import { + codexTurnApprovalFixture, + codexTurnToolFixture, + codexTurnTextFixture, +} from "./fixtures/providerRuntime.ts"; + +const runGit = (cwd: string, args: ReadonlyArray, allowNonZeroExit = false) => + runGitProcess({ + operation: "providerService.integration.git", + cwd, + args, + allowNonZeroExit, + }).pipe(Effect.provide(NodeServices.layer)); + +const gitRefExists = (cwd: string, ref: string) => + runGit(cwd, ["show-ref", "--verify", "--quiet", ref], true).pipe( + Effect.map((result) => result.code === 0), + ); + +const checkpointRefForTurn = (threadId: string, turnCount: number) => + `refs/t3/checkpoints/${Buffer.from(threadId, "utf8").toString("base64url")}/turn/${turnCount}`; + +const makeGitRepository = Effect.gen(function* () { + const fs = yield* FileSystem.FileSystem; + const pathService = yield* Path.Path; + const cwd = yield* fs.makeTempDirectory(); + + yield* runGit(cwd, ["init", "--initial-branch=main"]); + yield* runGit(cwd, ["config", "user.email", "test@example.com"]); + yield* runGit(cwd, ["config", "user.name", "Test User"]); + yield* fs.writeFileString(pathService.join(cwd, "README.md"), "v1\n"); + yield* runGit(cwd, ["add", "."]); + yield* runGit(cwd, ["commit", "-m", "Initial"]); + + return cwd; +}).pipe(Effect.provide(NodeServices.layer)); + +interface IntegrationFixture { + readonly cwd: string; + readonly harness: TestProviderAdapterHarness; + readonly layer: Layer.Layer; +} + +const makeIntegrationFixture = Effect.gen(function* () { + const cwd = yield* makeGitRepository; + const harness = yield* makeTestProviderAdapterHarness; + + const registry: typeof ProviderAdapterRegistry.Service = { + getByProvider: (provider) => + provider === "codex" + ? Effect.succeed(harness.adapter) + : Effect.fail(new ProviderUnsupportedError({ provider })), + listProviders: () => Effect.succeed(["codex"]), + }; + + const shared = Layer.mergeAll( + ProviderSessionDirectoryLive, + Layer.succeed(ProviderAdapterRegistry, registry), + Layer.provide(CheckpointStoreLive, NodeServices.layer), + Layer.provide(CheckpointRepositoryLive, SqlitePersistenceMemory), + ); + + const checkpointLayer = CheckpointServiceLive.pipe(Layer.provide(shared)); + + const layer = ProviderServiceLive.pipe( + Layer.provide(Layer.mergeAll(shared, checkpointLayer)), + Layer.merge(NodeServices.layer), + ); + + return { + cwd, + harness, + layer, + } satisfies IntegrationFixture; +}); + +const collectEventsDuring = ( + stream: Stream.Stream, + count: number, + action: Effect.Effect, +) => + Effect.gen(function* () { + const queue = yield* Queue.unbounded(); + yield* Stream.runForEach(stream, (event) => Queue.offer(queue, event).pipe(Effect.asVoid)).pipe( + Effect.forkScoped, + ); + + yield* action; + + return yield* Effect.forEach( + Array.from({ length: count }, () => undefined), + () => Queue.take(queue), + { discard: false }, + ); + }); + +const runTurn = (input: { + readonly provider: ProviderServiceShape; + readonly harness: TestProviderAdapterHarness; + readonly sessionId: string; + readonly userText: string; + readonly response: TestTurnResponse; +}) => + Effect.gen(function* () { + yield* input.harness.queueTurnResponse(input.sessionId, input.response); + + return yield* collectEventsDuring( + input.provider.streamEvents, + input.response.events.length + 1, + input.provider.sendTurn({ + sessionId: input.sessionId, + input: input.userText, + attachments: [], + }), + ); + }); + +it.effect("replays typed runtime fixture and emits checkpoint.captured", () => + Effect.gen(function* () { + const fixture = yield* makeIntegrationFixture; + + yield* Effect.gen(function* () { + const provider = yield* ProviderService; + const session = yield* provider.startSession({ + provider: "codex", + cwd: fixture.cwd, + }); + const threadId = session.threadId ?? ""; + assert.equal(threadId.length > 0, true); + + const observedEvents = yield* runTurn({ + provider, + harness: fixture.harness, + sessionId: session.sessionId, + userText: "hello", + response: { events: codexTurnTextFixture }, + }); + + assert.deepEqual( + observedEvents.map((event) => event.type), + [...codexTurnTextFixture.map((event) => event.type), "checkpoint.captured"], + ); + + const checkpoints = yield* provider.listCheckpoints({ sessionId: session.sessionId }); + assert.deepEqual( + checkpoints.checkpoints.map((checkpoint) => checkpoint.turnCount), + [0, 1], + ); + + const diff = yield* provider.getCheckpointDiff({ + sessionId: session.sessionId, + fromTurnCount: 0, + toTurnCount: 1, + }); + assert.equal(diff.diff.trim(), ""); + + const turnOneRefExists = yield* gitRefExists(fixture.cwd, checkpointRefForTurn(threadId, 1)); + assert.equal(turnOneRefExists, true); + }).pipe(Effect.provide(fixture.layer)); + }), +); + +it.effect("captures file-changing fixture turn and produces a non-empty diff", () => + Effect.gen(function* () { + const fixture = yield* makeIntegrationFixture; + const { join } = yield* Path.Path; + const { writeFileString } = yield* FileSystem.FileSystem; + + yield* Effect.gen(function* () { + const provider = yield* ProviderService; + const session = yield* provider.startSession({ + provider: "codex", + cwd: fixture.cwd, + }); + const threadId = session.threadId ?? ""; + assert.equal(threadId.length > 0, true); + + const observedEvents = yield* runTurn({ + provider, + harness: fixture.harness, + sessionId: session.sessionId, + userText: "make a small change", + response: { + events: codexTurnToolFixture, + mutateWorkspace: ({ cwd }) => + writeFileString(join(cwd, "README.md"), "v2\n").pipe(Effect.asVoid, Effect.ignore), + }, + }); + + assert.deepEqual( + observedEvents.map((event) => event.type), + [...codexTurnToolFixture.map((event) => event.type), "checkpoint.captured"], + ); + + const diff = yield* provider.getCheckpointDiff({ + sessionId: session.sessionId, + fromTurnCount: 0, + toTurnCount: 1, + }); + assert.equal(diff.diff.includes("README.md"), true); + + const turnOneRefExists = yield* gitRefExists(fixture.cwd, checkpointRefForTurn(threadId, 1)); + assert.equal(turnOneRefExists, true); + }).pipe(Effect.provide(fixture.layer)); + }).pipe(Effect.provide(NodeServices.layer)), +); + +it.effect("runs multi-turn tool/approval flow with contiguous checkpoints", () => + Effect.gen(function* () { + const fixture = yield* makeIntegrationFixture; + const { join } = yield* Path.Path; + const { writeFileString } = yield* FileSystem.FileSystem; + + yield* Effect.gen(function* () { + const provider = yield* ProviderService; + const session = yield* provider.startSession({ + provider: "codex", + cwd: fixture.cwd, + }); + const threadId = session.threadId ?? ""; + assert.equal(threadId.length > 0, true); + + const firstTurnEvents = yield* runTurn({ + provider, + harness: fixture.harness, + sessionId: session.sessionId, + userText: "turn 1", + response: { + events: codexTurnToolFixture, + mutateWorkspace: ({ cwd }) => + writeFileString(join(cwd, "README.md"), "v2\n").pipe(Effect.asVoid, Effect.ignore), + }, + }); + assert.deepEqual( + firstTurnEvents.map((event) => event.type), + [...codexTurnToolFixture.map((event) => event.type), "checkpoint.captured"], + ); + + const secondTurnEvents = yield* runTurn({ + provider, + harness: fixture.harness, + sessionId: session.sessionId, + userText: "turn 2 approval", + response: { + events: codexTurnApprovalFixture, + mutateWorkspace: ({ cwd }) => + writeFileString(join(cwd, "README.md"), "v3\n").pipe(Effect.asVoid, Effect.ignore), + }, + }); + assert.deepEqual( + secondTurnEvents.map((event) => event.type), + [...codexTurnApprovalFixture.map((event) => event.type), "checkpoint.captured"], + ); + + const turnStateBeforeRevert = yield* provider.listCheckpoints({ + sessionId: session.sessionId, + }); + assert.deepEqual( + turnStateBeforeRevert.checkpoints.map((checkpoint) => checkpoint.turnCount), + [0, 1, 2], + ); + const current = + turnStateBeforeRevert.checkpoints.find((checkpoint) => checkpoint.isCurrent) ?? + turnStateBeforeRevert.checkpoints[turnStateBeforeRevert.checkpoints.length - 1]; + assert.equal(current?.turnCount, 2); + }).pipe(Effect.provide(fixture.layer)); + }).pipe(Effect.provide(NodeServices.layer)), +); + +it.effect("reverts to turn 1 and prunes later checkpoint refs", () => + Effect.gen(function* () { + const fixture = yield* makeIntegrationFixture; + const { join } = yield* Path.Path; + const { writeFileString, readFileString } = yield* FileSystem.FileSystem; + + yield* Effect.gen(function* () { + const provider = yield* ProviderService; + const session = yield* provider.startSession({ + provider: "codex", + cwd: fixture.cwd, + }); + const threadId = session.threadId ?? ""; + assert.equal(threadId.length > 0, true); + + yield* runTurn({ + provider, + harness: fixture.harness, + sessionId: session.sessionId, + userText: "turn 1", + response: { + events: codexTurnToolFixture, + mutateWorkspace: ({ cwd }) => + writeFileString(join(cwd, "README.md"), "v2\n").pipe(Effect.asVoid, Effect.ignore), + }, + }); + + yield* runTurn({ + provider, + harness: fixture.harness, + sessionId: session.sessionId, + userText: "turn 2 approval", + response: { + events: codexTurnApprovalFixture, + mutateWorkspace: ({ cwd }) => + writeFileString(join(cwd, "README.md"), "v3\n").pipe(Effect.asVoid, Effect.ignore), + }, + }); + + const turnTwoRef = checkpointRefForTurn(threadId, 2); + const turnTwoRefBefore = yield* gitRefExists(fixture.cwd, turnTwoRef); + assert.equal(turnTwoRefBefore, true); + + const revert = yield* provider.revertToCheckpoint({ + sessionId: session.sessionId, + turnCount: 1, + }); + assert.equal(revert.rolledBackTurns, 1); + + const rollbackCalls = fixture.harness.getRollbackCalls(session.sessionId); + assert.deepEqual(rollbackCalls, [1]); + + const readme = yield* readFileString(join(fixture.cwd, "README.md")); + assert.equal(readme, "v2\n"); + + const afterRevertCheckpoints = yield* provider.listCheckpoints({ + sessionId: session.sessionId, + }); + assert.deepEqual( + afterRevertCheckpoints.checkpoints.map((checkpoint) => checkpoint.turnCount), + [0, 1], + ); + + const turnTwoRefAfter = yield* gitRefExists(fixture.cwd, turnTwoRef); + assert.equal(turnTwoRefAfter, false); + }).pipe(Effect.provide(fixture.layer)); + }).pipe(Effect.provide(NodeServices.layer)), +); diff --git a/apps/server/package.json b/apps/server/package.json index 4f311dc5817e85aeb39967fc5291049458c7f972..709ef2ba8ad5a1894fcd62e06917914cc93f5dfd 100644 --- a/apps/server/package.json +++ b/apps/server/package.json @@ -11,7 +11,7 @@ "main": "./dist/index.mjs", "scripts": { "dev": "VITE_DEV_SERVER_URL=${VITE_DEV_SERVER_URL:-http://localhost:5173} bun run src/index.ts", - "build": "tsdown && node scripts/bundle-client.mjs", + "build": "tsdown && node scripts/bundle-client.ts", "start": "node dist/index.mjs", "typecheck": "tsc --noEmit", "test": "vitest run" diff --git a/apps/server/scripts/bundle-client.mjs b/apps/server/scripts/bundle-client.mjs deleted file mode 100644 index 33b44cd5e75e9cdadb76f908bc60f1f90abe045d..0000000000000000000000000000000000000000 --- a/apps/server/scripts/bundle-client.mjs +++ /dev/null @@ -1,21 +0,0 @@ -/** - * Copies the built web app into dist/client/ so the published npm package - * includes the web UI. This runs as a post-build step. - */ -import fs from "node:fs"; -import path from "node:path"; -import { fileURLToPath } from "node:url"; - -const __dirname = path.dirname(fileURLToPath(import.meta.url)); -const webDist = path.resolve(__dirname, "../../web/dist"); -const target = path.resolve(__dirname, "../dist/client"); - -if (!fs.existsSync(webDist)) { - console.log( - "⚠ Web dist not found — skipping client bundle. Run `bun run --cwd apps/web build` first.", - ); - process.exit(0); -} - -fs.cpSync(webDist, target, { recursive: true }); -console.log("✓ Bundled web app into dist/client"); diff --git a/apps/server/scripts/bundle-client.ts b/apps/server/scripts/bundle-client.ts new file mode 100644 index 0000000000000000000000000000000000000000..85d5a9aebbdab16b94bb54d6664b24caa0bbd12c --- /dev/null +++ b/apps/server/scripts/bundle-client.ts @@ -0,0 +1,19 @@ +/** + * Copies the built web app into dist/client/ so the published npm package + * includes the web UI. This runs as a post-build step. + */ +import * as fs from "node:fs"; +import * as path from "node:path"; + +const webDist = path.resolve(import.meta.dirname, "../../web/dist"); +const target = path.resolve(import.meta.dirname, "../dist/client"); + +if (!fs.existsSync(webDist)) { + console.log( + "⚠ Web dist not found — skipping client bundle. Run `bun run --cwd apps/web build` first.", + ); + process.exit(0); +} + +fs.cpSync(webDist, target, { recursive: true }); +console.log("✓ Bundled web app into dist/client"); diff --git a/apps/server/src/checkpointing/Layers/CheckpointService.test.ts b/apps/server/src/checkpointing/Layers/CheckpointService.test.ts index 4347a6006c3fc8be3575e750def6e6aeb0571871..7aa4f4ec0999302861787fab9dfcbca48f1d315b 100644 --- a/apps/server/src/checkpointing/Layers/CheckpointService.test.ts +++ b/apps/server/src/checkpointing/Layers/CheckpointService.test.ts @@ -1,6 +1,5 @@ import type { ProviderApprovalDecision, - ProviderEvent, ProviderKind, ProviderSendTurnInput, ProviderSession, @@ -9,7 +8,7 @@ import type { } from "@t3tools/contracts"; import { NodeServices } from "@effect/platform-node"; import { it, assert } from "@effect/vitest"; -import { Effect, FileSystem, Layer, Path } from "effect"; +import { Effect, FileSystem, Layer, Path, Stream } from "effect"; import { runGit as runGitProcess } from "../../git/Process.ts"; import { CheckpointServiceLive } from "./CheckpointService.ts"; @@ -194,9 +193,7 @@ class InMemoryProviderAdapter implements ProviderAdapterShape void, - ): Effect.Effect<() => void, ProviderAdapterError> => Effect.succeed(() => undefined); + readonly streamEvents = Stream.empty; } function makeFixture(turns: ReadonlyArray) { diff --git a/apps/server/src/checkpointing/Layers/CheckpointService.ts b/apps/server/src/checkpointing/Layers/CheckpointService.ts index 2c54cb3c547f236f1ff1db72090b9876b88d1550..c0841ddc2fd4f93754b753fd9ace00caa11cb55b 100644 --- a/apps/server/src/checkpointing/Layers/CheckpointService.ts +++ b/apps/server/src/checkpointing/Layers/CheckpointService.ts @@ -377,6 +377,16 @@ const makeCheckpointService = Effect.gen(function* () { return next; }); + const clearSessionSemaphore = (sessionId: string) => + Ref.update(sessionSemaphoresRef, (current) => { + if (!current.has(sessionId)) { + return current; + } + const next = new Map(current); + next.delete(sessionId); + return next; + }); + const getTrackedCheckpointCwd = (sessionId: string) => Ref.get(sessionCheckpointCwdsRef).pipe( Effect.map((current) => Option.fromNullishOr(current.get(sessionId))), @@ -1024,12 +1034,24 @@ const makeCheckpointService = Effect.gen(function* () { ), ); + const releaseSession: CheckpointServiceShape["releaseSession"] = (input) => + Effect.gen(function* () { + const providerSessionId = yield* normalizeNonEmpty( + input.providerSessionId, + "CheckpointService.releaseSession", + "providerSessionId", + ); + yield* clearTrackedCheckpointCwd(providerSessionId); + yield* clearSessionSemaphore(providerSessionId); + }); + return { initializeForSession, captureCurrentTurn, listCheckpoints, getCheckpointDiff, revertToCheckpoint, + releaseSession, } satisfies CheckpointServiceShape; }); diff --git a/apps/server/src/checkpointing/Services/CheckpointService.ts b/apps/server/src/checkpointing/Services/CheckpointService.ts index 6a5260eb6c9e7993110da84b982893262cb4cc0a..9d782b52d85a6ed5ce3ac4b685b1538b7e535198 100644 --- a/apps/server/src/checkpointing/Services/CheckpointService.ts +++ b/apps/server/src/checkpointing/Services/CheckpointService.ts @@ -69,6 +69,16 @@ export interface CheckpointServiceShape { readonly revertToCheckpoint: ( input: ProviderRevertToCheckpointInput, ) => Effect.Effect; + + /** + * Release in-memory checkpoint session state after session shutdown. + * + * This only clears local coordination state (locks/cwd tracking). It does not + * delete durable checkpoint metadata or git refs. + */ + readonly releaseSession: (input: { + readonly providerSessionId: string; + }) => Effect.Effect; } /** diff --git a/apps/server/src/orchestration/Layer.ts b/apps/server/src/orchestration/Layer.ts index 1edacd925c0f36d46d6fbe02f0395f3ff7b437fb..6847c693d3302e4bd7a94b38d0e4101be8cbd42f 100644 --- a/apps/server/src/orchestration/Layer.ts +++ b/apps/server/src/orchestration/Layer.ts @@ -143,6 +143,19 @@ function mapCommandToEvent(command: OrchestrationCommand): Omit Effect.Effect; + readonly getSnapshot: () => Effect.Effect; /** * Replay persisted orchestration events from an exclusive sequence cursor. @@ -37,7 +37,7 @@ export interface OrchestrationEngineShape { */ readonly replayEvents: ( fromSequenceExclusive: number, - ) => Effect.Effect; + ) => Effect.Effect; /** * Validate and dispatch an unknown command payload. @@ -49,7 +49,7 @@ export interface OrchestrationEngineShape { */ readonly dispatchUnknown: ( command: unknown, - ) => Effect.Effect<{ sequence: number }, OrchestrationEngineError>; + ) => Effect.Effect<{ sequence: number }, OrchestrationEngineError, never>; /** * Dispatch a validated orchestration command. @@ -59,7 +59,7 @@ export interface OrchestrationEngineShape { */ readonly dispatch: ( command: OrchestrationCommand, - ) => Effect.Effect<{ sequence: number }, OrchestrationDispatchError>; + ) => Effect.Effect<{ sequence: number }, OrchestrationDispatchError, never>; /** * Subscribe to read model updates. @@ -69,7 +69,7 @@ export interface OrchestrationEngineShape { */ readonly subscribeToReadModel: ( callback: (snapshot: OrchestrationReadModel) => void, - ) => Effect.Effect<() => void>; + ) => Effect.Effect<() => void, never, never>; /** * Subscribe to domain event fan-out. @@ -79,7 +79,7 @@ export interface OrchestrationEngineShape { */ readonly subscribeToDomainEvents: ( callback: (event: OrchestrationEvent) => void, - ) => Effect.Effect<() => void>; + ) => Effect.Effect<() => void, never, never>; } /** diff --git a/apps/server/src/orchestration/reducer.test.ts b/apps/server/src/orchestration/reducer.test.ts index ac887e8dd925dbb6798e9446e5ebce0ddbbdf260..bf3cc633d72ea0b3893da09b04ac0c053111242b 100644 --- a/apps/server/src/orchestration/reducer.test.ts +++ b/apps/server/src/orchestration/reducer.test.ts @@ -46,6 +46,7 @@ describe("orchestration reducer", () => { latestTurnDurationMs: null, messages: [], session: null, + activities: [], turnDiffSummaries: [], error: null, }, diff --git a/apps/server/src/orchestration/reducer.ts b/apps/server/src/orchestration/reducer.ts index 5ba4334e0fc7ced6e0cdf891fd7e5c9a171825d3..40320779c1f6af4979ff69071f6b97342d156c0d 100644 --- a/apps/server/src/orchestration/reducer.ts +++ b/apps/server/src/orchestration/reducer.ts @@ -7,6 +7,7 @@ import type { OrchestrationThread, } from "@t3tools/contracts"; import { + OrchestrationThreadActivitySchema, OrchestrationGitReadModelSchema, OrchestrationMessageSchema, OrchestrationSessionSchema, @@ -89,6 +90,11 @@ const ThreadRevertedPayloadSchema = Schema.Struct({ messageCount: Schema.Number, }); +const ThreadActivityAppendedPayloadSchema = Schema.Struct({ + threadId: Schema.String, + activity: OrchestrationThreadActivitySchema, +}); + export function createEmptyReadModel(nowIso: string): OrchestrationReadModel { return { sequence: 0, @@ -135,6 +141,7 @@ export function reduceEvent( messages: [], session: null, turnDiffSummaries: [], + activities: [], error: null, }, event.type, @@ -214,11 +221,13 @@ export function reduceEvent( ? message.text : entry.text, streaming: message.streaming, - createdAt: message.createdAt, + createdAt: entry.createdAt, } : entry, ) - : [...targetThread.messages, message]; + : !message.streaming && message.text.length === 0 + ? targetThread.messages + : [...targetThread.messages, message]; return { ...nextBase, @@ -334,6 +343,16 @@ export function reduceEvent( .toSorted((left, right) => left.completedAt.localeCompare(right.completedAt)); const latestSummary = turnDiffSummaries.length > 0 ? turnDiffSummaries[turnDiffSummaries.length - 1] : null; + const activities = thread.activities.filter((activity) => { + if (!activity.turnId) { + return true; + } + const checkpointTurnCount = inferredTurnCountByTurnId.get(activity.turnId); + if (checkpointTurnCount === undefined) { + return true; + } + return checkpointTurnCount <= targetTurnCount; + }); return { ...nextBase, @@ -347,6 +366,35 @@ export function reduceEvent( latestTurnStartedAt: null, latestTurnCompletedAt: latestSummary?.completedAt ?? null, latestTurnDurationMs: null, + activities, + updatedAt: event.occurredAt, + }), + }; + }), + ); + case "thread.activity-appended": + return decodeForEvent( + ThreadActivityAppendedPayloadSchema, + event.payload, + event.type, + "payload", + ).pipe( + Effect.map((payload) => { + const thread = nextBase.threads.find((entry) => entry.id === payload.threadId); + if (!thread) { + return nextBase; + } + const activities = [ + ...thread.activities.filter((activity) => activity.id !== payload.activity.id), + payload.activity, + ] + .toSorted((left, right) => left.createdAt.localeCompare(right.createdAt)) + .slice(-500); + + return { + ...nextBase, + threads: updateThread(nextBase.threads, payload.threadId, { + activities, updatedAt: event.occurredAt, }), }; diff --git a/apps/server/src/provider/Layers/CodexAdapter.test.ts b/apps/server/src/provider/Layers/CodexAdapter.test.ts index 3e12a2a3e7c53b1a4e220a68d005393a3a4ef91f..632fd0a922126f1a1bbdf7980c5f78a608bc80c0 100644 --- a/apps/server/src/provider/Layers/CodexAdapter.test.ts +++ b/apps/server/src/provider/Layers/CodexAdapter.test.ts @@ -9,7 +9,7 @@ import type { import { afterAll, assert, it, vi } from "@effect/vitest"; import { assertFailure } from "@effect/vitest/utils"; -import { Effect } from "effect"; +import { Effect, Fiber, Stream } from "effect"; import { CodexAppServerManager } from "../../codexAppServerManager.ts"; import { ProviderAdapterValidationError } from "../Errors.ts"; @@ -170,11 +170,15 @@ const lifecycleManager = new FakeCodexManager(); const lifecycleLayer = it.layer(makeCodexAdapterLive({ manager: lifecycleManager })); lifecycleLayer("CodexAdapterLive lifecycle", (it) => { - it.effect("subscribes and unsubscribes from manager events", () => + it.effect("emits canonical events on stream and supports consumer interruption", () => Effect.gen(function* () { const adapter = yield* CodexAdapter; - const onEvent = vi.fn((_event: ProviderEvent) => undefined); - const unsubscribe = yield* adapter.subscribeToEvents(onEvent); + const onEvent = vi.fn(() => undefined); + const consumer = yield* Stream.runForEach(adapter.streamEvents, () => + Effect.sync(() => { + onEvent(); + }), + ).pipe(Effect.forkChild); const event: ProviderEvent = { id: "evt-1", @@ -183,14 +187,16 @@ lifecycleLayer("CodexAdapterLive lifecycle", (it) => { sessionId: "sess-1", createdAt: new Date().toISOString(), method: "turn/started", + turnId: "turn-1", }; lifecycleManager.emit("event", event); + yield* Effect.promise(() => new Promise((resolve) => setTimeout(resolve, 0))); assert.equal(onEvent.mock.calls.length, 1); - assert.equal(onEvent.mock.calls[0]?.[0]?.method, "turn/started"); - unsubscribe(); + yield* Fiber.interrupt(consumer); lifecycleManager.emit("event", event); + yield* Effect.promise(() => new Promise((resolve) => setTimeout(resolve, 0))); assert.equal(onEvent.mock.calls.length, 1); assert.equal(lifecycleManager.stopAllImpl.mock.calls.length, 0); }), diff --git a/apps/server/src/provider/Layers/CodexAdapter.ts b/apps/server/src/provider/Layers/CodexAdapter.ts index 2a3307a9626a9e91afc636f949a49a9ec65c9880..d8055a54460a0704c339f9ebc58ed0c19f0ad086 100644 --- a/apps/server/src/provider/Layers/CodexAdapter.ts +++ b/apps/server/src/provider/Layers/CodexAdapter.ts @@ -6,8 +6,13 @@ * * @module CodexAdapterLive */ -import type { ProviderEvent } from "@t3tools/contracts"; -import { Effect, Layer } from "effect"; +import type { + ProviderEvent, + ProviderRuntimeEvent, + ProviderRuntimeTurnStatus, + ProviderRuntimeToolKind, +} from "@t3tools/contracts"; +import { Effect, Layer, Queue, Stream } from "effect"; import { ProviderAdapterProcessError, @@ -19,12 +24,14 @@ import { } from "../Errors.ts"; import { CodexAdapter, type CodexAdapterShape } from "../Services/CodexAdapter.ts"; import { CodexAppServerManager } from "../../codexAppServerManager.ts"; +import { makeEventNdjsonLogger } from "./EventNdjsonLogger.ts"; const PROVIDER = "codex" as const; export interface CodexAdapterLiveOptions { readonly manager?: CodexAppServerManager; readonly makeManager?: () => CodexAppServerManager; + readonly nativeEventLogPath?: string; } function toMessage(cause: unknown, fallback: string): string { @@ -69,8 +76,309 @@ function toRequestError(sessionId: string, method: string, cause: unknown): Prov }); } +function asObject(value: unknown): Record | undefined { + if (!value || typeof value !== "object") { + return undefined; + } + return value as Record; +} + +function asString(value: unknown): string | undefined { + return typeof value === "string" ? value : undefined; +} + +function toTurnStatus(value: unknown): ProviderRuntimeTurnStatus | undefined { + switch (value) { + case "completed": + case "failed": + case "cancelled": + case "interrupted": + return value; + default: + return undefined; + } +} + +function normalizeItemType(raw: unknown): string { + const type = asString(raw); + if (!type) return "item"; + return type + .replace(/([a-z0-9])([A-Z])/g, "$1 $2") + .replace(/[._/-]/g, " ") + .replace(/\s+/g, " ") + .trim() + .toLowerCase(); +} + +function shouldDropItemType(type: string): boolean { + if (type.includes("preamble") || type.includes("reasoning") || type.includes("thought")) { + return true; + } + return type === "work" || type.startsWith("work "); +} + +function toolMeta(type: string): + | { + readonly toolKind: ProviderRuntimeToolKind; + readonly title: string; + } + | undefined { + if (type.includes("command")) { + return { toolKind: "command", title: "Command run" }; + } + if (type.includes("file change")) { + return { toolKind: "file-change", title: "File change" }; + } + if (type.includes("tool")) { + return { toolKind: "other", title: "Tool call" }; + } + return undefined; +} + +function itemDetail( + item: Record, + payload: Record, +): string | undefined { + const nestedResult = asObject(item.result); + const candidates = [ + asString(item.command), + asString(item.title), + asString(item.summary), + asString(item.text), + asString(item.path), + asString(item.prompt), + asString(nestedResult?.command), + asString(payload.command), + asString(payload.message), + asString(payload.prompt), + ]; + for (const candidate of candidates) { + if (!candidate) continue; + const trimmed = candidate.trim(); + if (trimmed.length === 0) continue; + return trimmed; + } + return undefined; +} + +function mapToolEvent(event: ProviderEvent): ProviderRuntimeEvent | undefined { + if (event.method !== "item/started" && event.method !== "item/completed") { + return undefined; + } + const payload = asObject(event.payload); + const item = asObject(payload?.item); + if (!item) { + return undefined; + } + const normalizedType = normalizeItemType(item.type ?? item.kind); + if (shouldDropItemType(normalizedType)) { + return undefined; + } + const meta = toolMeta(normalizedType); + if (!meta) { + return undefined; + } + + const eventBase = { + eventId: event.id, + provider: event.provider, + sessionId: event.sessionId, + createdAt: event.createdAt, + ...(event.threadId ? { threadId: event.threadId } : {}), + ...(event.turnId ? { turnId: event.turnId } : {}), + ...(event.itemId ? { itemId: event.itemId } : {}), + toolKind: meta.toolKind, + title: meta.title, + ...(payload ? { detail: itemDetail(item, payload) } : {}), + } as const; + + if (event.method === "item/started") { + return { + type: "tool.started", + ...eventBase, + }; + } + + return { + type: "tool.completed", + ...eventBase, + }; +} + +function mapToRuntimeEvents(event: ProviderEvent): ReadonlyArray { + const payload = asObject(event.payload); + const turn = asObject(payload?.turn); + + if (event.kind === "error") { + if (!event.message) { + return []; + } + return [ + { + type: "runtime.error", + eventId: event.id, + provider: event.provider, + sessionId: event.sessionId, + createdAt: event.createdAt, + ...(event.threadId ? { threadId: event.threadId } : {}), + ...(event.turnId ? { turnId: event.turnId } : {}), + ...(event.itemId ? { itemId: event.itemId } : {}), + message: event.message, + }, + ]; + } + + if (event.kind === "request" && event.requestId && event.requestKind) { + const detail = + asString(payload?.command) ?? asString(payload?.reason) ?? asString(payload?.prompt); + return [ + { + type: "approval.requested", + eventId: event.id, + provider: event.provider, + sessionId: event.sessionId, + createdAt: event.createdAt, + ...(event.threadId ? { threadId: event.threadId } : {}), + ...(event.turnId ? { turnId: event.turnId } : {}), + ...(event.itemId ? { itemId: event.itemId } : {}), + requestId: event.requestId, + requestKind: event.requestKind, + ...(detail ? { detail } : {}), + }, + ]; + } + + if (event.method === "item/requestApproval/decision" && event.requestId) { + const decision = asString(payload?.decision); + return [ + { + type: "approval.resolved", + eventId: event.id, + provider: event.provider, + sessionId: event.sessionId, + createdAt: event.createdAt, + ...(event.threadId ? { threadId: event.threadId } : {}), + ...(event.turnId ? { turnId: event.turnId } : {}), + ...(event.itemId ? { itemId: event.itemId } : {}), + requestId: event.requestId, + ...(event.requestKind ? { requestKind: event.requestKind } : {}), + ...(decision ? { decision } : {}), + }, + ]; + } + + const tool = mapToolEvent(event); + if (tool) { + return [tool]; + } + + if (event.method === "session/started") { + return [ + { + type: "session.started", + eventId: event.id, + provider: event.provider, + sessionId: event.sessionId, + createdAt: event.createdAt, + ...(event.threadId ? { threadId: event.threadId } : {}), + ...(event.message ? { message: event.message } : {}), + }, + ]; + } + + if (event.method === "session/exited") { + return [ + { + type: "session.exited", + eventId: event.id, + provider: event.provider, + sessionId: event.sessionId, + createdAt: event.createdAt, + ...(event.threadId ? { threadId: event.threadId } : {}), + ...(event.message ? { message: event.message } : {}), + }, + ]; + } + + if (event.method === "thread/started") { + const threadId = event.threadId ?? asString(asObject(payload?.thread)?.id); + if (!threadId) { + return []; + } + return [ + { + type: "thread.started", + eventId: event.id, + provider: event.provider, + sessionId: event.sessionId, + createdAt: event.createdAt, + threadId, + }, + ]; + } + + if (event.method === "turn/started") { + const turnId = event.turnId ?? asString(turn?.id); + if (!turnId) { + return []; + } + return [ + { + type: "turn.started", + eventId: event.id, + provider: event.provider, + sessionId: event.sessionId, + createdAt: event.createdAt, + ...(event.threadId ? { threadId: event.threadId } : {}), + turnId, + }, + ]; + } + + if (event.method === "turn/completed") { + return [ + { + type: "turn.completed", + eventId: event.id, + provider: event.provider, + sessionId: event.sessionId, + createdAt: event.createdAt, + ...(event.threadId ? { threadId: event.threadId } : {}), + ...(event.turnId ? { turnId: event.turnId } : {}), + ...(toTurnStatus(turn?.status) ? { status: toTurnStatus(turn?.status) } : {}), + ...(asString(asObject(turn?.error)?.message) + ? { errorMessage: asString(asObject(turn?.error)?.message) } + : {}), + }, + ]; + } + + if (event.method === "item/agentMessage/delta" && event.textDelta && event.textDelta.length > 0) { + return [ + { + type: "message.delta", + eventId: event.id, + provider: event.provider, + sessionId: event.sessionId, + createdAt: event.createdAt, + ...(event.threadId ? { threadId: event.threadId } : {}), + ...(event.turnId ? { turnId: event.turnId } : {}), + ...(event.itemId ? { itemId: event.itemId } : {}), + delta: event.textDelta, + }, + ]; + } + + return []; +} + const makeCodexAdapter = (options?: CodexAdapterLiveOptions) => Effect.gen(function* () { + const nativeEventLogger = + options?.nativeEventLogPath !== undefined + ? makeEventNdjsonLogger(options.nativeEventLogPath) + : undefined; + const manager = yield* Effect.acquireRelease( Effect.sync(() => { if (options?.manager) { @@ -175,20 +483,28 @@ const makeCodexAdapter = (options?: CodexAdapterLiveOptions) => manager.stopAll(); }); - const subscribeToEvents: CodexAdapterShape["subscribeToEvents"] = (callback) => + const runtimeEventQueue = yield* Queue.unbounded(); + + yield* Effect.acquireRelease( Effect.sync(() => { const listener = (event: ProviderEvent) => { - try { - callback(event); - } catch { - // Adapter subscribers must not destabilize provider event processing. - } + nativeEventLogger?.write({ + observedAt: new Date().toISOString(), + event, + }); + Queue.offerAllUnsafe(runtimeEventQueue, mapToRuntimeEvents(event)); }; manager.on("event", listener); - return () => { - manager.off("event", listener); - }; - }); + return listener; + }), + (listener) => + Effect.gen(function* () { + yield* Effect.sync(() => { + manager.off("event", listener); + }); + yield* Queue.shutdown(runtimeEventQueue); + }), + ); return { provider: PROVIDER, @@ -202,7 +518,7 @@ const makeCodexAdapter = (options?: CodexAdapterLiveOptions) => listSessions, hasSession, stopAll, - subscribeToEvents, + streamEvents: Stream.fromQueue(runtimeEventQueue), } satisfies CodexAdapterShape; }); diff --git a/apps/server/src/provider/Layers/EventNdjsonLogger.test.ts b/apps/server/src/provider/Layers/EventNdjsonLogger.test.ts new file mode 100644 index 0000000000000000000000000000000000000000..73e110295f3335680569cd42e793b5677ccf2b1d --- /dev/null +++ b/apps/server/src/provider/Layers/EventNdjsonLogger.test.ts @@ -0,0 +1,37 @@ +import fs from "node:fs"; +import os from "node:os"; +import path from "node:path"; + +import { describe, expect, it } from "vitest"; + +import { makeEventNdjsonLogger } from "./EventNdjsonLogger.ts"; + +describe("EventNdjsonLogger", () => { + it("writes NDJSON records to disk", () => { + const tempDir = fs.mkdtempSync(path.join(os.tmpdir(), "t3-provider-log-")); + const filePath = path.join(tempDir, "provider-native.ndjson"); + + try { + const logger = makeEventNdjsonLogger(filePath); + expect(logger).toBeDefined(); + if (!logger) { + return; + } + + logger.write({ type: "native", payload: { id: "evt-1" } }); + logger.write({ type: "canonical", payload: { id: "evt-2" } }); + logger.close(); + + const raw = fs.readFileSync(filePath, "utf8").trim(); + const lines = raw.split("\n"); + expect(lines.length).toBe(2); + + const first = JSON.parse(lines[0] ?? "{}"); + const second = JSON.parse(lines[1] ?? "{}"); + expect(first).toEqual({ type: "native", payload: { id: "evt-1" } }); + expect(second).toEqual({ type: "canonical", payload: { id: "evt-2" } }); + } finally { + fs.rmSync(tempDir, { recursive: true, force: true }); + } + }); +}); diff --git a/apps/server/src/provider/Layers/EventNdjsonLogger.ts b/apps/server/src/provider/Layers/EventNdjsonLogger.ts new file mode 100644 index 0000000000000000000000000000000000000000..b842f076cb5d706510a58a23b88c10e27f5e8259 --- /dev/null +++ b/apps/server/src/provider/Layers/EventNdjsonLogger.ts @@ -0,0 +1,66 @@ +/** + * Provider event NDJSON logger helper. + * + * Best-effort writer for observability logs. Failures are downgraded to warnings + * so provider runtime behavior is unaffected. + */ +import fs from "node:fs"; +import path from "node:path"; + +import { createLogger } from "../../logger.ts"; + +const logger = createLogger("provider-observability"); + +export interface EventNdjsonLogger { + readonly filePath: string; + write: (record: unknown) => void; + close: () => void; +} + +function stringifyRecord(record: unknown): string | undefined { + try { + return JSON.stringify(record); + } catch (error) { + logger.warn("failed to serialize provider event log record", { error }); + return undefined; + } +} + +export function makeEventNdjsonLogger(filePath: string): EventNdjsonLogger | undefined { + try { + fs.mkdirSync(path.dirname(filePath), { recursive: true }); + fs.appendFileSync(filePath, "", { encoding: "utf8", flag: "a" }); + } catch (error) { + logger.warn("failed to create provider event log directory", { filePath, error }); + return undefined; + } + + try { + let broken = false; + + return { + filePath, + write(record: unknown) { + if (broken) { + return; + } + const serialized = stringifyRecord(record); + if (!serialized) { + return; + } + try { + fs.appendFileSync(filePath, `${serialized}\n`, { encoding: "utf8", flag: "a" }); + } catch (error) { + broken = true; + logger.warn("provider event log write failed", { filePath, error }); + } + }, + close() { + // no-op for sync append logger + }, + }; + } catch (error) { + logger.warn("failed to create provider event log file", { filePath, error }); + return undefined; + } +} diff --git a/apps/server/src/provider/Layers/ProviderAdapterRegistry.test.ts b/apps/server/src/provider/Layers/ProviderAdapterRegistry.test.ts index 25d8cf07ac176b079ea8b7f81e33eb57873e0813..48c1a76538b68bd4e07bf67f4fab4b5e02a1bc3a 100644 --- a/apps/server/src/provider/Layers/ProviderAdapterRegistry.test.ts +++ b/apps/server/src/provider/Layers/ProviderAdapterRegistry.test.ts @@ -2,7 +2,7 @@ import type { ProviderKind } from "@t3tools/contracts"; import { it, assert, vi } from "@effect/vitest"; import { assertFailure } from "@effect/vitest/utils"; -import { Effect, Layer } from "effect"; +import { Effect, Layer, Stream } from "effect"; import { CodexAdapter, CodexAdapterShape } from "../Services/CodexAdapter.ts"; import { ProviderAdapterRegistry } from "../Services/ProviderAdapterRegistry.ts"; @@ -22,7 +22,7 @@ const fakeCodexAdapter: CodexAdapterShape = { readThread: vi.fn(), rollbackThread: vi.fn(), stopAll: vi.fn(), - subscribeToEvents: vi.fn(), + streamEvents: Stream.empty, }; const layer = it.layer( diff --git a/apps/server/src/provider/Layers/ProviderService.test.ts b/apps/server/src/provider/Layers/ProviderService.test.ts index 2d7ecb4e2eb36f91f6c294197eb9658b1d50bea6..46182941b8c01c0b467b0253be76aa4d487b1fa2 100644 --- a/apps/server/src/provider/Layers/ProviderService.test.ts +++ b/apps/server/src/provider/Layers/ProviderService.test.ts @@ -1,11 +1,11 @@ import type { ProviderApprovalDecision, - ProviderEvent, ProviderGetCheckpointDiffInput, ProviderGetCheckpointDiffResult, ProviderListCheckpointsInput, ProviderListCheckpointsResult, ProviderRevertToCheckpointInput, + ProviderRuntimeEvent, ProviderRevertToCheckpointResult, ProviderSendTurnInput, ProviderSession, @@ -16,7 +16,7 @@ import { providerSessionStartInputSchema } from "@t3tools/contracts"; import { it, assert, vi } from "@effect/vitest"; import { assertFailure } from "@effect/vitest/utils"; -import { Effect, Layer } from "effect"; +import { Effect, Fiber, Layer, PubSub, Ref, Stream } from "effect"; import { CheckpointService } from "../../checkpointing/Services/CheckpointService.ts"; import { @@ -35,7 +35,7 @@ import { NodeServices } from "@effect/platform-node"; function makeFakeCodexAdapter() { const sessions = new Map(); - const eventListeners = new Set<(event: ProviderEvent) => void>(); + const runtimeEventPubSub = Effect.runSync(PubSub.unbounded()); let nextSession = 1; const startSession = vi.fn((input: ProviderSessionStartInput) => @@ -127,16 +127,6 @@ function makeFakeCodexAdapter() { }), ); - const subscribeToEvents = vi.fn( - (callback: (event: ProviderEvent) => void): Effect.Effect<() => void, ProviderAdapterError> => - Effect.sync(() => { - eventListeners.add(callback); - return () => { - eventListeners.delete(callback); - }; - }), - ); - const adapter: ProviderAdapterShape = { provider: "codex", startSession, @@ -149,13 +139,11 @@ function makeFakeCodexAdapter() { readThread, rollbackThread, stopAll, - subscribeToEvents, + streamEvents: Stream.fromPubSub(runtimeEventPubSub), }; - const emit = (event: ProviderEvent): void => { - for (const listener of eventListeners) { - listener(event); - } + const emit = (event: ProviderRuntimeEvent): void => { + Effect.runSync(PubSub.publish(runtimeEventPubSub, event)); }; return { @@ -171,7 +159,6 @@ function makeFakeCodexAdapter() { readThread, rollbackThread, stopAll, - subscribeToEvents, }; } @@ -224,6 +211,7 @@ function makeCheckpointServiceDouble() { ], }), ); + const releaseSession = vi.fn((_input: { providerSessionId: string }) => Effect.void); const service: typeof CheckpointService.Service = { initializeForSession, @@ -231,6 +219,7 @@ function makeCheckpointServiceDouble() { listCheckpoints, getCheckpointDiff, revertToCheckpoint, + releaseSession, }; return { @@ -240,6 +229,7 @@ function makeCheckpointServiceDouble() { listCheckpoints, getCheckpointDiff, revertToCheckpoint, + releaseSession, }; } @@ -334,6 +324,9 @@ routing.layer("ProviderServiceLive routing", (it) => { ]); yield* provider.stopSession({ sessionId: session.sessionId }); + assert.deepEqual(routing.checkpoint.releaseSession.mock.calls, [ + [{ providerSessionId: session.sessionId }], + ]); const sendAfterStop = yield* Effect.result( provider.sendTurn({ sessionId: session.sessionId, @@ -347,6 +340,29 @@ routing.layer("ProviderServiceLive routing", (it) => { ); }), ); + + it.effect("releases checkpoint session state when stopping all sessions", () => + Effect.gen(function* () { + const provider = yield* ProviderService; + + const first = yield* provider.startSession({ + provider: "codex", + }); + const second = yield* provider.startSession({ + provider: "codex", + }); + + yield* provider.stopAll(); + + assert.deepEqual(routing.checkpoint.releaseSession.mock.calls.slice(-2), [ + [{ providerSessionId: first.sessionId }], + [{ providerSessionId: second.sessionId }], + ]); + + const remaining = yield* provider.listSessions(); + assert.equal(remaining.length, 0); + }), + ); }); const fanout = makeProviderServiceLayer(); @@ -358,40 +374,111 @@ fanout.layer("ProviderServiceLive fanout", (it) => { provider: "codex", }); - const onEvent = vi.fn((_event: ProviderEvent) => undefined); - const unsubscribe = yield* provider.subscribeToEvents(onEvent); + const eventsRef = yield* Ref.make>([]); + const consumer = yield* Stream.runForEach(provider.streamEvents, (event) => + Ref.update(eventsRef, (current) => [...current, event]), + ).pipe( + Effect.forkChild, + ); + yield* Effect.promise(() => new Promise((resolve) => setTimeout(resolve, 0))); - const completedEvent: ProviderEvent = { - id: "evt-1", - kind: "notification", + const completedEvent: ProviderRuntimeEvent = { + type: "turn.completed", + eventId: "evt-1", provider: "codex", sessionId: session.sessionId, createdAt: new Date().toISOString(), - method: "turn/completed", threadId: "thread-1", turnId: "turn-1", - payload: { - turn: { - status: "completed", - }, - }, + status: "completed", }; fanout.codex.emit(completedEvent); yield* Effect.promise(() => new Promise((resolve) => setTimeout(resolve, 0))); - assert.equal(onEvent.mock.calls.length, 1); - assert.equal(onEvent.mock.calls[0]?.[0]?.method, "turn/completed"); + const events = yield* Ref.get(eventsRef); + yield* Fiber.interrupt(consumer); + + assert.equal(events.length >= 2, true); + assert.equal(events[0]?.type, "turn.completed"); + assert.equal(events[1]?.type, "checkpoint.captured"); assert.deepEqual(fanout.checkpoint.captureCurrentTurn.mock.calls, [ [{ providerSessionId: session.sessionId, turnId: "turn-1", status: "completed" }], ]); + }), + ); - unsubscribe(); - fanout.codex.emit({ - ...completedEvent, - id: "evt-2", + it.effect("keeps subscriber delivery ordered and isolates failing subscribers", () => + Effect.gen(function* () { + const provider = yield* ProviderService; + const session = yield* provider.startSession({ + provider: "codex", }); - assert.equal(onEvent.mock.calls.length, 1); + fanout.checkpoint.captureCurrentTurn.mockClear(); + + const receivedByHealthy: string[] = []; + const healthyFiber = yield* Stream.take(provider.streamEvents, 3).pipe( + Stream.runForEach((event) => + Effect.sync(() => { + receivedByHealthy.push(event.eventId); + }), + ), + Effect.forkChild, + ); + const failingFiber = yield* Stream.take(provider.streamEvents, 1).pipe( + Stream.runForEach(() => Effect.fail("listener crash")), + Effect.forkChild, + ); + yield* Effect.promise(() => new Promise((resolve) => setTimeout(resolve, 0))); + + const events: ReadonlyArray = [ + { + type: "tool.completed", + eventId: "evt-ordered-1", + provider: "codex", + sessionId: session.sessionId, + createdAt: new Date().toISOString(), + threadId: "thread-1", + turnId: "turn-1", + toolKind: "command", + title: "Command run", + detail: "echo one", + }, + { + type: "message.delta", + eventId: "evt-ordered-2", + provider: "codex", + sessionId: session.sessionId, + createdAt: new Date().toISOString(), + threadId: "thread-1", + turnId: "turn-1", + delta: "hello", + }, + { + type: "turn.completed", + eventId: "evt-ordered-3", + provider: "codex", + sessionId: session.sessionId, + createdAt: new Date().toISOString(), + threadId: "thread-1", + turnId: "turn-1", + status: "completed", + }, + ]; + + for (const event of events) { + fanout.codex.emit(event); + } + const failingResult = yield* Effect.result(Fiber.join(failingFiber)); + assert.equal(failingResult._tag, "Failure"); + yield* Fiber.join(healthyFiber); + + assert.deepEqual(receivedByHealthy.slice(0, 3), [ + "evt-ordered-1", + "evt-ordered-2", + "evt-ordered-3", + ]); + assert.equal(receivedByHealthy.length, 3); }), ); }); diff --git a/apps/server/src/provider/Layers/ProviderService.ts b/apps/server/src/provider/Layers/ProviderService.ts index f445d157d41702ae5ff0e16f53744c42076732f8..e1f7d52d694e828f76d6cf68bab54b3712af9400 100644 --- a/apps/server/src/provider/Layers/ProviderService.ts +++ b/apps/server/src/provider/Layers/ProviderService.ts @@ -12,7 +12,6 @@ */ import { randomUUID } from "node:crypto"; -import type { ProviderEvent } from "@t3tools/contracts"; import { providerGetCheckpointDiffInputSchema, providerInterruptTurnInputSchema, @@ -22,46 +21,54 @@ import { providerSendTurnInputSchema, providerSessionStartInputSchema, providerStopSessionInputSchema, + type ProviderRuntimeEvent, + type ProviderRuntimeTurnCompletedEvent, } from "@t3tools/contracts"; -import { Effect, Layer } from "effect"; +import { Effect, Layer, PubSub, Queue, Stream } from "effect"; import { CheckpointService } from "../../checkpointing/Services/CheckpointService.ts"; import { ProviderValidationError } from "../Errors.ts"; import { ProviderAdapterRegistry } from "../Services/ProviderAdapterRegistry.ts"; import { ProviderService, type ProviderServiceShape } from "../Services/ProviderService.ts"; import { ProviderSessionDirectory } from "../Services/ProviderSessionDirectory.ts"; +import { makeEventNdjsonLogger } from "./EventNdjsonLogger.ts"; -function asObject(value: unknown): Record | undefined { - if (!value || typeof value !== "object") { - return undefined; - } - return value as Record; -} - -function asString(value: unknown): string | undefined { - return typeof value === "string" ? value : undefined; -} - -function readTurnCompletionStatus(event: ProviderEvent): string | undefined { - const payload = asObject(event.payload); - const turn = asObject(payload?.turn); - return asString(turn?.status); +export interface ProviderServiceLiveOptions { + readonly canonicalEventLogPath?: string; } function toCheckpointCaptureErrorEvent( - event: ProviderEvent, + event: ProviderRuntimeTurnCompletedEvent, error: { readonly message: string }, -): ProviderEvent { +): ProviderRuntimeEvent { return { - id: randomUUID(), - kind: "error", + type: "runtime.error", + eventId: randomUUID(), provider: event.provider, sessionId: event.sessionId, createdAt: new Date().toISOString(), - method: "checkpoint/captureError", - message: error.message, ...(event.threadId !== undefined ? { threadId: event.threadId } : {}), ...(event.turnId !== undefined ? { turnId: event.turnId } : {}), + message: error.message, + }; +} + +function toCheckpointCapturedEvent(input: { + readonly event: ProviderRuntimeTurnCompletedEvent; + readonly threadId: string; + readonly turnCount: number; +}): ProviderRuntimeEvent { + const { event, threadId, turnCount } = input; + return { + type: "checkpoint.captured", + eventId: randomUUID(), + provider: event.provider, + sessionId: event.sessionId, + createdAt: new Date().toISOString(), + threadId, + ...(event.turnId !== undefined ? { turnId: event.turnId } : {}), + turnCount, + ...(event.status !== undefined ? { status: event.status } : {}), }; } @@ -77,58 +84,80 @@ function toValidationError( }); } -const makeProviderService = Effect.gen(function* () { - const registry = yield* ProviderAdapterRegistry; - const directory = yield* ProviderSessionDirectory; - const checkpointService = yield* CheckpointService; +const makeProviderService = (options?: ProviderServiceLiveOptions) => + Effect.gen(function* () { + const canonicalEventLogger = + options?.canonicalEventLogPath !== undefined + ? makeEventNdjsonLogger(options.canonicalEventLogPath) + : undefined; - const eventSubscribers = new Set<(event: ProviderEvent) => void>(); - const notifySubscribers = (event: ProviderEvent): void => { - for (const callback of eventSubscribers) { - try { - callback(event); - } catch { - // Subscriber callbacks must not destabilize provider event fanout. - } - } - }; + const registry = yield* ProviderAdapterRegistry; + const directory = yield* ProviderSessionDirectory; + const checkpointService = yield* CheckpointService; - const providers = yield* registry.listProviders(); - const adapters = yield* Effect.forEach(providers, (provider) => registry.getByProvider(provider)); + const runtimeEventQueue = yield* Queue.unbounded(); + const runtimeEventPubSub = yield* PubSub.unbounded(); - yield* Effect.acquireRelease( - Effect.forEach(adapters, (adapter) => - adapter.subscribeToEvents((event) => { - notifySubscribers(event); - if (event.kind !== "notification" || event.method !== "turn/completed") { - return; + const publishRuntimeEvent = (event: ProviderRuntimeEvent): Effect.Effect => + Effect.sync(() => { + canonicalEventLogger?.write({ + observedAt: new Date().toISOString(), + event, + }); + }).pipe( + Effect.flatMap(() => PubSub.publish(runtimeEventPubSub, event)), + Effect.asVoid, + ); + + const providers = yield* registry.listProviders(); + const adapters = yield* Effect.forEach(providers, (provider) => registry.getByProvider(provider)); + + const onTurnCompleted = (event: ProviderRuntimeTurnCompletedEvent): Effect.Effect => + checkpointService + .captureCurrentTurn({ + providerSessionId: event.sessionId, + ...(event.turnId !== undefined ? { turnId: event.turnId } : {}), + ...(event.status !== undefined ? { status: event.status } : {}), + }) + .pipe( + Effect.flatMap(() => checkpointService.listCheckpoints({ sessionId: event.sessionId })), + Effect.flatMap((result) => { + const currentCheckpoint = + result.checkpoints.find((checkpoint) => checkpoint.isCurrent) ?? + result.checkpoints[result.checkpoints.length - 1]; + if (!currentCheckpoint) { + return Effect.void; + } + + return publishRuntimeEvent( + toCheckpointCapturedEvent({ + event, + threadId: result.threadId, + turnCount: currentCheckpoint.turnCount, + }), + ); + }), + Effect.catch((error) => publishRuntimeEvent(toCheckpointCaptureErrorEvent(event, error))), + ); + + const processRuntimeEvent = (event: ProviderRuntimeEvent): Effect.Effect => + publishRuntimeEvent(event).pipe( + Effect.flatMap(() => { + if (event.type !== "turn.completed") { + return Effect.void; } - - const status = readTurnCompletionStatus(event); - const captureInput = { - providerSessionId: event.sessionId, - ...(event.turnId !== undefined ? { turnId: event.turnId } : {}), - ...(status !== undefined ? { status } : {}), - }; - - Effect.runFork( - checkpointService.captureCurrentTurn(captureInput).pipe( - Effect.catch((error) => - Effect.sync(() => { - notifySubscribers(toCheckpointCaptureErrorEvent(event, error)); - }), - ), - ), - ); + return onTurnCompleted(event); }), - ), - (unsubscribeAll) => - Effect.forEach(unsubscribeAll, (unsubscribe) => - Effect.sync(() => { - unsubscribe(); - }), - ).pipe(Effect.asVoid), - ); + ); + + const worker = Effect.forever(Queue.take(runtimeEventQueue).pipe(Effect.flatMap(processRuntimeEvent))); + yield* Effect.forkScoped(worker); + + yield* Effect.forEach(adapters, (adapter) => + Stream.runForEach(adapter.streamEvents, (event) => + Queue.offer(runtimeEventQueue, event).pipe(Effect.asVoid), + ).pipe(Effect.forkScoped), + ).pipe(Effect.asVoid); const adapterForSession = (sessionId: string) => directory @@ -252,6 +281,7 @@ const makeProviderService = Effect.gen(function* () { const input = parsed.data; const adapter = yield* adapterForSession(input.sessionId); yield* adapter.stopSession(input.sessionId); + yield* checkpointService.releaseSession({ providerSessionId: input.sessionId }); yield* directory.remove(input.sessionId); }); @@ -316,34 +346,34 @@ const makeProviderService = Effect.gen(function* () { const stopAll: ProviderServiceShape["stopAll"] = () => Effect.gen(function* () { - yield* Effect.forEach(adapters, (adapter) => adapter.stopAll()).pipe(Effect.asVoid); const sessionIds = yield* directory.listSessionIds(); - yield* Effect.forEach(sessionIds, (sessionId) => directory.remove(sessionId)).pipe( - Effect.asVoid, - ); + yield* Effect.forEach(adapters, (adapter) => adapter.stopAll()).pipe(Effect.asVoid); + yield* Effect.forEach( + sessionIds, + (sessionId) => + checkpointService.releaseSession({ providerSessionId: sessionId }).pipe( + Effect.flatMap(() => directory.remove(sessionId)), + ), + ).pipe(Effect.asVoid); }); - const subscribeToEvents: ProviderServiceShape["subscribeToEvents"] = (callback) => - Effect.sync(() => { - eventSubscribers.add(callback); - return () => { - eventSubscribers.delete(callback); - }; - }); - - return { - startSession, - sendTurn, - interruptTurn, + return { + startSession, + sendTurn, + interruptTurn, respondToRequest, stopSession, listSessions, listCheckpoints, getCheckpointDiff, revertToCheckpoint, - stopAll, - subscribeToEvents, - } satisfies ProviderServiceShape; -}); + stopAll, + streamEvents: Stream.fromPubSub(runtimeEventPubSub), + } satisfies ProviderServiceShape; + }); -export const ProviderServiceLive = Layer.effect(ProviderService, makeProviderService); +export const ProviderServiceLive = Layer.effect(ProviderService, makeProviderService()); + +export function makeProviderServiceLive(options?: ProviderServiceLiveOptions) { + return Layer.effect(ProviderService, makeProviderService(options)); +} diff --git a/apps/server/src/provider/Services/ProviderAdapter.ts b/apps/server/src/provider/Services/ProviderAdapter.ts index 21b6d8c3dde2456e0ca6ad6dfb3830a0dcd615bb..bff88ca30f350583ee7788ff1df637af77f84580 100644 --- a/apps/server/src/provider/Services/ProviderAdapter.ts +++ b/apps/server/src/provider/Services/ProviderAdapter.ts @@ -9,14 +9,15 @@ */ import type { ProviderApprovalDecision, - ProviderEvent, ProviderKind, + ProviderRuntimeEvent, ProviderSendTurnInput, ProviderSession, ProviderSessionStartInput, ProviderTurnStartResult, } from "@t3tools/contracts"; import type { Effect } from "effect"; +import type { Stream } from "effect"; export interface ProviderThreadTurnSnapshot { readonly id: string; @@ -96,9 +97,7 @@ export interface ProviderAdapterShape { readonly stopAll: () => Effect.Effect; /** - * Subscribe to provider events emitted by this adapter. + * Canonical runtime event stream emitted by this adapter. */ - readonly subscribeToEvents: ( - callback: (event: ProviderEvent) => void, - ) => Effect.Effect<() => void, TError>; + readonly streamEvents: Stream.Stream; } diff --git a/apps/server/src/provider/Services/ProviderService.ts b/apps/server/src/provider/Services/ProviderService.ts index 6d546bee6e516eb517170c3f26dd1ad506966880..3d5798b93bf2de3eb8c36c1646edf9f9f8219ed0 100644 --- a/apps/server/src/provider/Services/ProviderService.ts +++ b/apps/server/src/provider/Services/ProviderService.ts @@ -12,7 +12,6 @@ * @module ProviderService */ import type { - ProviderEvent, ProviderGetCheckpointDiffInput, ProviderGetCheckpointDiffResult, ProviderInterruptTurnInput, @@ -26,9 +25,10 @@ import type { ProviderSessionStartInput, ProviderStopSessionInput, ProviderTurnStartResult, + ProviderRuntimeEvent, } from "@t3tools/contracts"; import { ServiceMap } from "effect"; -import type { Effect } from "effect"; +import type { Effect, Stream } from "effect"; import type { ProviderServiceError } from "../Errors.ts"; @@ -100,13 +100,11 @@ export interface ProviderServiceShape { readonly stopAll: () => Effect.Effect; /** - * Subscribe to provider event stream. + * Canonical provider runtime event stream. * * Fan-out is owned by ProviderService (not by a standalone event-bus service). */ - readonly subscribeToEvents: ( - callback: (event: ProviderEvent) => void, - ) => Effect.Effect<() => void, ProviderServiceError>; + readonly streamEvents: Stream.Stream; } /** diff --git a/apps/server/src/provider/contracts.test.ts b/apps/server/src/provider/contracts.test.ts deleted file mode 100644 index 7f216b1c08c9c89b9d2dcbfac0368ab506761ff5..0000000000000000000000000000000000000000 --- a/apps/server/src/provider/contracts.test.ts +++ /dev/null @@ -1,245 +0,0 @@ -import { describe, expect, it } from "vitest"; - -import { Effect, Layer, ManagedRuntime, Option } from "effect"; - -import { - ProviderAdapterValidationError, - ProviderFilesystemError, - ProviderSessionNotFoundError, - ProviderUnsupportedError, - type ProviderAdapterError, - type ProviderServiceError, -} from "./Errors.ts"; -import { - CheckpointGitCommandError, - CheckpointServiceValidationError, - type CheckpointServiceError, - type CheckpointStoreError, -} from "../checkpointing/Errors.ts"; -import { - CheckpointRepositoryValidationError, - type CheckpointRepositoryError, -} from "../persistence/Errors.ts"; -import { CheckpointService } from "../checkpointing/Services/CheckpointService.ts"; -import { CheckpointStore } from "../checkpointing/Services/CheckpointStore.ts"; -import { CheckpointRepository } from "../persistence/Services/Checkpoints.ts"; -import { CodexAdapter } from "./Services/CodexAdapter.ts"; -import { ProviderAdapterRegistry } from "./Services/ProviderAdapterRegistry.ts"; -import { ProviderService } from "./Services/ProviderService.ts"; -import { ProviderSessionDirectory } from "./Services/ProviderSessionDirectory.ts"; - -describe("provider/checkpoint service contracts", () => { - it("resolves all service tags from ServiceMap", async () => { - const providerLive: typeof ProviderService.Service = { - startSession: () => - Effect.fail(new ProviderUnsupportedError({ provider: "x" })), - sendTurn: () => - Effect.fail(new ProviderUnsupportedError({ provider: "x" })), - interruptTurn: () => - Effect.fail(new ProviderUnsupportedError({ provider: "x" })), - respondToRequest: () => - Effect.fail(new ProviderUnsupportedError({ provider: "x" })), - stopSession: () => Effect.void, - listSessions: () => Effect.succeed([]), - listCheckpoints: () => - Effect.fail(new ProviderUnsupportedError({ provider: "x" })), - getCheckpointDiff: () => - Effect.fail(new ProviderUnsupportedError({ provider: "x" })), - revertToCheckpoint: () => - Effect.fail(new ProviderUnsupportedError({ provider: "x" })), - stopAll: () => Effect.void, - subscribeToEvents: () => Effect.succeed(() => undefined), - }; - - const codexLive: typeof CodexAdapter.Service = { - provider: "codex", - startSession: () => - Effect.fail( - new ProviderAdapterValidationError({ - provider: "codex", - operation: "startSession", - issue: "not implemented", - }), - ), - sendTurn: () => - Effect.fail( - new ProviderAdapterValidationError({ - provider: "codex", - operation: "sendTurn", - issue: "not implemented", - }), - ), - interruptTurn: () => - Effect.fail( - new ProviderAdapterValidationError({ - provider: "codex", - operation: "interruptTurn", - issue: "not implemented", - }), - ), - readThread: () => - Effect.fail( - new ProviderAdapterValidationError({ - provider: "codex", - operation: "readThread", - issue: "not implemented", - }), - ), - rollbackThread: () => - Effect.fail( - new ProviderAdapterValidationError({ - provider: "codex", - operation: "rollbackThread", - issue: "not implemented", - }), - ), - respondToRequest: () => - Effect.fail( - new ProviderAdapterValidationError({ - provider: "codex", - operation: "respondToRequest", - issue: "not implemented", - }), - ), - stopSession: () => Effect.void, - listSessions: () => Effect.succeed([]), - hasSession: () => Effect.succeed(false), - stopAll: () => Effect.void, - subscribeToEvents: () => Effect.succeed(() => undefined), - }; - - const checkpointStoreLive: typeof CheckpointStore.Service = { - isGitRepository: () => Effect.succeed(false), - captureCheckpoint: () => Effect.void, - hasCheckpointRef: () => Effect.succeed(false), - ensureRootCheckpoint: () => Effect.succeed(false), - restoreCheckpoint: () => Effect.succeed(false), - diffCheckpoints: () => - Effect.fail( - new CheckpointGitCommandError({ - operation: "diffCheckpoints", - command: "git diff", - cwd: "/tmp", - detail: "not implemented", - }), - ), - deleteCheckpointRefs: () => Effect.void, - }; - - const checkpointRepositoryLive: typeof CheckpointRepository.Service = { - upsertCheckpoint: () => Effect.void, - listCheckpoints: () => Effect.succeed([]), - getCheckpoint: () => Effect.succeed(Option.none()), - deleteAfterTurn: () => Effect.void, - deleteAllForSession: () => Effect.void, - }; - - const checkpointServiceLive: typeof CheckpointService.Service = { - initializeForSession: () => Effect.void, - captureCurrentTurn: () => Effect.void, - listCheckpoints: () => - Effect.fail( - new CheckpointServiceValidationError({ - operation: "listCheckpoints", - issue: "not implemented", - }), - ), - getCheckpointDiff: () => - Effect.fail( - new CheckpointServiceValidationError({ - operation: "getCheckpointDiff", - issue: "not implemented", - }), - ), - revertToCheckpoint: () => - Effect.fail( - new CheckpointServiceValidationError({ - operation: "revertToCheckpoint", - issue: "not implemented", - }), - ), - }; - - const providerAdapterRegistryLive: typeof ProviderAdapterRegistry.Service = { - getByProvider: (provider) => - provider === "codex" - ? Effect.succeed(codexLive) - : Effect.fail(new ProviderUnsupportedError({ provider })), - listProviders: () => Effect.succeed(["codex"]), - }; - - const providerSessionDirectoryLive: typeof ProviderSessionDirectory.Service = { - upsert: () => Effect.void, - getProvider: (sessionId) => - Effect.fail(new ProviderSessionNotFoundError({ sessionId })), - getThreadId: (sessionId) => - Effect.fail(new ProviderSessionNotFoundError({ sessionId })), - remove: () => Effect.void, - listSessionIds: () => Effect.succeed([]), - }; - - const runtime = ManagedRuntime.make( - Layer.mergeAll( - Layer.succeed(ProviderService, providerLive), - Layer.succeed(CodexAdapter, codexLive), - Layer.succeed(CheckpointStore, checkpointStoreLive), - Layer.succeed(CheckpointRepository, checkpointRepositoryLive), - Layer.succeed(CheckpointService, checkpointServiceLive), - Layer.succeed(ProviderAdapterRegistry, providerAdapterRegistryLive), - Layer.succeed(ProviderSessionDirectory, providerSessionDirectoryLive), - ), - ); - - const [ - provider, - codex, - checkpointStore, - checkpointRepository, - checkpointService, - providerAdapterRegistry, - providerSessionDirectory, - ] = await Promise.all([ - runtime.runPromise(Effect.service(ProviderService)), - runtime.runPromise(Effect.service(CodexAdapter)), - runtime.runPromise(Effect.service(CheckpointStore)), - runtime.runPromise(Effect.service(CheckpointRepository)), - runtime.runPromise(Effect.service(CheckpointService)), - runtime.runPromise(Effect.service(ProviderAdapterRegistry)), - runtime.runPromise(Effect.service(ProviderSessionDirectory)), - ]); - - expect(provider).toBe(providerLive); - expect(codex).toBe(codexLive); - expect(checkpointStore).toBe(checkpointStoreLive); - expect(checkpointRepository).toBe(checkpointRepositoryLive); - expect(checkpointService).toBe(checkpointServiceLive); - expect(providerAdapterRegistry).toBe(providerAdapterRegistryLive); - expect(providerSessionDirectory).toBe(providerSessionDirectoryLive); - - await runtime.dispose(); - }); -}); - -describe("provider/checkpoint errors", () => { - it("preserves cause and message on tagged errors", () => { - const fsCause = new Error("filesystem failed"); - const fsError = new ProviderFilesystemError({ - sessionId: "sess-1", - detail: "capture failed", - cause: fsCause, - }); - - expect(fsError._tag).toBe("ProviderFilesystemError"); - expect(fsError.message).toContain("sess-1"); - expect(fsError.cause).toBe(fsCause); - }); - - it("accepts checkpoint repository error channel type", () => { - const error: CheckpointRepositoryError = new CheckpointRepositoryValidationError({ - operation: "upsertCheckpoint", - issue: "turnCount must be >= 0", - }); - - expect(error._tag).toBe("CheckpointRepositoryValidationError"); - }); -}); diff --git a/apps/server/src/wsServer.test.ts b/apps/server/src/wsServer.test.ts index d38d004e48bbeede1850a6f00390f174862d1a36..b00dbbe9e95d80ac73a72a93fe4d856f3dc9f6ed 100644 --- a/apps/server/src/wsServer.test.ts +++ b/apps/server/src/wsServer.test.ts @@ -628,8 +628,7 @@ describe("WebSocket Server", () => { const ws = await connectWs(port); connections.push(ws); - // Consume welcome + initial orchestration snapshot pushes - await waitForMessage(ws); + // Consume welcome push await waitForMessage(ws); const response = await sendRequest(ws, "nonexistent.method"); diff --git a/apps/server/src/wsServer.ts b/apps/server/src/wsServer.ts index 8d33223d22128f4ec36d623bf4aa58fee8670bc2..0b1dd45d9f902e041420937606e60ace1b44d278 100644 --- a/apps/server/src/wsServer.ts +++ b/apps/server/src/wsServer.ts @@ -11,7 +11,7 @@ import { ORCHESTRATION_WS_METHODS, WS_CHANNELS, WS_METHODS, - type OrchestrationReadModelPush, + type ProviderRuntimeEvent, type TerminalEvent, type WsPush, type WsRequest, @@ -19,7 +19,7 @@ import { wsRequestSchema, } from "@t3tools/contracts"; import { NodeServices } from "@effect/platform-node"; -import { Effect, Layer, ManagedRuntime } from "effect"; +import { Effect, Exit, Layer, ManagedRuntime, Queue, Scope, Stream } from "effect"; import { WebSocketServer, type WebSocket } from "ws"; import { createLogger } from "./logger"; @@ -43,14 +43,18 @@ import { makeSqlitePersistenceLive } from "./persistence/Layers/Sqlite"; import { ProjectRepository, type ProjectRepositoryShape } from "./persistence/Services/Projects"; import assert from "node:assert"; import { OrchestrationEventRepositoryLive } from "./persistence/Layers/OrchestrationEvents"; -import { CodexAdapterLive } from "./provider/Layers/CodexAdapter"; +import { makeCodexAdapterLive } from "./provider/Layers/CodexAdapter"; import { ProviderAdapterRegistryLive } from "./provider/Layers/ProviderAdapterRegistry"; -import { ProviderServiceLive } from "./provider/Layers/ProviderService"; +import { makeProviderServiceLive } from "./provider/Layers/ProviderService"; import { ProviderSessionDirectoryLive } from "./provider/Layers/ProviderSessionDirectory"; import { ProviderService, type ProviderServiceShape } from "./provider/Services/ProviderService"; import { CheckpointStoreLive } from "./checkpointing/Layers/CheckpointStore"; import { CheckpointServiceLive } from "./checkpointing/Layers/CheckpointService"; import { CheckpointRepositoryLive } from "./persistence/Layers/Checkpoints"; +import { + makeEventNdjsonLogger, + type EventNdjsonLogger, +} from "./provider/Layers/EventNdjsonLogger"; import * as SqlClient from "effect/unstable/sql/SqlClient"; import * as SqlError from "effect/unstable/sql/SqlError"; import * as Migrator from "effect/unstable/sql/Migrator"; @@ -114,17 +118,90 @@ function parseBooleanEnv(value: string | undefined): boolean | undefined { return undefined; } -function asObject(value: unknown): Record | null { - if (!value || typeof value !== "object") return null; - return value as Record; +const providerTurnKey = (sessionId: string, turnId: string) => `${sessionId}:${turnId}`; + +type ActivityTone = "thinking" | "tool" | "info" | "error"; + +interface ThreadActivityInput { + readonly id: string; + readonly createdAt: string; + readonly label: string; + readonly detail?: string; + readonly tone: ActivityTone; + readonly turnId?: string; + readonly requestId?: string; + readonly requestKind?: "command" | "file-change"; } -function asString(value: unknown): string | null { - return typeof value === "string" ? value : null; +function truncateDetail(value: string, limit = 180): string { + return value.length > limit ? `${value.slice(0, limit - 1)}...` : value; } -function asNumber(value: unknown): number | null { - return typeof value === "number" && Number.isFinite(value) ? value : null; +function runtimeEventToActivities(event: ProviderRuntimeEvent): ReadonlyArray { + switch (event.type) { + case "approval.requested": + return [ + { + id: event.eventId, + createdAt: event.createdAt, + label: + event.requestKind === "command" + ? "Command approval requested" + : "File-change approval requested", + ...(event.detail ? { detail: truncateDetail(event.detail) } : {}), + tone: "tool", + ...(event.turnId ? { turnId: event.turnId } : {}), + requestId: event.requestId, + requestKind: event.requestKind, + }, + ]; + case "approval.resolved": + return [ + { + id: event.eventId, + createdAt: event.createdAt, + label: "Approval resolved", + tone: "info", + ...(event.turnId ? { turnId: event.turnId } : {}), + requestId: event.requestId, + ...(event.requestKind ? { requestKind: event.requestKind } : {}), + }, + ]; + case "runtime.error": + return [ + { + id: event.eventId, + createdAt: event.createdAt, + label: "Runtime error", + detail: truncateDetail(event.message), + tone: "error", + ...(event.turnId ? { turnId: event.turnId } : {}), + }, + ]; + case "checkpoint.captured": + return [ + { + id: event.eventId, + createdAt: event.createdAt, + label: "Checkpoint captured", + tone: "info", + ...(event.turnId ? { turnId: event.turnId } : {}), + }, + ]; + case "tool.completed": + return [ + { + id: event.eventId, + createdAt: event.createdAt, + label: `${event.title} complete`, + ...(event.detail ? { detail: truncateDetail(event.detail) } : {}), + tone: "tool", + ...(event.turnId ? { turnId: event.turnId } : {}), + }, + ]; + default: + return []; + } } const noop = () => {}; @@ -157,7 +234,8 @@ export function createServer(options: ServerOptions) { const logWebSocketEvents = explicitLogWsEvents ?? parseBooleanEnv(process.env.T3CODE_LOG_WS_EVENTS) ?? Boolean(devUrl); let keybindingsConfig = loadResolvedKeybindingsConfig(logger); - let unsubscribeReadModel = noop; + let orchestrationDomainEventLogger: EventNdjsonLogger | undefined; + let orchestrationCommandLogger: EventNdjsonLogger | undefined; let unsubscribeDomainEvents = noop; let unsubscribeProviderEvents = noop; @@ -197,30 +275,12 @@ export function createServer(options: ServerOptions) { >, orchestrationEngine: OrchestrationEngineShape, ) { - unsubscribeReadModel = await runtime.runPromise( - orchestrationEngine.subscribeToReadModel((snapshot) => { - const push: WsPush = { - type: "push", - channel: ORCHESTRATION_WS_CHANNELS.readModel, - data: { - sequence: snapshot.sequence, - snapshot, - } satisfies OrchestrationReadModelPush, - }; - const message = JSON.stringify(push); - let recipients = 0; - for (const client of clients) { - if (client.readyState === client.OPEN) { - client.send(message); - recipients += 1; - } - } - logOutgoingPush(push, recipients); - }), - ); - unsubscribeDomainEvents = await runtime.runPromise( orchestrationEngine.subscribeToDomainEvents((event) => { + orchestrationDomainEventLogger?.write({ + observedAt: new Date().toISOString(), + event, + }); const push: WsPush = { type: "push", channel: ORCHESTRATION_WS_CHANNELS.domainEvent, @@ -247,53 +307,96 @@ export function createServer(options: ServerOptions) { providerService: ProviderServiceShape, orchestrationEngine: OrchestrationEngineShape, ) { - unsubscribeProviderEvents = await runtime.runPromise( - providerService.subscribeToEvents((event) => { - void (async () => { - const snapshot = await runtime.runPromise(orchestrationEngine.getSnapshot()); - const thread = snapshot.threads.find( - (entry) => entry.session?.sessionId === event.sessionId, - ); - if (!thread) return; - const now = event.createdAt; - if (event.method === "turn/started" || event.method === "turn/completed") { - await runtime.runPromise( - orchestrationEngine.dispatch({ - type: "thread.session", - commandId: crypto.randomUUID(), - threadId: thread.id, - session: { - sessionId: event.sessionId, - provider: event.provider, - status: event.method === "turn/started" ? "running" : "ready", - threadId: thread.id, - activeTurnId: event.method === "turn/started" ? (event.turnId ?? null) : null, - createdAt: thread.session?.createdAt ?? now, - updatedAt: now, - lastError: null, - }, - createdAt: now, - }), - ); + const turnMessageIdsByTurnKey = new Map>(); + const latestMessageIdByTurnKey = new Map(); + + const rememberAssistantMessageId = (sessionId: string, turnId: string, messageId: string) => { + const key = providerTurnKey(sessionId, turnId); + const existingIds = turnMessageIdsByTurnKey.get(key); + if (existingIds) { + existingIds.add(messageId); + } else { + turnMessageIdsByTurnKey.set(key, new Set([messageId])); + } + latestMessageIdByTurnKey.set(key, messageId); + }; + + const getAssistantMessageIdsForTurn = (sessionId: string, turnId: string) => { + return turnMessageIdsByTurnKey.get(providerTurnKey(sessionId, turnId)) ?? new Set(); + }; + + const clearAssistantMessageIdsForTurn = (sessionId: string, turnId: string) => { + turnMessageIdsByTurnKey.delete(providerTurnKey(sessionId, turnId)); + }; + + const getLatestAssistantMessageIdForTurn = (sessionId: string, turnId: string) => { + return latestMessageIdByTurnKey.get(providerTurnKey(sessionId, turnId)); + }; + + const clearTurnStateForSession = (sessionId: string) => { + const prefix = `${sessionId}:`; + for (const key of turnMessageIdsByTurnKey.keys()) { + if (key.startsWith(prefix)) { + turnMessageIdsByTurnKey.delete(key); + } + } + for (const key of latestMessageIdByTurnKey.keys()) { + if (key.startsWith(prefix)) { + latestMessageIdByTurnKey.delete(key); + } + } + }; + + const processEvent = (event: ProviderRuntimeEvent) => + Effect.gen(function* () { + const snapshot = yield* orchestrationEngine.getSnapshot(); + const thread = snapshot.threads.find((entry) => entry.session?.sessionId === event.sessionId); + if (!thread) return; + + const now = event.createdAt; + if (event.type === "turn.started" || event.type === "turn.completed") { + yield* orchestrationEngine.dispatch({ + type: "thread.session", + commandId: crypto.randomUUID(), + threadId: thread.id, + session: { + sessionId: event.sessionId, + provider: event.provider, + status: event.type === "turn.started" ? "running" : "ready", + threadId: thread.id, + activeTurnId: event.type === "turn.started" ? event.turnId : null, + createdAt: thread.session?.createdAt ?? now, + updatedAt: now, + lastError: + event.type === "turn.completed" && event.status === "failed" + ? (event.errorMessage ?? null) + : null, + }, + createdAt: now, + }); + } + + if (event.type === "message.delta" && event.delta.length > 0) { + const assistantMessageId = `assistant:${event.itemId ?? event.turnId ?? event.sessionId}`; + if (event.turnId) { + rememberAssistantMessageId(event.sessionId, event.turnId, assistantMessageId); } - if (event.textDelta && event.textDelta.length > 0) { - const assistantMessageId = `assistant:${event.turnId ?? event.itemId ?? event.sessionId}`; - await runtime.runPromise( - orchestrationEngine.dispatch({ - type: "message.send", - commandId: crypto.randomUUID(), - threadId: thread.id, - messageId: assistantMessageId, - role: "assistant", - text: event.textDelta, - streaming: true, - createdAt: now, - }), - ); - } - if (event.method === "turn/completed") { - const assistantMessageId = `assistant:${event.turnId ?? event.itemId ?? event.sessionId}`; - await runtime.runPromise( + yield* orchestrationEngine.dispatch({ + type: "message.send", + commandId: crypto.randomUUID(), + threadId: thread.id, + messageId: assistantMessageId, + role: "assistant", + text: event.delta, + streaming: true, + createdAt: now, + }); + } + + if (event.type === "turn.completed") { + if (event.turnId) { + const assistantMessageIds = getAssistantMessageIdsForTurn(event.sessionId, event.turnId); + yield* Effect.forEach(assistantMessageIds, (assistantMessageId) => orchestrationEngine.dispatch({ type: "message.send", commandId: crypto.randomUUID(), @@ -304,55 +407,85 @@ export function createServer(options: ServerOptions) { streaming: false, createdAt: now, }), - ); + ).pipe(Effect.asVoid); + clearAssistantMessageIdsForTurn(event.sessionId, event.turnId); } - if (event.method === "checkpoint/captured") { - const payload = asObject(event.payload); - const turnId = event.turnId ?? asString(payload?.turnId); - if (turnId) { - await runtime.runPromise( - orchestrationEngine.dispatch({ - type: "thread.turnDiff.complete", - commandId: crypto.randomUUID(), - threadId: thread.id, - turnId, - completedAt: now, - ...(asString(payload?.status) !== null - ? { status: asString(payload?.status) ?? undefined } - : {}), - files: [], - assistantMessageId: `assistant:${turnId}`, - ...(asNumber(payload?.turnCount) !== null - ? { checkpointTurnCount: asNumber(payload?.turnCount) ?? undefined } - : {}), - createdAt: now, - }), - ); - } - } - if (event.kind === "error" && event.message) { - await runtime.runPromise( - orchestrationEngine.dispatch({ - type: "thread.session", - commandId: crypto.randomUUID(), - threadId: thread.id, - session: { - sessionId: event.sessionId, - provider: event.provider, - status: "error", - threadId: thread.id, - activeTurnId: event.turnId ?? null, - createdAt: thread.session?.createdAt ?? now, - updatedAt: now, - lastError: event.message, - }, - createdAt: now, - }), - ); - } - })().catch(() => undefined); - }), + } + + if (event.type === "session.exited") { + clearTurnStateForSession(event.sessionId); + } + + if (event.type === "checkpoint.captured" && event.turnId) { + const assistantMessageId = + getLatestAssistantMessageIdForTurn(event.sessionId, event.turnId) ?? + `assistant:${event.turnId}`; + yield* orchestrationEngine.dispatch({ + type: "thread.turnDiff.complete", + commandId: crypto.randomUUID(), + threadId: thread.id, + turnId: event.turnId, + completedAt: now, + ...(event.status ? { status: event.status } : {}), + files: [], + assistantMessageId, + checkpointTurnCount: event.turnCount, + createdAt: now, + }); + } + + if (event.type === "runtime.error") { + yield* orchestrationEngine.dispatch({ + type: "thread.session", + commandId: crypto.randomUUID(), + threadId: thread.id, + session: { + sessionId: event.sessionId, + provider: event.provider, + status: "error", + threadId: thread.id, + activeTurnId: event.turnId ?? null, + createdAt: thread.session?.createdAt ?? now, + updatedAt: now, + lastError: event.message, + }, + createdAt: now, + }); + } + + const activities = runtimeEventToActivities(event); + yield* Effect.forEach(activities, (activity) => + orchestrationEngine.dispatch({ + type: "thread.activity.append", + commandId: crypto.randomUUID(), + threadId: thread.id, + activity, + createdAt: activity.createdAt, + }), + ).pipe(Effect.asVoid); + }); + + const subscriptionScope = await runtime.runPromise(Scope.make("sequential")); + await runtime.runPromise( + Effect.gen(function* () { + const providerEventQueue = yield* Queue.unbounded(); + + yield* Effect.addFinalizer(() => Queue.shutdown(providerEventQueue).pipe(Effect.asVoid)); + + yield* Effect.forkScoped( + Effect.forever(Queue.take(providerEventQueue).pipe(Effect.flatMap(processEvent))), + ); + yield* Effect.forkScoped( + Stream.runForEach(providerService.streamEvents, (event) => + Queue.offer(providerEventQueue, event).pipe(Effect.asVoid), + ), + ); + }).pipe(Scope.provide(subscriptionScope)), ); + + unsubscribeProviderEvents = () => { + runtime.runFork(Scope.close(subscriptionScope, Exit.void)); + }; } const onTerminalEvent = (event: TerminalEvent) => { @@ -456,8 +589,6 @@ export function createServer(options: ServerOptions) { wss.on("connection", (ws) => { clients.add(ws); - const orchestrationEngine = getOrchestrationEngine(); - const runtime = getEffectRuntime(); // Send welcome message with project info const segments = cwd.split(/[/\\]/).filter(Boolean); @@ -471,25 +602,6 @@ export function createServer(options: ServerOptions) { logOutgoingPush(welcome, 1); ws.send(JSON.stringify(welcome)); - void runtime - .runPromise(orchestrationEngine.getSnapshot()) - .then((snapshot) => { - if (ws.readyState !== ws.OPEN) { - return; - } - const snapshotPush: WsPush = { - type: "push", - channel: ORCHESTRATION_WS_CHANNELS.readModel, - data: { - sequence: snapshot.sequence, - snapshot, - } satisfies OrchestrationReadModelPush, - }; - logOutgoingPush(snapshotPush, 1); - ws.send(JSON.stringify(snapshotPush)); - }) - .catch(() => undefined); - ws.on("message", (raw) => { void handleMessage(ws, raw); }); @@ -745,6 +857,11 @@ export function createServer(options: ServerOptions) { return effectRuntime.runPromise(orchestrationEngine.getSnapshot()); case ORCHESTRATION_WS_METHODS.dispatchCommand: + orchestrationCommandLogger?.write({ + observedAt: new Date().toISOString(), + requestId: request.id, + command: request.params, + }); return effectRuntime.runPromise(orchestrationEngine.dispatchUnknown(request.params)); case ORCHESTRATION_WS_METHODS.replayEvents: @@ -769,12 +886,23 @@ export function createServer(options: ServerOptions) { async function createEffectRuntime() { const dbPath = path.join(resolvedStateDir, "orchestration.sqlite"); + const providerLogsDir = path.join(resolvedStateDir, "logs", "providers"); + const orchestrationLogsDir = path.join(resolvedStateDir, "logs", "orchestration"); + orchestrationDomainEventLogger = makeEventNdjsonLogger( + path.join(orchestrationLogsDir, "orchestration-domain.ndjson"), + ); + orchestrationCommandLogger = makeEventNdjsonLogger( + path.join(orchestrationLogsDir, "orchestration-command.ndjson"), + ); const persistenceLayer = customPersistenceLayer ?? makeSqlitePersistenceLive(dbPath); const orchestrationLayer = Layer.provide( OrchestrationEngineLive, OrchestrationEventRepositoryLive, ); - const adapterRegistryLayer = ProviderAdapterRegistryLive.pipe(Layer.provide(CodexAdapterLive)); + const codexAdapterLayer = makeCodexAdapterLive({ + nativeEventLogPath: path.join(providerLogsDir, "provider-native.ndjson"), + }); + const adapterRegistryLayer = ProviderAdapterRegistryLive.pipe(Layer.provide(codexAdapterLayer)); const checkpointStoreLayer = CheckpointStoreLive.pipe(Layer.provide(NodeServices.layer)); const checkpointDependenciesLayer = Layer.mergeAll( checkpointStoreLayer, @@ -785,7 +913,9 @@ export function createServer(options: ServerOptions) { const checkpointServiceLayer = CheckpointServiceLive.pipe( Layer.provideMerge(checkpointDependenciesLayer), ); - const providerLayer = ProviderServiceLive.pipe(Layer.provideMerge(checkpointServiceLayer)); + const providerLayer = makeProviderServiceLive({ + canonicalEventLogPath: path.join(providerLogsDir, "provider-canonical.ndjson"), + }).pipe(Layer.provideMerge(checkpointServiceLayer)); const layer = Layer.mergeAll(orchestrationLayer, ProjectRepositoryLive, providerLayer).pipe( Layer.provide(persistenceLayer), @@ -814,12 +944,14 @@ export function createServer(options: ServerOptions) { } async function disposeEffectRuntime() { - unsubscribeReadModel(); unsubscribeDomainEvents(); unsubscribeProviderEvents(); - unsubscribeReadModel = noop; unsubscribeDomainEvents = noop; unsubscribeProviderEvents = noop; + orchestrationDomainEventLogger?.close(); + orchestrationDomainEventLogger = undefined; + orchestrationCommandLogger?.close(); + orchestrationCommandLogger = undefined; const runtime = effectRuntime; const liveProviderService = providerService; diff --git a/apps/server/tsconfig.json b/apps/server/tsconfig.json index 776a47cce04100d562403014f72b25d17ca41c46..122fd028672c33e174ef76ef90764192910e1617 100644 --- a/apps/server/tsconfig.json +++ b/apps/server/tsconfig.json @@ -7,5 +7,5 @@ "noEmit": true, "allowImportingTsExtensions": true }, - "include": ["src", "tsdown.config.ts"] + "include": ["src", "integration", "tsdown.config.ts", "scripts"] } diff --git a/apps/web/src/components/ChatView.tsx b/apps/web/src/components/ChatView.tsx index 0a1423863ea3372ebfefe71df270ab353b6e0c69..8c68aa3a82108f808f1e0ca634194b2ace2b174f 100644 --- a/apps/web/src/components/ChatView.tsx +++ b/apps/web/src/components/ChatView.tsx @@ -432,20 +432,35 @@ export default function ChatView({ threadId }: ChatViewProps) { ); const phase = derivePhase(activeThread?.session ?? null); const isWorking = phase === "running" || isSending || isConnecting || isRevertingCheckpoint; + const activeTurnId = activeThread?.session?.activeTurnId ?? undefined; const nowIso = new Date(nowTick).toISOString(); - const workLogEntries = useMemo(() => deriveWorkLogEntries([], undefined), []); + const threadActivities = activeThread?.activities ?? []; + const workLogEntries = useMemo( + () => deriveWorkLogEntries(threadActivities, undefined), + [threadActivities], + ); const latestTurnWorkEntries = useMemo( - () => deriveWorkLogEntries([], activeThread?.latestTurnId), - [activeThread?.latestTurnId], + () => deriveWorkLogEntries(threadActivities, activeTurnId ?? activeThread?.latestTurnId), + [activeThread?.latestTurnId, activeTurnId, threadActivities], + ); + const pendingApprovals = useMemo( + () => derivePendingApprovals(threadActivities), + [threadActivities], ); - const pendingApprovals = useMemo(() => derivePendingApprovals([]), []); - const assistantCompletionByItemId = useMemo(() => new Map(), []); const timelineEntries = useMemo( () => deriveTimelineEntries(activeThread?.messages ?? [], workLogEntries), [activeThread?.messages, workLogEntries], ); const { turnDiffSummaries, inferredCheckpointTurnCountByTurnId } = useTurnDiffSummaries(activeThread); + const assistantCompletionByItemId = useMemo(() => { + const completionByMessageId = new Map(); + for (const summary of turnDiffSummaries) { + if (!summary.assistantMessageId) continue; + completionByMessageId.set(summary.assistantMessageId, summary.completedAt); + } + return completionByMessageId; + }, [turnDiffSummaries]); const turnDiffSummaryByAssistantMessageId = useMemo(() => { const byMessageId = new Map(); for (const summary of turnDiffSummaries) { diff --git a/apps/web/src/routes/__root.tsx b/apps/web/src/routes/__root.tsx index 8a6ee0752bc72e1099a2a202b38a60d0396a5f7f..041c57fdfecaed6edafca3963b269bd1e7f66d32 100644 --- a/apps/web/src/routes/__root.tsx +++ b/apps/web/src/routes/__root.tsx @@ -126,17 +126,59 @@ function EventRouter() { useEffect(() => { if (!api) return; - void Promise.all([api.projects.list(), api.orchestration.getSnapshot()]) - .then(([projects, snapshot]) => { + let disposed = false; + let latestSequence = 0; + let syncing = false; + let pending = false; + + const flushSnapshotSync = async (): Promise => { + const snapshot = await api.orchestration.getSnapshot(); + if (disposed) return; + latestSequence = Math.max(latestSequence, snapshot.sequence); + dispatch({ type: "SYNC_SERVER_READ_MODEL", readModel: snapshot }); + if (pending) { + pending = false; + await flushSnapshotSync(); + } + }; + + const syncSnapshot = async () => { + if (syncing) { + pending = true; + return; + } + syncing = true; + try { + pending = false; + await flushSnapshotSync(); + } catch { + // Keep prior state and wait for next domain event to trigger a resync. + } finally { + syncing = false; + } + }; + + void Promise.all([api.projects.list(), syncSnapshot()]) + .then(([projects]) => { + if (disposed) return; dispatch({ type: "SYNC_PROJECTS", projects }); - dispatch({ type: "SYNC_SERVER_READ_MODEL", readModel: snapshot }); }) .catch(() => undefined); - const unsubReadModel = api.orchestration.onReadModel((snapshot) => { - dispatch({ type: "SYNC_SERVER_READ_MODEL", readModel: snapshot }); + + const unsubDomainEvent = api.orchestration.onDomainEvent((event) => { + if (event.sequence <= latestSequence) { + return; + } + latestSequence = event.sequence; + void syncSnapshot(); + }); + const unsubWelcome = onServerWelcome(() => { + void syncSnapshot(); }); return () => { - unsubReadModel(); + disposed = true; + unsubDomainEvent(); + unsubWelcome(); }; }, [api, dispatch]); diff --git a/apps/web/src/session-logic.test.ts b/apps/web/src/session-logic.test.ts new file mode 100644 index 0000000000000000000000000000000000000000..c154a76644c9a108598a01810fd53f5bdda010e8 --- /dev/null +++ b/apps/web/src/session-logic.test.ts @@ -0,0 +1,90 @@ +import type { OrchestrationThreadActivity } from "@t3tools/contracts"; +import { describe, expect, it } from "vitest"; + +import { derivePendingApprovals, deriveWorkLogEntries } from "./session-logic"; + +function makeActivity( + overrides: Partial, +): OrchestrationThreadActivity { + return { + id: overrides.id ?? crypto.randomUUID(), + createdAt: overrides.createdAt ?? "2026-02-23T00:00:00.000Z", + label: overrides.label ?? "Tool call", + tone: overrides.tone ?? "tool", + ...(overrides.detail !== undefined ? { detail: overrides.detail } : {}), + ...(overrides.turnId !== undefined ? { turnId: overrides.turnId } : {}), + ...(overrides.requestId !== undefined ? { requestId: overrides.requestId } : {}), + ...(overrides.requestKind !== undefined ? { requestKind: overrides.requestKind } : {}), + }; +} + +describe("derivePendingApprovals", () => { + it("tracks open approvals and removes resolved ones", () => { + const activities: OrchestrationThreadActivity[] = [ + makeActivity({ + id: "approval-open", + createdAt: "2026-02-23T00:00:01.000Z", + label: "Command approval requested", + detail: "bun run lint", + tone: "tool", + requestId: "req-1", + requestKind: "command", + }), + makeActivity({ + id: "approval-close", + createdAt: "2026-02-23T00:00:02.000Z", + label: "Approval resolved", + tone: "info", + requestId: "req-2", + }), + makeActivity({ + id: "approval-closed-request", + createdAt: "2026-02-23T00:00:01.500Z", + label: "File-change approval requested", + tone: "tool", + requestId: "req-2", + requestKind: "file-change", + }), + ]; + + expect(derivePendingApprovals(activities)).toEqual([ + { + requestId: "req-1", + requestKind: "command", + createdAt: "2026-02-23T00:00:01.000Z", + detail: "bun run lint", + }, + ]); + }); +}); + +describe("deriveWorkLogEntries", () => { + it("returns entries in chronological order", () => { + const activities: OrchestrationThreadActivity[] = [ + makeActivity({ + id: "tool-complete", + createdAt: "2026-02-23T00:00:03.000Z", + label: "Tool call complete", + }), + makeActivity({ + id: "tool-start", + createdAt: "2026-02-23T00:00:02.000Z", + label: "Tool call", + }), + ]; + + const entries = deriveWorkLogEntries(activities, undefined); + expect(entries.map((entry) => entry.id)).toEqual(["tool-start", "tool-complete"]); + }); + + it("filters by turn id when provided", () => { + const activities: OrchestrationThreadActivity[] = [ + makeActivity({ id: "turn-1", turnId: "turn-1", label: "Tool call" }), + makeActivity({ id: "turn-2", turnId: "turn-2", label: "Tool call complete" }), + makeActivity({ id: "no-turn", label: "Checkpoint captured", tone: "info" }), + ]; + + const entries = deriveWorkLogEntries(activities, "turn-2"); + expect(entries.map((entry) => entry.id)).toEqual(["turn-2"]); + }); +}); diff --git a/apps/web/src/session-logic.ts b/apps/web/src/session-logic.ts index f8a5592be86ea070951243a3bf10fc69c4f4895b..6ccfb21223ccfd12b7d275d5e03151f3efe6a36f 100644 --- a/apps/web/src/session-logic.ts +++ b/apps/web/src/session-logic.ts @@ -1,4 +1,8 @@ -import type { ProviderKind, ProviderSession } from "@t3tools/contracts"; +import type { + OrchestrationThreadActivity, + ProviderKind, + ProviderSession, +} from "@t3tools/contracts"; import type { ChatMessage, SessionPhase, TurnDiffFileChange, TurnDiffSummary } from "./types"; @@ -70,15 +74,56 @@ export function formatElapsed(startIso: string, endIso: string | undefined): str return formatDuration(endedAt - startedAt); } -export function derivePendingApprovals(_events: unknown[]): PendingApproval[] { - return []; +export function derivePendingApprovals( + activities: ReadonlyArray, +): PendingApproval[] { + const openByRequestId = new Map(); + const ordered = [...activities].toSorted((left, right) => + left.createdAt.localeCompare(right.createdAt), + ); + + for (const activity of ordered) { + if (activity.requestId && activity.requestKind) { + openByRequestId.set(activity.requestId, { + requestId: activity.requestId, + requestKind: activity.requestKind, + createdAt: activity.createdAt, + ...(activity.detail ? { detail: activity.detail } : {}), + }); + continue; + } + + if (activity.requestId && activity.label === "Approval resolved") { + openByRequestId.delete(activity.requestId); + } + } + + return [...openByRequestId.values()].toSorted((left, right) => + left.createdAt.localeCompare(right.createdAt), + ); } export function deriveWorkLogEntries( - _events: unknown[], - _latestTurnId: string | undefined, + activities: ReadonlyArray, + latestTurnId: string | undefined, ): WorkLogEntry[] { - return []; + const ordered = [...activities].toSorted((left, right) => + left.createdAt.localeCompare(right.createdAt), + ); + return ordered + .filter((activity) => (latestTurnId ? activity.turnId === latestTurnId : true)) + .map((activity) => { + const entry: WorkLogEntry = { + id: activity.id, + createdAt: activity.createdAt, + label: activity.label, + tone: activity.tone, + }; + if (activity.detail) { + entry.detail = activity.detail; + } + return entry; + }); } export function deriveTimelineEntries( @@ -97,7 +142,7 @@ export function deriveTimelineEntries( createdAt: entry.createdAt, entry, })); - return [...messageRows, ...workRows].sort((a, b) => a.createdAt.localeCompare(b.createdAt)); + return [...messageRows, ...workRows].toSorted((a, b) => a.createdAt.localeCompare(b.createdAt)); } export function deriveTurnDiffFilesFromUnifiedDiff(diff: string): TurnDiffFileChange[] { @@ -116,7 +161,7 @@ export function deriveTurnDiffFilesFromUnifiedDiff(diff: string): TurnDiffFileCh export function inferCheckpointTurnCountByTurnId( summaries: TurnDiffSummary[], ): Record { - const sorted = [...summaries].sort((a, b) => a.completedAt.localeCompare(b.completedAt)); + const sorted = [...summaries].toSorted((a, b) => a.completedAt.localeCompare(b.completedAt)); const result: Record = {}; for (let index = 0; index < sorted.length; index += 1) { const summary = sorted[index]; @@ -126,7 +171,7 @@ export function inferCheckpointTurnCountByTurnId( return result; } -export function deriveTurnDiffSummaries(_events: unknown[]): TurnDiffSummary[] { +export function deriveTurnDiffSummaries(_events: ReadonlyArray): TurnDiffSummary[] { return []; } diff --git a/apps/web/src/store.ts b/apps/web/src/store.ts index 46de08875d1f5abc13af10fe2a9875b0c436f484..faf2cb3c9b519ccacef525299ed8b942afae7cd6 100644 --- a/apps/web/src/store.ts +++ b/apps/web/src/store.ts @@ -398,6 +398,7 @@ export function reducer(state: AppState, action: Action): AppState { ...summary, files: summary.files.map((file) => ({ ...file })), })), + activities: thread.activities.map((activity) => ({ ...activity })), }); }); return { diff --git a/apps/web/src/threadFactory.ts b/apps/web/src/threadFactory.ts index 01a638140b98c500bf0934adbf553f2f27240a24..d5a2b0e5df155f7ef715a0f168968b1e6a98bb1c 100644 --- a/apps/web/src/threadFactory.ts +++ b/apps/web/src/threadFactory.ts @@ -33,6 +33,7 @@ export function createThread(projectId: string, options: CreateThreadOptions = { session: null, messages: [], turnDiffSummaries: [], + activities: [], error: null, createdAt, branch: options.branch ?? null, diff --git a/apps/web/src/types.ts b/apps/web/src/types.ts index 88188e426da5f24f2bcd09b8ef641e28ca3cc31a..b7fb449e6759d9e8617244aada834a92d6c27870 100644 --- a/apps/web/src/types.ts +++ b/apps/web/src/types.ts @@ -1,4 +1,5 @@ import type { + OrchestrationThreadActivity, ProjectScript as ContractProjectScript, ProviderSession, } from "@t3tools/contracts"; @@ -86,4 +87,5 @@ export interface Thread { branch: string | null; worktreePath: string | null; turnDiffSummaries: TurnDiffSummary[]; + activities: OrchestrationThreadActivity[]; } diff --git a/apps/web/src/worktreeCleanup.test.ts b/apps/web/src/worktreeCleanup.test.ts index 423b86d3fc7d1ba79f50e04ab538f668bfe10fc2..a00ab258fec9e95abee762a9dde5b9677cb23d4b 100644 --- a/apps/web/src/worktreeCleanup.test.ts +++ b/apps/web/src/worktreeCleanup.test.ts @@ -25,6 +25,7 @@ function makeThread(overrides: Partial = {}): Thread { session: null, messages: [], turnDiffSummaries: [], + activities: [], error: null, createdAt: "2026-02-13T00:00:00.000Z", branch: null, diff --git a/apps/web/src/wsNativeApi.ts b/apps/web/src/wsNativeApi.ts index 7a52f75365bb7d0c99736212d48f1b0407b4bf9e..d40439c888b5b63876f4949bddd1682bc1ccc00a 100644 --- a/apps/web/src/wsNativeApi.ts +++ b/apps/web/src/wsNativeApi.ts @@ -4,8 +4,6 @@ import { type NativeApi, WS_CHANNELS, WS_METHODS, - type OrchestrationEvent, - type OrchestrationReadModel, type WsWelcomePayload, } from "@t3tools/contracts"; @@ -161,14 +159,6 @@ export function createWsNativeApi(): NativeApi { transport.request(ORCHESTRATION_WS_METHODS.dispatchCommand, command), replayEvents: (fromSequenceExclusive) => transport.request(ORCHESTRATION_WS_METHODS.replayEvents, { fromSequenceExclusive }), - onReadModel: (callback) => - transport.subscribe( - ORCHESTRATION_WS_CHANNELS.readModel, - (data: unknown) => { - const payload = data as { snapshot: OrchestrationReadModel }; - callback(payload.snapshot); - }, - ), onDomainEvent: (callback) => transport.subscribe( ORCHESTRATION_WS_CHANNELS.domainEvent, diff --git a/packages/contracts/src/index.ts b/packages/contracts/src/index.ts index fea1328075006b7e47884fae17cfccf0ae049a8f..252cf2e3e2a3037af6af87b502736e5ed7f3d3f9 100644 --- a/packages/contracts/src/index.ts +++ b/packages/contracts/src/index.ts @@ -3,6 +3,7 @@ export * from "./ipc"; export * from "./agent"; export * from "./terminal"; export * from "./provider"; +export * from "./providerRuntime"; export * from "./model"; export * from "./ws"; export * from "./keybindings"; diff --git a/packages/contracts/src/ipc.ts b/packages/contracts/src/ipc.ts index b3649d0907f84806f47afd05b19a7b8efcca1cf7..8f199dda1429e8b3f0072adb3afbaab6edce2210 100644 --- a/packages/contracts/src/ipc.ts +++ b/packages/contracts/src/ipc.ts @@ -146,7 +146,6 @@ export interface NativeApi { getSnapshot: () => Promise; dispatchCommand: (command: OrchestrationCommand) => Promise<{ sequence: number }>; replayEvents: (fromSequenceExclusive: number) => Promise; - onReadModel: (callback: (snapshot: OrchestrationReadModel) => void) => () => void; onDomainEvent: (callback: (event: OrchestrationEvent) => void) => () => void; }; } diff --git a/packages/contracts/src/orchestration.ts b/packages/contracts/src/orchestration.ts index ce12299721b7c6f6d872b2781c86dcf57d24ad6f..f2851d880f90d4019c6d6310cf1dec478ca7eb94 100644 --- a/packages/contracts/src/orchestration.ts +++ b/packages/contracts/src/orchestration.ts @@ -7,7 +7,6 @@ export const ORCHESTRATION_WS_METHODS = { } as const; export const ORCHESTRATION_WS_CHANNELS = { - readModel: "orchestration.readModel", domainEvent: "orchestration.domainEvent", } as const; @@ -59,6 +58,30 @@ export type OrchestrationTurnDiffSummary = Schema.Schema.Type< typeof OrchestrationTurnDiffSummarySchema >; +export const OrchestrationThreadActivityToneSchema = Schema.Literals([ + "thinking", + "tool", + "info", + "error", +]); + +export type OrchestrationThreadActivityTone = Schema.Schema.Type< + typeof OrchestrationThreadActivityToneSchema +>; + +export const OrchestrationThreadActivitySchema = Schema.Struct({ + id: IdSchema, + createdAt: IsoDateTimeSchema, + label: Schema.String, + detail: Schema.optional(Schema.String), + tone: OrchestrationThreadActivityToneSchema, + turnId: Schema.optional(Schema.String), + requestId: Schema.optional(Schema.String), + requestKind: Schema.optional(Schema.Literals(["command", "file-change"])), +}); + +export type OrchestrationThreadActivity = Schema.Schema.Type; + export const OrchestrationThreadSchema = Schema.Struct({ id: IdSchema, projectId: IdSchema, @@ -75,6 +98,7 @@ export const OrchestrationThreadSchema = Schema.Struct({ messages: Schema.Array(OrchestrationMessageSchema), session: Schema.NullOr(OrchestrationSessionSchema), turnDiffSummaries: Schema.Array(OrchestrationTurnDiffSummarySchema), + activities: Schema.Array(OrchestrationThreadActivitySchema), error: Schema.NullOr(Schema.String), }); @@ -182,6 +206,14 @@ export const RevertThreadCommandSchema = Schema.Struct({ createdAt: IsoDateTimeSchema, }); +export const AppendThreadActivityCommandSchema = Schema.Struct({ + type: Schema.Literal("thread.activity.append"), + commandId: IdSchema, + threadId: IdSchema, + activity: OrchestrationThreadActivitySchema, + createdAt: IsoDateTimeSchema, +}); + export const OrchestrationCommandSchema = Schema.Union([ CreateThreadCommandSchema, DeleteThreadCommandSchema, @@ -190,6 +222,7 @@ export const OrchestrationCommandSchema = Schema.Union([ SetThreadSessionSchema, UpsertGitReadModelCommandSchema, CompleteThreadTurnDiffCommandSchema, + AppendThreadActivityCommandSchema, RevertThreadCommandSchema,] ); @@ -207,10 +240,3 @@ export const OrchestrationEventSchema = Schema.Struct({ }); export type OrchestrationEvent = Schema.Schema.Type; - -export const OrchestrationReadModelPushSchema = Schema.Struct({ - sequence: Schema.Number, - snapshot: OrchestrationReadModelSchema, -}); - -export type OrchestrationReadModelPush = Schema.Schema.Type; diff --git a/packages/contracts/src/providerRuntime.ts b/packages/contracts/src/providerRuntime.ts new file mode 100644 index 0000000000000000000000000000000000000000..95e9b97e11de5465880011ce5df50f807d425764 --- /dev/null +++ b/packages/contracts/src/providerRuntime.ts @@ -0,0 +1,217 @@ +import { Schema } from "effect"; + +const IsoDateTimeSchema = Schema.String; +const IdSchema = Schema.String; + +export const ProviderRuntimeProviderSchema = Schema.Literals(["codex", "claudeCode"]); +export type ProviderRuntimeProvider = Schema.Schema.Type; + +export const ProviderRuntimeApprovalKindSchema = Schema.Literals(["command", "file-change"]); +export type ProviderRuntimeApprovalKind = Schema.Schema.Type< + typeof ProviderRuntimeApprovalKindSchema +>; + +export const ProviderRuntimeToolKindSchema = Schema.Literals(["command", "file-change", "other"]); +export type ProviderRuntimeToolKind = Schema.Schema.Type; + +export const ProviderRuntimeTurnStatusSchema = Schema.Literals([ + "completed", + "failed", + "cancelled", + "interrupted", +]); +export type ProviderRuntimeTurnStatus = Schema.Schema.Type; + +export const ProviderRuntimeSessionStartedEventSchema = Schema.Struct({ + type: Schema.Literal("session.started"), + eventId: IdSchema, + provider: ProviderRuntimeProviderSchema, + sessionId: IdSchema, + createdAt: IsoDateTimeSchema, + threadId: Schema.optional(Schema.String), + message: Schema.optional(Schema.String), +}); + +export const ProviderRuntimeSessionExitedEventSchema = Schema.Struct({ + type: Schema.Literal("session.exited"), + eventId: IdSchema, + provider: ProviderRuntimeProviderSchema, + sessionId: IdSchema, + createdAt: IsoDateTimeSchema, + threadId: Schema.optional(Schema.String), + message: Schema.optional(Schema.String), +}); + +export const ProviderRuntimeThreadStartedEventSchema = Schema.Struct({ + type: Schema.Literal("thread.started"), + eventId: IdSchema, + provider: ProviderRuntimeProviderSchema, + sessionId: IdSchema, + createdAt: IsoDateTimeSchema, + threadId: IdSchema, +}); + +export const ProviderRuntimeTurnStartedEventSchema = Schema.Struct({ + type: Schema.Literal("turn.started"), + eventId: IdSchema, + provider: ProviderRuntimeProviderSchema, + sessionId: IdSchema, + createdAt: IsoDateTimeSchema, + threadId: Schema.optional(Schema.String), + turnId: IdSchema, +}); + +export const ProviderRuntimeTurnCompletedEventSchema = Schema.Struct({ + type: Schema.Literal("turn.completed"), + eventId: IdSchema, + provider: ProviderRuntimeProviderSchema, + sessionId: IdSchema, + createdAt: IsoDateTimeSchema, + threadId: Schema.optional(Schema.String), + turnId: Schema.optional(Schema.String), + status: Schema.optional(ProviderRuntimeTurnStatusSchema), + errorMessage: Schema.optional(Schema.String), +}); + +export const ProviderRuntimeMessageDeltaEventSchema = Schema.Struct({ + type: Schema.Literal("message.delta"), + eventId: IdSchema, + provider: ProviderRuntimeProviderSchema, + sessionId: IdSchema, + createdAt: IsoDateTimeSchema, + threadId: Schema.optional(Schema.String), + turnId: Schema.optional(Schema.String), + itemId: Schema.optional(Schema.String), + delta: Schema.String, +}); + +export const ProviderRuntimeToolStartedEventSchema = Schema.Struct({ + type: Schema.Literal("tool.started"), + eventId: IdSchema, + provider: ProviderRuntimeProviderSchema, + sessionId: IdSchema, + createdAt: IsoDateTimeSchema, + threadId: Schema.optional(Schema.String), + turnId: Schema.optional(Schema.String), + itemId: Schema.optional(Schema.String), + toolKind: ProviderRuntimeToolKindSchema, + title: Schema.String, + detail: Schema.optional(Schema.String), +}); + +export const ProviderRuntimeToolCompletedEventSchema = Schema.Struct({ + type: Schema.Literal("tool.completed"), + eventId: IdSchema, + provider: ProviderRuntimeProviderSchema, + sessionId: IdSchema, + createdAt: IsoDateTimeSchema, + threadId: Schema.optional(Schema.String), + turnId: Schema.optional(Schema.String), + itemId: Schema.optional(Schema.String), + toolKind: ProviderRuntimeToolKindSchema, + title: Schema.String, + detail: Schema.optional(Schema.String), +}); + +export const ProviderRuntimeApprovalRequestedEventSchema = Schema.Struct({ + type: Schema.Literal("approval.requested"), + eventId: IdSchema, + provider: ProviderRuntimeProviderSchema, + sessionId: IdSchema, + createdAt: IsoDateTimeSchema, + threadId: Schema.optional(Schema.String), + turnId: Schema.optional(Schema.String), + itemId: Schema.optional(Schema.String), + requestId: IdSchema, + requestKind: ProviderRuntimeApprovalKindSchema, + detail: Schema.optional(Schema.String), +}); + +export const ProviderRuntimeApprovalResolvedEventSchema = Schema.Struct({ + type: Schema.Literal("approval.resolved"), + eventId: IdSchema, + provider: ProviderRuntimeProviderSchema, + sessionId: IdSchema, + createdAt: IsoDateTimeSchema, + threadId: Schema.optional(Schema.String), + turnId: Schema.optional(Schema.String), + itemId: Schema.optional(Schema.String), + requestId: IdSchema, + requestKind: Schema.optional(ProviderRuntimeApprovalKindSchema), + decision: Schema.optional(Schema.String), +}); + +export const ProviderRuntimeCheckpointCapturedEventSchema = Schema.Struct({ + type: Schema.Literal("checkpoint.captured"), + eventId: IdSchema, + provider: ProviderRuntimeProviderSchema, + sessionId: IdSchema, + createdAt: IsoDateTimeSchema, + threadId: IdSchema, + turnId: Schema.optional(Schema.String), + turnCount: Schema.Number, + status: Schema.optional(ProviderRuntimeTurnStatusSchema), +}); + +export const ProviderRuntimeErrorEventSchema = Schema.Struct({ + type: Schema.Literal("runtime.error"), + eventId: IdSchema, + provider: ProviderRuntimeProviderSchema, + sessionId: IdSchema, + createdAt: IsoDateTimeSchema, + threadId: Schema.optional(Schema.String), + turnId: Schema.optional(Schema.String), + itemId: Schema.optional(Schema.String), + message: Schema.String, +}); + +export const ProviderRuntimeEventSchema = Schema.Union([ + ProviderRuntimeSessionStartedEventSchema, + ProviderRuntimeSessionExitedEventSchema, + ProviderRuntimeThreadStartedEventSchema, + ProviderRuntimeTurnStartedEventSchema, + ProviderRuntimeTurnCompletedEventSchema, + ProviderRuntimeMessageDeltaEventSchema, + ProviderRuntimeToolStartedEventSchema, + ProviderRuntimeToolCompletedEventSchema, + ProviderRuntimeApprovalRequestedEventSchema, + ProviderRuntimeApprovalResolvedEventSchema, + ProviderRuntimeCheckpointCapturedEventSchema, + ProviderRuntimeErrorEventSchema, +]); + +export type ProviderRuntimeSessionStartedEvent = Schema.Schema.Type< + typeof ProviderRuntimeSessionStartedEventSchema +>; +export type ProviderRuntimeSessionExitedEvent = Schema.Schema.Type< + typeof ProviderRuntimeSessionExitedEventSchema +>; +export type ProviderRuntimeThreadStartedEvent = Schema.Schema.Type< + typeof ProviderRuntimeThreadStartedEventSchema +>; +export type ProviderRuntimeTurnStartedEvent = Schema.Schema.Type< + typeof ProviderRuntimeTurnStartedEventSchema +>; +export type ProviderRuntimeTurnCompletedEvent = Schema.Schema.Type< + typeof ProviderRuntimeTurnCompletedEventSchema +>; +export type ProviderRuntimeMessageDeltaEvent = Schema.Schema.Type< + typeof ProviderRuntimeMessageDeltaEventSchema +>; +export type ProviderRuntimeToolStartedEvent = Schema.Schema.Type< + typeof ProviderRuntimeToolStartedEventSchema +>; +export type ProviderRuntimeToolCompletedEvent = Schema.Schema.Type< + typeof ProviderRuntimeToolCompletedEventSchema +>; +export type ProviderRuntimeApprovalRequestedEvent = Schema.Schema.Type< + typeof ProviderRuntimeApprovalRequestedEventSchema +>; +export type ProviderRuntimeApprovalResolvedEvent = Schema.Schema.Type< + typeof ProviderRuntimeApprovalResolvedEventSchema +>; +export type ProviderRuntimeCheckpointCapturedEvent = Schema.Schema.Type< + typeof ProviderRuntimeCheckpointCapturedEventSchema +>; +export type ProviderRuntimeErrorEvent = Schema.Schema.Type; +export type ProviderRuntimeEvent = Schema.Schema.Type;