From 7307079d4063a5c546a4c277dcc3b6594b41a846 Mon Sep 17 00:00:00 2001 From: Julius Marminge Date: Mon, 23 Feb 2026 14:32:30 -0800 Subject: [PATCH] spec --- .plans/15-effect-server.md | 11 +- .vscode/settings.json | 3 +- AGENTS.md | 2 + SPEC.md | 569 ++++++++++++++++++ .../TestProviderAdapter.integration.ts | 24 +- apps/server/package.json | 3 + .../Layers/CheckpointService.test.ts | 164 ++++- .../checkpointing/Layers/CheckpointService.ts | 232 +++---- apps/server/src/index.ts | 2 + .../persistence/Layers/Checkpoints.test.ts | 153 +++-- .../src/persistence/Layers/Checkpoints.ts | 57 +- .../server/src/persistence/Layers/Projects.ts | 2 +- apps/server/src/persistence/Layers/Sqlite.ts | 37 +- .../persistence/Migrations/003_Checkpoints.ts | 13 +- ...lient.test.ts => NodeSqliteClient.test.ts} | 2 +- .../{SqliteClient.ts => NodeSqliteClient.ts} | 0 .../src/persistence/Services/Checkpoints.ts | 26 +- .../provider/Layers/ProviderService.test.ts | 144 ++++- .../src/provider/Layers/ProviderService.ts | 566 ++++++++++++----- .../Layers/ProviderSessionDirectory.test.ts | 47 +- .../Layers/ProviderSessionDirectory.ts | 33 - .../Services/ProviderAdapterRegistry.ts | 2 + .../src/provider/Services/ProviderService.ts | 5 +- .../Services/ProviderSessionDirectory.ts | 13 - apps/server/tsconfig.json | 7 +- bun.lock | 7 + packages/contracts/package.json | 2 + packages/contracts/tsconfig.json | 7 + 28 files changed, 1613 insertions(+), 520 deletions(-) create mode 100644 SPEC.md rename apps/server/src/persistence/{SqliteClient.test.ts => NodeSqliteClient.test.ts} (94%) rename apps/server/src/persistence/{SqliteClient.ts => NodeSqliteClient.ts} (100%) diff --git a/.plans/15-effect-server.md b/.plans/15-effect-server.md index 91cdb5978bb7eb6de72dba26c932189c58f3790e..50e869d49946f17c476cf00487ae6eb735ff9850 100644 --- a/.plans/15-effect-server.md +++ b/.plans/15-effect-server.md @@ -1 +1,10 @@ -Rewrite `createServer` and `index.ts` to be Effect native. \ No newline at end of file +Rewrite `createServer` and `index.ts` to be Effect native. + +Maybe use `effect/unstable/Socket` for the web socket server +- https://github.com/Effect-TS/effect-smol/blob/main/packages/effect/src/unstable/socket/SocketServer.ts +- https://github.com/Effect-TS/effect-smol/blob/main/packages/platform-node/test/NodeSocket.test.ts + +- Migrate remaining runtime code to Effect + - `gitManager` -> `src/git` + - `terminalManager` -> `src/terminal` (Manager + PTY) + - ... \ No newline at end of file diff --git a/.vscode/settings.json b/.vscode/settings.json index 554e703ccb97fa4acaa2a449465df2ec36de8209..752d9a90713e78fab605774a1976117166b672bd 100644 --- a/.vscode/settings.json +++ b/.vscode/settings.json @@ -4,5 +4,6 @@ "editor.codeActionsOnSave": { "source.fixAll.oxc": "always" }, - "oxc.unusedDisableDirectives": "warn" + "oxc.unusedDisableDirectives": "warn", + "typescript.tsdk": "node_modules/typescript/lib" } diff --git a/AGENTS.md b/AGENTS.md index affea58d01f34d6826be7531dfb6f3d758725d79..7a3b300d9594832533bd1bf78efb143680af0317 100644 --- a/AGENTS.md +++ b/AGENTS.md @@ -41,3 +41,5 @@ Docs: - Codex-Monitor (Tauri, feature-complete, strong reference implementation): https://github.com/Dimillian/CodexMonitor Use these as implementation references when designing protocol handling, UX flows, and operational safeguards. + + diff --git a/SPEC.md b/SPEC.md new file mode 100644 index 0000000000000000000000000000000000000000..95266217808c9a5b1ca8dfa63f430f2488e922b1 --- /dev/null +++ b/SPEC.md @@ -0,0 +1,569 @@ +# Orchestration Interface Spec (Required State) + +Status: Required target architecture (not a description of current behavior). + +## 1. Identifier Schema (Branded) + +All identifiers are branded and never plain `string` in schema definitions. + +```ts +const ThreadId = Schema.String.pipe(Schema.brand("ThreadId")); +const ProjectId = Schema.String.pipe(Schema.brand("ProjectId")); +const CommandId = Schema.String.pipe(Schema.brand("CommandId")); +const EventId = Schema.String.pipe(Schema.brand("EventId")); +const MessageId = Schema.String.pipe(Schema.brand("MessageId")); +const TurnId = Schema.String.pipe(Schema.brand("TurnId")); + +const ProviderSessionId = Schema.String.pipe(Schema.brand("ProviderSessionId")); +const ProviderThreadId = Schema.String.pipe(Schema.brand("ProviderThreadId")); +const ProviderTurnId = Schema.String.pipe(Schema.brand("ProviderTurnId")); +const ProviderItemId = Schema.String.pipe(Schema.brand("ProviderItemId")); +const ApprovalRequestId = Schema.String.pipe(Schema.brand("ApprovalRequestId")); + +const CheckpointRef = Schema.String.pipe(Schema.brand("CheckpointRef")); +``` + +Rules: + +- `ThreadId` is app/orchestration thread identity. +- `ProviderThreadId` is provider runtime thread identity. +- They are never assigned to each other. +- `ProviderSessionId` is internal server/provider routing identity; clients do not send it. +- use e.g. `ThreadId.make(randomUUID())` to create new entity ids at the source (e.g. client create thread id and send to server, provider adapter creates the provider thread id from the codex app server etc) + +## 2. Client Commands (Domain Commands) + +All commands are sent through one RPC: `orchestration.dispatchCommand`. + +Output for every command: `DispatchResultSchema = { sequence: number }`. + +### 2.1 Client-Dispatchable Commands + +1. `project.create` + +- Input: + - `commandId: CommandId` + - `projectId: ProjectId` + - `title: string` + - `workspaceRoot: string` + - `defaultModel?: string` + - `createdAt: IsoDateTime` +- Output: `DispatchResult` + +2. `project.meta.update` + +- Input: + - `commandId: CommandId` + - `projectId: ProjectId` + - optional `{ title, workspaceRoot, defaultModel }` +- Output: `DispatchResult` + +3. `project.delete` + +- Input: `commandId`, `projectId` +- Output: `DispatchResult` + +4. `thread.create` + +- Input: + - `commandId: CommandId` + - `threadId: ThreadId` + - `projectId: ProjectId` + - `title: string` + - `model: string` + - `branch: string | null` + - `worktreePath: string | null` + - `createdAt: IsoDateTime` +- Output: `DispatchResult` + +5. `thread.delete` + +- Input: `commandId`, `threadId` +- Output: `DispatchResult` + +6. `thread.meta.update` + +- Input: `commandId`, `threadId`, optional `{ title, model, branch, worktreePath }` +- Output: `DispatchResult` + +7. `thread.turn.start` + +- Input: + - `commandId` + - `threadId` + - `message`: + - `messageId: MessageId` + - `role: "user"` + - `text: string` + - `attachments: ChatAttachment[]` (schema-shared) + - `model?: string` + - `effort?: string` + - `createdAt` +- Output: `DispatchResult` + +8. `thread.turn.interrupt` + +- Input: `commandId`, `threadId`, optional `turnId: TurnId`, `createdAt` +- Output: `DispatchResult` + +9. `thread.approval.respond` + +- Input: + - `commandId` + - `threadId` + - `requestId: ApprovalRequestId` + - `decision: "accept" | "acceptForSession" | "decline" | "cancel"` + - `createdAt` +- Output: `DispatchResult` + +10. `thread.checkpoint.revert` + +- Input: `commandId`, `threadId`, `turnCount: number`, `createdAt` +- Output: `DispatchResult` + +11. `thread.session.stop` + +- Input: `commandId`, `threadId`, `createdAt` +- Output: `DispatchResult` + +### 2.2 Internal-Only Commands (Not Client Dispatchable) + +1. `thread.session.set` + +- Server-owned projection update for session lifecycle. + +2. `thread.message.assistant.delta` + +- Server-owned incremental assistant content append. + +3. `thread.message.assistant.complete` + +- Server-owned assistant message completion marker. + +4. `thread.turn.diff.complete` + +- Server-owned checkpoint diff summary write. + +5. `thread.activity.append` + +- Server-owned activity feed append. + +## 3. Client RPC (Required) + +These are the required client-facing RPC methods. + +1. `orchestration.getSnapshot` + +- Input: `{}` +- Output: `OrchestrationReadModel` + +2. `orchestration.dispatchCommand` + +- Input: `ClientOrchestrationCommand` (`Schema.Union` of all client-dispatchable commands) +- Output: `DispatchResult` + +3. `orchestration.getTurnDiff` + +- Input: + - `threadId: ThreadId` + - `fromTurnCount: number` + - `toTurnCount: number` +- Output: + - `threadId: ThreadId` + - `fromTurnCount: number` + - `toTurnCount: number` + - `diff: string` + +4. `orchestration.replayEvents` (optional operational/debug API) + +- Input: `{ fromSequenceExclusive: number }` +- Output: `OrchestrationEvent[]` + +### 3.1 RPC To Remove + +Remove from client boundary: + +- `providers.startSession` +- `providers.sendTurn` +- `providers.interruptTurn` +- `providers.respondToRequest` +- `providers.stopSession` +- `providers.listCheckpoints` +- `providers.getCheckpointDiff` +- `providers.revertToCheckpoint` + +Rationale: + +- Session/turn/checkpoint control belongs to orchestrator command path. +- Snapshot already includes checkpoint summaries (`checkpoints`); only full textual diff needs explicit query RPC. + +## 4. Server Orchestrator Command <-> Event Schemas + +Each command deterministically yields one or more domain events that are persisted to the event store. + +1. `project.create` -> `project.created` +2. `project.meta.update` -> `project.meta-updated` +3. `project.delete` -> `project.deleted` +4. `thread.create` -> `thread.created` +5. `thread.delete` -> `thread.deleted` +6. `thread.meta.update` -> `thread.meta-updated` +7. `thread.turn.start` -> + +- `thread.message-sent` (user message append) +- `thread.turn-start-requested` +- then (async side effect success) `thread.session-set` + provider-driven message/turn events + +8. `thread.turn.interrupt` -> `thread.turn-interrupt-requested` +9. `thread.approval.respond` -> `thread.approval-response-requested` +10. `thread.checkpoint.revert` -> `thread.checkpoint-revert-requested` then `thread.reverted` +11. `thread.session.stop` -> `thread.session-stop-requested` then `thread.session-set` + +Internal command/event path: + +- `thread.session.set` -> `thread.session-set` +- `thread.message.assistant.delta` -> `thread.message-sent` (assistant streaming) +- `thread.message.assistant.complete` -> `thread.message-sent` (streaming false) +- `thread.turn.diff.complete` -> `thread.turn-diff-completed` +- `thread.activity.append` -> `thread.activity-appended` + +## 5. Orchestration Events <-> Generic Provider Events <-> Raw Provider Events + +Required mapping pipeline: + +`Raw Provider Event (adapter-specific)` +-> `GenericProviderRuntimeEvent` (provider-agnostic schema) +-> `Internal Orchestration Command` +-> `Orchestration Domain Event` + +### 5.1 Mapping Table for Codex Adapter + +1. Raw `thread/started` + +- Generic: `thread.started` +- Internal command: `thread.session.set` (attach `ProviderThreadId`, status `ready|running`) +- Domain event: `thread.session-set` + +2. Raw `turn/started` + +- Generic: `turn.started` +- Internal command: `thread.session.set` (status `running`, active turn) +- Domain event: `thread.session-set` + +3. Raw `item/agentMessage/delta` + +- Generic: `message.delta` +- Internal command: `thread.message.assistant.delta` +- Domain event: `thread.message-sent` (assistant, streaming) + +4. Raw `item/completed` (agent message) + +- Generic: `message.completed` +- Internal command: `thread.message.assistant.complete` +- Domain event: `thread.message-sent` (assistant, completed) + +5. Raw `turn/completed` + +- Generic: `turn.completed` +- Internal commands: + - `thread.session.set` (ready/error) + - `thread.activity.append` (optional summary) +- Domain events: `thread.session-set`, `thread.activity-appended` + +6. Raw approval request + +- Generic: `approval.requested` +- Internal command: `thread.activity.append` (typed `requestId`, `requestKind`) +- Domain event: `thread.activity-appended` + +7. Raw approval decision ack + +- Generic: `approval.resolved` +- Internal command: `thread.activity.append` +- Domain event: `thread.activity-appended` + +8. Raw tool start/complete + +- Generic: `tool.started` / `tool.completed` +- Internal command: `thread.activity.append` +- Domain event: `thread.activity-appended` + +9. Checkpoint capture (server-produced) + +- Generic: `checkpoint.captured` +- Internal command: `thread.turn.diff.complete` +- Domain event: `thread.turn-diff-completed` + +10. Raw/runtime error + +- Generic: `runtime.error` +- Internal commands: + - `thread.session.set` (status `error`) + - `thread.activity.append` (`tone=error`) +- Domain events: `thread.session-set`, `thread.activity-appended` + +## 6. Schema Section (Boundary + Reuse) + +### 6.1 Shared Schemas (contracts package) + +Expose to both web and server: + +- `OrchestrationReadModelSchema` +- `ClientOrchestrationCommandSchema` +- `OrchestrationEventSchema` +- `OrchestrationRpcSchemas` (`getSnapshot`, `dispatchCommand`, `getTurnDiff`, optional replay) +- `GenericProviderRuntimeEventSchema` (used for diagnostics/ingestion tests) +- Shared message/activity/checkpoint summary schemas used in snapshot + +### 6.2 Server-Internal Schemas + +Not exposed to web: + +- `InternalOrchestrationCommandSchema` +- Adapter raw-event schemas and provider-specific payload schemas +- Provider session directory records and persistence-only row schemas +- Git service cache/query schemas (non-orchestration read model) + +### 6.3 Reuse Requirements + +- Snapshot checkpoint summary schema is source of truth for checkpoint list UI. +- `getTurnDiff` response reuses `ThreadId`/turn count brands and textual `diff` only. +- Activity schemas are reused for approval/tool/runtime annotations; no separate UI-only structure. +- Session schema must include both `threadId: ThreadId` and optional `providerThreadId: ProviderThreadId` (explicitly distinct). + +## 7. Persistence Model (Required) + +This section defines required persisted tables, the canonical persisted event envelope, and required projected/read tables. + +### 7.1 Write-Side Persisted Tables + +1. `orchestration_events` (append-only event store) + +- `sequence: number` (global monotonic sequence, primary key) +- `eventId: EventId` (unique) +- `aggregateKind: "project" | "thread"` +- `streamId: ProjectId | ThreadId` (aggregate stream id) +- `streamVersion: number` (per-aggregate monotonic version) +- `eventType: OrchestrationEventType` +- `occurredAt: IsoDateTime` +- `commandId: CommandId | null` +- `causationEventId: EventId | null` +- `correlationId: CommandId | null` +- `actorKind: "client" | "server" | "provider"` +- `payload: OrchestrationEventPayload` (type-validated by `eventType`) +- `metadata: OrchestrationEventMetadata` + +2. `orchestration_command_receipts` (idempotency + ack replay) + +- `commandId: CommandId` (primary key) +- `aggregateKind: "project" | "thread"` +- `aggregateId: ProjectId | ThreadId` +- `acceptedAt: IsoDateTime` +- `resultSequence: number` +- `status: "accepted" | "rejected"` +- `error: string | null` + +3. `checkpoint_diff_blobs` (large plaintext diffs; separate from summaries) + +- `threadId: ThreadId` +- `fromTurnCount: number` +- `toTurnCount: number` +- `diff: string` +- `createdAt: IsoDateTime` +- unique key: `(threadId, fromTurnCount, toTurnCount)` + +4. `provider_session_runtime` (server-internal adapter resume state) + +- `providerSessionId: ProviderSessionId` (primary key) +- `threadId: ThreadId` +- `providerName: string` +- `adapterKey: string` +- `providerThreadId: ProviderThreadId | null` +- `status: "starting" | "running" | "stopped" | "error"` +- `lastSeenAt: IsoDateTime` +- `resumeCursor: JsonValue | null` (adapter-specific opaque state) +- `runtimePayload: JsonValue | null` (adapter-specific opaque state) + +### 7.2 Canonical Persisted Event Schema + +`OrchestrationPersistedEventSchema` (full envelope): + +```ts +type OrchestrationPersistedEvent = { + sequence: number; + eventId: EventId; + aggregateKind: "project" | "thread"; + streamId: ProjectId | ThreadId; + streamVersion: number; + eventType: OrchestrationEventType; + occurredAt: IsoDateTime; + commandId: CommandId | null; + causationEventId: EventId | null; + correlationId: CommandId | null; + actorKind: "client" | "server" | "provider"; + payload: OrchestrationEventPayload; + metadata: { + providerSessionId?: ProviderSessionId; + providerThreadId?: ProviderThreadId; + providerTurnId?: ProviderTurnId; + providerItemId?: ProviderItemId; + adapterKey?: string; + requestId?: ApprovalRequestId; + ingestedAt?: IsoDateTime; + }; +}; +``` + +Rules: + +- `payload` must be schema-discriminated by `eventType`. +- provider identifiers only appear in `metadata` and provider-specific payload sub-shapes; they do not replace `ThreadId`. +- `streamVersion` is concurrency guard for aggregate writes. + +### 7.3 Required Projected Tables (Read Models) + +1. `projection_projects` + +- `projectId: ProjectId` (primary key) +- `title: string` +- `workspaceRoot: string` +- `defaultModel: string | null` +- `createdAt: IsoDateTime` +- `updatedAt: IsoDateTime` +- `deletedAt: IsoDateTime | null` + +2. `projection_threads` + +- `threadId: ThreadId` (primary key) +- `projectId: ProjectId` +- `title: string` +- `model: string` +- `branch: string | null` +- `worktreePath: string | null` +- `latestTurnId: TurnId | null` +- `createdAt: IsoDateTime` +- `updatedAt: IsoDateTime` +- `deletedAt: IsoDateTime | null` + +3. `projection_thread_messages` + +- `messageId: MessageId` (primary key) +- `threadId: ThreadId` +- `turnId: TurnId | null` +- `role: "user" | "assistant" | "system"` +- `text: string` +- `isStreaming: boolean` +- `createdAt: IsoDateTime` +- `updatedAt: IsoDateTime` + +4. `projection_thread_activities` + +- `activityId: EventId` (primary key; derived from source event) +- `threadId: ThreadId` +- `turnId: TurnId | null` +- `tone: "info" | "tool" | "approval" | "error"` +- `kind: string` +- `summary: string` +- `payload: JsonValue` +- `createdAt: IsoDateTime` + +5. `projection_thread_sessions` + +- `threadId: ThreadId` (primary key) +- `status: "idle" | "starting" | "running" | "ready" | "interrupted" | "stopped" | "error"` +- `providerName: string | null` +- `providerSessionId: ProviderSessionId | null` +- `providerThreadId: ProviderThreadId | null` +- `activeTurnId: TurnId | null` +- `lastError: string | null` +- `updatedAt: IsoDateTime` + +6. `projection_thread_turns` + +- `turnId: TurnId` (primary key) +- `threadId: ThreadId` +- `turnCount: number` +- `status: "running" | "completed" | "interrupted" | "error"` +- `userMessageId: MessageId | null` +- `assistantMessageId: MessageId | null` +- `startedAt: IsoDateTime` +- `completedAt: IsoDateTime | null` + +7. `projection_checkpoints` + +- `threadId: ThreadId` +- `turnId: TurnId` +- `checkpointTurnCount: number` +- `checkpointRef: CheckpointRef` +- `status: "ready" | "missing" | "error"` +- `files: JsonArray` (typed as file diff summary schema) +- `assistantMessageId: MessageId | null` +- `completedAt: IsoDateTime` +- unique key: `(threadId, checkpointTurnCount)` + +8. `projection_pending_approvals` + +- `requestId: ApprovalRequestId` (primary key) +- `threadId: ThreadId` +- `turnId: TurnId | null` +- `status: "pending" | "resolved"` +- `decision: "accept" | "acceptForSession" | "decline" | "cancel" | null` +- `createdAt: IsoDateTime` +- `resolvedAt: IsoDateTime | null` + +9. `projection_state` + +- `projector: string` (primary key; e.g. `threads`, `messages`, `sessions`) +- `lastAppliedSequence: number` +- `updatedAt: IsoDateTime` + +Projection consistency rules: + +- Every projector applies read-model row updates and `projection_state.lastAppliedSequence` in the same database transaction. +- Optional debug field on projection rows: `lastEventSequence: number` (not required for correctness). + +### 7.4 Snapshot and RPC Requirements + +1. `orchestration.getSnapshot` is fully served from projection tables and returns `snapshotSequence: number`. +2. Snapshot must include `projects[]` from `projection_projects`. +3. Thread snapshot must include `checkpoints[]` from `projection_checkpoints`: +- `turnId: TurnId` +- `completedAt: IsoDateTime` +- `status` +- `files[]` (`path`, `kind`, `additions`, `deletions`) +- `checkpointRef: CheckpointRef` +- `assistantMessageId?: MessageId` +- `checkpointTurnCount: number` +4. Client does not require `listCheckpoints` RPC: +- checkpoint list comes from snapshot projections +- full diff text comes from `orchestration.getTurnDiff` backed by `checkpoint_diff_blobs` +5. Provider session identity is not a client routing key: +- client acts on `ThreadId` +- server resolves provider session internally via `projection_thread_sessions` +6. `snapshotSequence` is derived from `projection_state`: +- if snapshot depends on multiple projectors, use the minimum `lastAppliedSequence` across those projectors. +7. Event subscription handoff contract: +- client performs `getSnapshot` and reads `snapshotSequence` +- client subscribes/replays with `fromSequenceExclusive = snapshotSequence` +- server guarantees no gap between snapshot visibility and subsequent event stream from that sequence. + +### 7.5 External Derived State (Non-Orchestration) + +1. Current git/worktree state is not projected from orchestration events. +2. If persisted, it belongs in a separate git cache/read model (for example `git_state_cache`) owned by a git service. +3. Orchestration may embed git metadata only when captured as a domain fact (for example checkpoint metadata at turn completion). +4. Any RPC for current git status should be outside orchestration RPC (for example `git.getThreadState`). + +### 7.6 Existing Repository Placement + +1. `ProjectsRepository` +- Fits as projection/query access on top of `projection_projects` plus orchestration command dispatch for writes. +- Must not bypass command->event append path for mutations. + +2. `CheckpointsRepository` +- Fits as projection/query access on top of `projection_checkpoints` plus `checkpoint_diff_blobs` (or git-service on-demand diff implementation). +- Represents the same concept as prior `turn_diff_summary`; canonical naming is `checkpoint`. +- Must not be an independent source of truth outside orchestration events. + +3. `ProviderSessionsRepository` +- Fits as server-internal runtime persistence on top of `provider_session_runtime`, with domain-visible state projected into `projection_thread_sessions`. +- Not part of client RPC/domain aggregate boundary. diff --git a/apps/server/integration/TestProviderAdapter.integration.ts b/apps/server/integration/TestProviderAdapter.integration.ts index e121aed10679537fbcfbf9dde07bf97fb154bf5d..c5e6aa232580f29bb1bb358b6a126da8012115e5 100644 --- a/apps/server/integration/TestProviderAdapter.integration.ts +++ b/apps/server/integration/TestProviderAdapter.integration.ts @@ -70,13 +70,11 @@ export const makeTestProviderAdapterHarness = Effect.gen(function* () { 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}'.`, - }), - ); + return yield* new ProviderAdapterValidationError({ + provider: PROVIDER, + operation: "startSession", + issue: `Expected provider '${PROVIDER}' but received '${input.provider}'.`, + }); } sessionCount += 1; @@ -121,13 +119,11 @@ export const makeTestProviderAdapterHarness = Effect.gen(function* () { 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}.`, - }), - ); + return yield* new ProviderAdapterValidationError({ + provider: PROVIDER, + operation: "sendTurn", + issue: `No queued turn response for session ${input.sessionId}.`, + }); } const assistantDeltas: string[] = []; diff --git a/apps/server/package.json b/apps/server/package.json index e05937f2b3856271675a0cd75e85a8a236ed3582..89d53aa3ccc88035745590789127f69c544c2674 100644 --- a/apps/server/package.json +++ b/apps/server/package.json @@ -13,11 +13,13 @@ "dev": "VITE_DEV_SERVER_URL=${VITE_DEV_SERVER_URL:-http://localhost:5173} bun run src/index.ts", "build": "tsdown && node scripts/bundle-client.ts", "start": "node dist/index.mjs", + "prepare": "effect-language-service patch", "typecheck": "tsc --noEmit", "test": "vitest run" }, "dependencies": { "@effect/platform-node": "4.0.0-beta.10", + "@effect/sql-sqlite-bun": "4.0.0-beta.10", "@pierre/diffs": "^1.1.0-beta.16", "effect": "4.0.0-beta.10", "node-pty": "^1.1.0", @@ -25,6 +27,7 @@ "ws": "^8.18.0" }, "devDependencies": { + "@effect/language-service": "0.75.1", "@effect/vitest": "4.0.0-beta.10", "@t3tools/contracts": "workspace:*", "@t3tools/web": "workspace:*", diff --git a/apps/server/src/checkpointing/Layers/CheckpointService.test.ts b/apps/server/src/checkpointing/Layers/CheckpointService.test.ts index 983ecdd33b91e1c7f9a67683b073994cc98c9933..b2d4479f6e4789b38a35d6252ca6aae3e6b15ec4 100644 --- a/apps/server/src/checkpointing/Layers/CheckpointService.test.ts +++ b/apps/server/src/checkpointing/Layers/CheckpointService.test.ts @@ -60,41 +60,56 @@ const makeGitRepository = Effect.gen(function* () { class InMemoryProviderAdapter implements ProviderAdapterShape { readonly provider = "codex" as const; - private active = true; private readThreadFailure: ProviderAdapterError | undefined; - private snapshot: ProviderThreadSnapshot; - private readonly session: ProviderSession; + private readonly sessions = new Map(); + private readonly snapshots = new Map(); constructor( readonly sessionId: string, readonly cwd: string, turns: ReadonlyArray, ) { + this.addSession({ + sessionId, + threadId: "thread-1", + turns, + }); + } + + addSession(input: { + readonly sessionId: string; + readonly threadId: string; + readonly turns: ReadonlyArray; + }): void { const now = new Date().toISOString(); - this.session = { - sessionId, + this.sessions.set(input.sessionId, { + sessionId: input.sessionId, provider: "codex", status: "ready", - threadId: "thread-1", - cwd, + threadId: input.threadId, + cwd: this.cwd, createdAt: now, updatedAt: now, - }; - this.snapshot = { - threadId: this.session.threadId ?? "thread-1", - turns, - }; + }); + this.snapshots.set(input.sessionId, { + threadId: input.threadId, + turns: input.turns, + }); } - setTurns(turns: ReadonlyArray): void { - this.snapshot = { - threadId: this.snapshot.threadId, + setTurns(turns: ReadonlyArray, sessionId = this.sessionId): void { + const snapshot = this.snapshots.get(sessionId); + if (!snapshot) { + return; + } + this.snapshots.set(sessionId, { + threadId: snapshot.threadId, turns, - }; + }); } - getTurns(): ReadonlyArray { - return this.snapshot.turns; + getTurns(sessionId = this.sessionId): ReadonlyArray { + return this.snapshots.get(sessionId)?.turns ?? []; } setReadThreadFailure(error: ProviderAdapterError | undefined): void { @@ -136,16 +151,15 @@ class InMemoryProviderAdapter implements ProviderAdapterShape => Effect.sync(() => { - if (sessionId === this.sessionId) { - this.active = false; - } + this.sessions.delete(sessionId); + this.snapshots.delete(sessionId); }); readonly listSessions = (): Effect.Effect> => - Effect.sync(() => (this.active ? [this.session] : [])); + Effect.sync(() => Array.from(this.sessions.values())); readonly hasSession = (sessionId: string): Effect.Effect => - Effect.succeed(this.active && sessionId === this.sessionId); + Effect.succeed(this.sessions.has(sessionId)); readonly readThread = ( sessionId: string, @@ -155,7 +169,18 @@ class InMemoryProviderAdapter implements ProviderAdapterShape { + const snapshot = this.snapshots.get(sessionId); + if (!snapshot) { + return Effect.fail( + new ProviderAdapterSessionNotFoundError({ + provider: this.provider, + sessionId, + }), + ); + } + return Effect.succeed(snapshot); + })() : Effect.fail( new ProviderAdapterSessionNotFoundError({ provider: this.provider, @@ -184,14 +209,15 @@ class InMemoryProviderAdapter implements ProviderAdapterShape => Effect.sync(() => { - this.active = false; + this.sessions.clear(); + this.snapshots.clear(); }); readonly streamEvents = Stream.empty; @@ -315,6 +341,88 @@ it.effect("captures checkpoints, diffs refs, and reverts workspace + turns", () }).pipe(Effect.provide(NodeServices.layer)), ); +it.effect( + "CheckpointServiceLive reuses checkpoint history after session churn on the same thread", + () => + Effect.gen(function* () { + const firstTurn: ProviderThreadTurnSnapshot = { + id: "turn-1", + items: [{ type: "userMessage", content: [{ type: "text", text: "first" }] }], + }; + const secondTurn: ProviderThreadTurnSnapshot = { + id: "turn-2", + items: [{ type: "agentMessage", text: "second" }], + }; + + const fixture = yield* makeFixture([]); + + yield* Effect.gen(function* () { + const fs = yield* FileSystem.FileSystem; + const path = yield* Path.Path; + const service = yield* CheckpointService; + const directory = yield* ProviderSessionDirectory; + + yield* directory.upsert({ + sessionId: fixture.sessionId, + provider: "codex", + threadId: "thread-1", + }); + yield* service.initializeForSession({ + providerSessionId: fixture.sessionId, + cwd: fixture.cwd, + }); + + yield* fs.writeFileString(path.join(fixture.cwd, "README.md"), "v2\n"); + fixture.adapter.setTurns([firstTurn]); + yield* service.captureCurrentTurn({ + providerSessionId: fixture.sessionId, + }); + + const resumedSessionId = "sess-2"; + fixture.adapter.addSession({ + sessionId: resumedSessionId, + threadId: "thread-1", + turns: [firstTurn, secondTurn], + }); + yield* directory.upsert({ + sessionId: resumedSessionId, + provider: "codex", + threadId: "thread-1", + }); + + yield* fs.writeFileString(path.join(fixture.cwd, "README.md"), "v3\n"); + yield* service.captureCurrentTurn({ + providerSessionId: resumedSessionId, + }); + + const listed = yield* service.listCheckpoints({ sessionId: resumedSessionId }); + assert.equal(listed.threadId, "thread-1"); + assert.equal(listed.checkpoints.length, 3); + assert.equal(listed.checkpoints[2]?.id, "turn-2"); + assert.equal(listed.checkpoints[2]?.isCurrent, true); + + const diff = yield* service.getCheckpointDiff({ + sessionId: resumedSessionId, + fromTurnCount: 1, + toTurnCount: 2, + }); + assert.equal(diff.threadId, "thread-1"); + assert.equal(diff.diff.includes("-v2"), true); + assert.equal(diff.diff.includes("+v3"), true); + + const reverted = yield* service.revertToCheckpoint({ + sessionId: resumedSessionId, + turnCount: 1, + }); + const afterRevert = yield* fs.readFileString(path.join(fixture.cwd, "README.md")); + assert.equal(afterRevert, "v2\n"); + assert.equal(reverted.turnCount, 1); + assert.equal(reverted.rolledBackTurns, 1); + assert.equal(fixture.adapter.getTurns(resumedSessionId).length, 1); + }).pipe(Effect.provide(fixture.layer)); + }).pipe(Effect.provide(NodeServices.layer)), +); + it.effect( "CheckpointServiceLive fails with CheckpointSessionNotFoundError for unknown sessions", () => @@ -374,7 +482,7 @@ it.effect( }); const checkpoint = yield* checkpointRepository.getCheckpoint({ - providerSessionId: fixture.sessionId, + threadId: "thread-1", turnCount: 1, }); if (checkpoint._tag !== "Some") { @@ -429,7 +537,7 @@ it.effect("CheckpointServiceLive initializes root checkpoints without requiring }); const root = yield* checkpointRepository.getCheckpoint({ - providerSessionId: fixture.sessionId, + threadId: "thread-1", turnCount: 0, }); assert.equal(root._tag, "Some"); diff --git a/apps/server/src/checkpointing/Layers/CheckpointService.ts b/apps/server/src/checkpointing/Layers/CheckpointService.ts index 907e7df71034efa1648c7f035b6eedddb6e056e3..9ec5a37011e461ac538c6ab609ca051a52fd6912 100644 --- a/apps/server/src/checkpointing/Layers/CheckpointService.ts +++ b/apps/server/src/checkpointing/Layers/CheckpointService.ts @@ -481,11 +481,7 @@ const makeCheckpointService = Effect.gen(function* () { ): Effect.Effect => directory.getThreadId(sessionId).pipe( Effect.mapError((cause) => - mapDirectoryLookupError( - sessionId, - `${operation}:getThreadId`, - cause, - ), + mapDirectoryLookupError(sessionId, `${operation}:getThreadId`, cause), ), Effect.flatMap((threadId) => Option.match(threadId, { @@ -495,10 +491,7 @@ const makeCheckpointService = Effect.gen(function* () { return Effect.succeed(normalizedThreadId); } return Effect.fail( - invariantError( - operation, - `Resolved empty thread id for session ${sessionId}.`, - ), + invariantError(operation, `Resolved empty thread id for session ${sessionId}.`), ); }, onNone: () => @@ -512,6 +505,30 @@ const makeCheckpointService = Effect.gen(function* () { ), ); + const ensureThreadMatchesSnapshot = ( + resolvedThreadId: string, + snapshot: ProviderThreadSnapshot, + operation: string, + ): Effect.Effect => { + const snapshotThreadId = snapshot.threadId.trim(); + if (snapshotThreadId.length === 0) { + return Effect.fail( + invariantError(operation, "Provider thread snapshot is missing threadId."), + ); + } + + if (snapshotThreadId !== resolvedThreadId) { + return Effect.fail( + invariantError( + operation, + `Resolved thread ${resolvedThreadId} does not match provider snapshot thread ${snapshotThreadId}.`, + ), + ); + } + + return Effect.void; + }; + const rollbackThreadSnapshot = ( sessionId: string, adapter: ProviderAdapterShape, @@ -601,29 +618,30 @@ const makeCheckpointService = Effect.gen(function* () { }; const ensureRootCheckpointMetadata = ( - providerSessionId: string, + threadId: string, operation: string, ): Effect.Effect => - checkpointRepository.getCheckpoint({ providerSessionId, turnCount: 0 }).pipe( - Effect.flatMap((rootEntry) => - Option.isSome(rootEntry) - ? Effect.void - : Effect.fail( - invariantError( - operation, - `Missing root checkpoint metadata for session ${providerSessionId}.`, + checkpointRepository + .getCheckpoint({ threadId, turnCount: 0 }) + .pipe( + Effect.flatMap((rootEntry) => + Option.isSome(rootEntry) + ? Effect.void + : Effect.fail( + invariantError( + operation, + `Missing root checkpoint metadata for thread ${threadId}.`, + ), ), - ), - ), - ); + ), + ); const getCheckpointRefOrFail = ( - providerSessionId: string, threadId: string, turnCount: number, operation: string, ): Effect.Effect => - checkpointRepository.getCheckpoint({ providerSessionId, turnCount }).pipe( + checkpointRepository.getCheckpoint({ threadId, turnCount }).pipe( Effect.mapError((error): CheckpointServiceError => error), Effect.flatMap((checkpoint): Effect.Effect => { if (Option.isNone(checkpoint)) { @@ -716,17 +734,15 @@ const makeCheckpointService = Effect.gen(function* () { } const existingRoot = yield* checkpointRepository.getCheckpoint({ - providerSessionId: input.providerSessionId, + threadId, turnCount: 0, }); if (Option.isSome(existingRoot)) { const existingRef = existingRoot.value.checkpointRef.trim(); if (existingRef.length === 0) { - return yield* Effect.fail( - invariantError( - "CheckpointService.initializeForSession", - "Existing root checkpoint metadata is missing checkpointRef.", - ), + return yield* invariantError( + "CheckpointService.initializeForSession", + "Existing root checkpoint metadata is missing checkpointRef.", ); } const exists = yield* store.hasCheckpointRef({ @@ -734,11 +750,9 @@ const makeCheckpointService = Effect.gen(function* () { checkpointRef: existingRef, }); if (!exists) { - return yield* Effect.fail( - invariantError( - "CheckpointService.initializeForSession", - "Root checkpoint metadata exists but filesystem ref is missing.", - ), + return yield* invariantError( + "CheckpointService.initializeForSession", + "Root checkpoint metadata exists but filesystem ref is missing.", ); } yield* setTrackedCheckpointCwd(input.providerSessionId, input.cwd); @@ -770,6 +784,15 @@ const makeCheckpointService = Effect.gen(function* () { adapter, "CheckpointService.captureCurrentTurn:readThread", ); + const threadId = yield* resolveThreadIdForSession( + input.providerSessionId, + "CheckpointService.captureCurrentTurn", + ); + yield* ensureThreadMatchesSnapshot( + threadId, + snapshot, + "CheckpointService.captureCurrentTurn", + ); if (snapshot.turns.length === 0) { return; } @@ -784,23 +807,18 @@ const makeCheckpointService = Effect.gen(function* () { "CheckpointService.captureCurrentTurn", ); - yield* ensureRootCheckpointMetadata( - input.providerSessionId, - "CheckpointService.captureCurrentTurn", - ); + yield* ensureRootCheckpointMetadata(threadId, "CheckpointService.captureCurrentTurn"); const existing = yield* checkpointRepository.getCheckpoint({ - providerSessionId: input.providerSessionId, + threadId, turnCount: checkpoint.turnCount, }); if (Option.isSome(existing)) { const checkpointRef = existing.value.checkpointRef.trim(); if (checkpointRef.length === 0) { - return yield* Effect.fail( - invariantError( - "CheckpointService.captureCurrentTurn", - `Checkpoint metadata for turn ${checkpoint.turnCount} is missing checkpointRef.`, - ), + return yield* invariantError( + "CheckpointService.captureCurrentTurn", + `Checkpoint metadata for turn ${checkpoint.turnCount} is missing checkpointRef.`, ); } const exists = yield* store.hasCheckpointRef({ @@ -808,11 +826,9 @@ const makeCheckpointService = Effect.gen(function* () { checkpointRef, }); if (!exists) { - return yield* Effect.fail( - invariantError( - "CheckpointService.captureCurrentTurn", - `Checkpoint metadata exists for turn ${checkpoint.turnCount} but filesystem ref is missing.`, - ), + return yield* invariantError( + "CheckpointService.captureCurrentTurn", + `Checkpoint metadata exists for turn ${checkpoint.turnCount} but filesystem ref is missing.`, ); } return; @@ -821,7 +837,7 @@ const makeCheckpointService = Effect.gen(function* () { yield* captureAndPersistCheckpoint({ providerSessionId: input.providerSessionId, cwd: resolution.checkpointCwd.value, - threadId: snapshot.threadId, + threadId, checkpoint, }); }), @@ -842,31 +858,36 @@ const makeCheckpointService = Effect.gen(function* () { adapter, "CheckpointService.listCheckpoints:readThread", ); + const threadId = yield* resolveThreadIdForSession( + input.sessionId, + "CheckpointService.listCheckpoints", + ); + yield* ensureThreadMatchesSnapshot( + threadId, + snapshot, + "CheckpointService.listCheckpoints", + ); const resolution = yield* resolveCheckpointCwd(input.sessionId, adapter); if (Option.isNone(resolution.checkpointCwd)) { - return yield* Effect.fail( - new CheckpointRepositoryError({ - cwd: resolution.candidateCwd, - detail: "Filesystem checkpoints are unavailable for this session.", - }), - ); + return yield* new CheckpointRepositoryError({ + cwd: resolution.candidateCwd, + detail: "Filesystem checkpoints are unavailable for this session.", + }); } const checkpoints = yield* checkpointRepository.listCheckpoints({ - providerSessionId: input.sessionId, + threadId, }); if (checkpoints.length === 0) { - return yield* Effect.fail( - invariantError( - "CheckpointService.listCheckpoints", - `Missing checkpoint metadata for active session ${input.sessionId}.`, - ), + return yield* invariantError( + "CheckpointService.listCheckpoints", + `Missing checkpoint metadata for active session ${input.sessionId}.`, ); } return { - threadId: snapshot.threadId, + threadId, checkpoints: Array.from(checkpoints), }; }), @@ -887,35 +908,38 @@ const makeCheckpointService = Effect.gen(function* () { adapter, "CheckpointService.getCheckpointDiff:readThread", ); + const threadId = yield* resolveThreadIdForSession( + input.sessionId, + "CheckpointService.getCheckpointDiff", + ); + yield* ensureThreadMatchesSnapshot( + threadId, + snapshot, + "CheckpointService.getCheckpointDiff", + ); if (input.toTurnCount > snapshot.turns.length) { - return yield* Effect.fail( - validationError( - "CheckpointService.getCheckpointDiff", - `Checkpoint turn count ${input.toTurnCount} exceeds current turn count ${snapshot.turns.length}.`, - ), + return yield* validationError( + "CheckpointService.getCheckpointDiff", + `Checkpoint turn count ${input.toTurnCount} exceeds current turn count ${snapshot.turns.length}.`, ); } const resolution = yield* resolveCheckpointCwd(input.sessionId, adapter); if (Option.isNone(resolution.checkpointCwd)) { - return yield* Effect.fail( - new CheckpointRepositoryError({ - cwd: resolution.candidateCwd, - detail: "Filesystem checkpoints are unavailable for this session.", - }), - ); + return yield* new CheckpointRepositoryError({ + cwd: resolution.candidateCwd, + detail: "Filesystem checkpoints are unavailable for this session.", + }); } const fromCheckpointRef = yield* getCheckpointRefOrFail( - input.sessionId, - snapshot.threadId, + threadId, input.fromTurnCount, "CheckpointService.getCheckpointDiff:fromCheckpoint", ); const toCheckpointRef = yield* getCheckpointRefOrFail( - input.sessionId, - snapshot.threadId, + threadId, input.toTurnCount, "CheckpointService.getCheckpointDiff:toCheckpoint", ); @@ -928,7 +952,7 @@ const makeCheckpointService = Effect.gen(function* () { }); return { - threadId: snapshot.threadId, + threadId, fromTurnCount: input.fromTurnCount, toTurnCount: input.toTurnCount, diff, @@ -951,31 +975,35 @@ const makeCheckpointService = Effect.gen(function* () { adapter, "CheckpointService.revertToCheckpoint:readThread", ); + const threadId = yield* resolveThreadIdForSession( + input.sessionId, + "CheckpointService.revertToCheckpoint", + ); + yield* ensureThreadMatchesSnapshot( + threadId, + beforeSnapshot, + "CheckpointService.revertToCheckpoint", + ); const currentTurnCount = beforeSnapshot.turns.length; if (input.turnCount > currentTurnCount) { - return yield* Effect.fail( - validationError( - "CheckpointService.revertToCheckpoint", - `Checkpoint turn count ${input.turnCount} exceeds current turn count ${currentTurnCount}.`, - ), + return yield* validationError( + "CheckpointService.revertToCheckpoint", + `Checkpoint turn count ${input.turnCount} exceeds current turn count ${currentTurnCount}.`, ); } const resolution = yield* resolveCheckpointCwd(input.sessionId, adapter); if (Option.isNone(resolution.checkpointCwd)) { - return yield* Effect.fail( - new CheckpointRepositoryError({ - cwd: resolution.candidateCwd, - detail: "Filesystem checkpoints are unavailable for this session.", - }), - ); + return yield* new CheckpointRepositoryError({ + cwd: resolution.candidateCwd, + detail: "Filesystem checkpoints are unavailable for this session.", + }); } const checkpointCwd = resolution.checkpointCwd.value; const checkpointRef = yield* getCheckpointRefOrFail( - input.sessionId, - beforeSnapshot.threadId, + threadId, input.turnCount, "CheckpointService.revertToCheckpoint:checkpointLookup", ); @@ -986,13 +1014,11 @@ const makeCheckpointService = Effect.gen(function* () { fallbackToHead: input.turnCount === 0, }); if (!restored) { - return yield* Effect.fail( - new CheckpointUnavailableError({ - threadId: beforeSnapshot.threadId, - turnCount: input.turnCount, - detail: `Filesystem checkpoint is unavailable for turn ${input.turnCount} in thread ${beforeSnapshot.threadId}.`, - }), - ); + return yield* new CheckpointUnavailableError({ + threadId, + turnCount: input.turnCount, + detail: `Filesystem checkpoint is unavailable for turn ${input.turnCount} in thread ${threadId}.`, + }); } const requestedRollbackTurns = currentTurnCount - input.turnCount; @@ -1015,7 +1041,7 @@ const makeCheckpointService = Effect.gen(function* () { staleTurnCounts, (turnCount) => checkpointRepository.getCheckpoint({ - providerSessionId: input.sessionId, + threadId, turnCount, }), { discard: false }, @@ -1034,11 +1060,11 @@ const makeCheckpointService = Effect.gen(function* () { checkpointRefs: staleCheckpointRefs, }); yield* checkpointRepository.deleteAfterTurn({ - providerSessionId: input.sessionId, + threadId, maxTurnCount: input.turnCount, }); const checkpoints = yield* checkpointRepository.listCheckpoints({ - providerSessionId: input.sessionId, + threadId, }); const currentCheckpoint = checkpoints.find((checkpoint) => checkpoint.isCurrent) ?? @@ -1047,7 +1073,7 @@ const makeCheckpointService = Effect.gen(function* () { const rolledBackTurns = Math.max(0, currentTurnCount - afterSnapshot.turns.length); return { - threadId: afterSnapshot.threadId, + threadId, turnCount: currentCheckpoint?.turnCount ?? 0, messageCount: currentCheckpoint?.messageCount ?? 0, rolledBackTurns, diff --git a/apps/server/src/index.ts b/apps/server/src/index.ts index 00886cb8d054eaf67739814678efe5c64e8f2d34..61c3813c17e3ffbc4526e3dc16776d10771bc72a 100644 --- a/apps/server/src/index.ts +++ b/apps/server/src/index.ts @@ -11,6 +11,8 @@ import { createServer } from "./wsServer"; fixPath(); const DEFAULT_PORT = 3773; +// Dummy comment for a no-op change request. +// Additional dummy comment requested in chat. const cwd = process.cwd(); const logger = createLogger("server"); diff --git a/apps/server/src/persistence/Layers/Checkpoints.test.ts b/apps/server/src/persistence/Layers/Checkpoints.test.ts index 437bf734733c0151ae77f399ad23f6301778c22b..99f85748ce6259986a75fa57e44967038d4effc0 100644 --- a/apps/server/src/persistence/Layers/Checkpoints.test.ts +++ b/apps/server/src/persistence/Layers/Checkpoints.test.ts @@ -11,72 +11,105 @@ function makeCheckpointRepositoryLayer() { it.effect("CheckpointRepositoryLive upserts, lists, reads, and prunes checkpoint metadata", () => Effect.gen(function* () { - yield* Effect.gen(function* () { - const repository = yield* CheckpointRepository; + const repository = yield* CheckpointRepository; - yield* repository.upsertCheckpoint({ - providerSessionId: "sess-1", - threadId: "thread-1", - checkpointId: "root", - checkpointRef: "refs/t3/checkpoints/thread-1/turn/0", - turnCount: 0, - messageCount: 0, - label: "Start of conversation", - createdAt: new Date().toISOString(), - }); + yield* repository.upsertCheckpoint({ + providerSessionId: "sess-1", + threadId: "thread-1", + checkpointId: "root", + checkpointRef: "refs/t3/checkpoints/thread-1/turn/0", + turnCount: 0, + messageCount: 0, + label: "Start of conversation", + createdAt: new Date().toISOString(), + }); - yield* repository.upsertCheckpoint({ - providerSessionId: "sess-1", - threadId: "thread-1", - checkpointId: "turn-1", - checkpointRef: "refs/t3/checkpoints/thread-1/turn/1", - turnCount: 1, - messageCount: 2, - label: "Turn 1", - preview: "hello", - createdAt: new Date().toISOString(), - }); + yield* repository.upsertCheckpoint({ + providerSessionId: "sess-1", + threadId: "thread-1", + checkpointId: "turn-1", + checkpointRef: "refs/t3/checkpoints/thread-1/turn/1", + turnCount: 1, + messageCount: 2, + label: "Turn 1", + preview: "hello", + createdAt: new Date().toISOString(), + }); - yield* repository.upsertCheckpoint({ - providerSessionId: "sess-1", - threadId: "thread-1", - checkpointId: "turn-2", - checkpointRef: "refs/t3/checkpoints/thread-1/turn/2", - turnCount: 2, - messageCount: 4, - label: "Turn 2", - preview: "done", - createdAt: new Date().toISOString(), - }); + yield* repository.upsertCheckpoint({ + providerSessionId: "sess-1", + threadId: "thread-1", + checkpointId: "turn-2", + checkpointRef: "refs/t3/checkpoints/thread-1/turn/2", + turnCount: 2, + messageCount: 4, + label: "Turn 2", + preview: "done", + createdAt: new Date().toISOString(), + }); - const checkpoints = yield* repository.listCheckpoints({ providerSessionId: "sess-1" }); - assert.equal(checkpoints.length, 3); - assert.equal(checkpoints[0]?.id, "root"); - assert.equal(checkpoints[0]?.isCurrent, false); - assert.equal(checkpoints[2]?.id, "turn-2"); - assert.equal(checkpoints[2]?.isCurrent, true); + const checkpoints = yield* repository.listCheckpoints({ threadId: "thread-1" }); + assert.equal(checkpoints.length, 3); + assert.equal(checkpoints[0]?.id, "root"); + assert.equal(checkpoints[0]?.isCurrent, false); + assert.equal(checkpoints[2]?.id, "turn-2"); + assert.equal(checkpoints[2]?.isCurrent, true); - const one = yield* repository.getCheckpoint({ providerSessionId: "sess-1", turnCount: 1 }); - assert.equal(Option.isSome(one), true); - if (Option.isSome(one)) { - assert.equal(one.value.id, "turn-1"); - assert.equal(one.value.isCurrent, false); - assert.equal(one.value.checkpointRef, "refs/t3/checkpoints/thread-1/turn/1"); - } + const one = yield* repository.getCheckpoint({ threadId: "thread-1", turnCount: 1 }); + assert.equal(Option.isSome(one), true); + if (Option.isSome(one)) { + assert.equal(one.value.id, "turn-1"); + assert.equal(one.value.isCurrent, false); + assert.equal(one.value.checkpointRef, "refs/t3/checkpoints/thread-1/turn/1"); + } - yield* repository.deleteAfterTurn({ - providerSessionId: "sess-1", - maxTurnCount: 1, - }); + yield* repository.upsertCheckpoint({ + providerSessionId: "sess-2", + threadId: "thread-1", + checkpointId: "turn-3", + checkpointRef: "refs/t3/checkpoints/thread-1/turn/3", + turnCount: 3, + messageCount: 6, + label: "Turn 3", + preview: "new session", + createdAt: new Date().toISOString(), + }); - const pruned = yield* repository.listCheckpoints({ providerSessionId: "sess-1" }); - assert.equal(pruned.length, 2); - assert.equal(pruned[1]?.turnCount, 1); - assert.equal(pruned[1]?.isCurrent, true); + const resumed = yield* repository.listCheckpoints({ threadId: "thread-1" }); + assert.equal(resumed.length, 4); + assert.equal(resumed[3]?.id, "turn-3"); + assert.equal(resumed[3]?.isCurrent, true); - yield* repository.deleteAllForSession({ providerSessionId: "sess-1" }); - const empty = yield* repository.listCheckpoints({ providerSessionId: "sess-1" }); - assert.equal(empty.length, 0); - }).pipe(Effect.provide(makeCheckpointRepositoryLayer())); - }), + yield* repository.upsertCheckpoint({ + providerSessionId: "sess-3", + threadId: "thread-2", + checkpointId: "root", + checkpointRef: "refs/t3/checkpoints/thread-2/turn/0", + turnCount: 0, + messageCount: 0, + label: "Start of conversation", + createdAt: new Date().toISOString(), + }); + + const otherThread = yield* repository.listCheckpoints({ threadId: "thread-2" }); + assert.equal(otherThread.length, 1); + assert.equal(otherThread[0]?.id, "root"); + + yield* repository.deleteAfterTurn({ + threadId: "thread-1", + maxTurnCount: 1, + }); + + const pruned = yield* repository.listCheckpoints({ threadId: "thread-1" }); + assert.equal(pruned.length, 2); + assert.equal(pruned[1]?.turnCount, 1); + assert.equal(pruned[1]?.isCurrent, true); + + yield* repository.deleteAllForThread({ threadId: "thread-1" }); + const empty = yield* repository.listCheckpoints({ threadId: "thread-1" }); + assert.equal(empty.length, 0); + + const stillPresent = yield* repository.listCheckpoints({ threadId: "thread-2" }); + assert.equal(stillPresent.length, 1); + }).pipe(Effect.provide(makeCheckpointRepositoryLayer())), ); diff --git a/apps/server/src/persistence/Layers/Checkpoints.ts b/apps/server/src/persistence/Layers/Checkpoints.ts index c6228c80974b9aade0cec93f890c30ae5e9148aa..b30f658243d9e63b841b4143aea7f808c65e2587 100644 --- a/apps/server/src/persistence/Layers/Checkpoints.ts +++ b/apps/server/src/persistence/Layers/Checkpoints.ts @@ -29,7 +29,7 @@ import { const NonEmptyString = Schema.NonEmptyString; const CheckpointRowSchema = Schema.Struct({ - providerSessionId: Schema.String, + providerSessionId: Schema.NullOr(Schema.String), threadId: Schema.String, checkpointId: Schema.String, checkpointRef: Schema.String, @@ -40,12 +40,12 @@ const CheckpointRowSchema = Schema.Struct({ createdAt: Schema.String, }); -const SessionRequestSchema = Schema.Struct({ - providerSessionId: NonEmptyString, +const ThreadRequestSchema = Schema.Struct({ + threadId: NonEmptyString, }); const UpsertRequestSchema = Schema.Struct({ - providerSessionId: NonEmptyString, + providerSessionId: Schema.NullOr(NonEmptyString), threadId: NonEmptyString, checkpointId: NonEmptyString, checkpointRef: NonEmptyString, @@ -57,12 +57,12 @@ const UpsertRequestSchema = Schema.Struct({ }); const GetCheckpointRequestSchema = Schema.Struct({ - providerSessionId: NonEmptyString, + threadId: NonEmptyString, turnCount: Schema.Int, }); const DeleteAfterTurnRequestSchema = Schema.Struct({ - providerSessionId: NonEmptyString, + threadId: NonEmptyString, maxTurnCount: Schema.Int, }); @@ -176,8 +176,9 @@ const makeCheckpointRepository = Effect.gen(function* () { ${request.preview}, ${request.createdAt} ) - ON CONFLICT (provider_session_id, turn_count) + ON CONFLICT (thread_id, turn_count) DO UPDATE SET + provider_session_id = excluded.provider_session_id, thread_id = excluded.thread_id, checkpoint_id = excluded.checkpoint_id, checkpoint_ref = excluded.checkpoint_ref, @@ -199,9 +200,9 @@ const makeCheckpointRepository = Effect.gen(function* () { }); const listCheckpointRows = SqlSchema.findAll({ - Request: SessionRequestSchema, + Request: ThreadRequestSchema, Result: CheckpointRowSchema, - execute: ({ providerSessionId }) => + execute: ({ threadId }) => sql` SELECT provider_session_id AS "providerSessionId", @@ -214,15 +215,15 @@ const makeCheckpointRepository = Effect.gen(function* () { preview, created_at AS "createdAt" FROM provider_checkpoints - WHERE provider_session_id = ${providerSessionId} + WHERE thread_id = ${threadId} ORDER BY turn_count ASC `, }); - const listForSession = ( - providerSessionId: string, + const listForThread = ( + threadId: string, ): Effect.Effect, CheckpointRepositoryError> => - listCheckpointRows({ providerSessionId }).pipe( + listCheckpointRows({ threadId }).pipe( Effect.mapError((cause) => toPersistenceError("CheckpointRepository.listCheckpoints:query", cause), ), @@ -238,7 +239,7 @@ const makeCheckpointRepository = Effect.gen(function* () { const parsed = yield* decodeInput( UpsertRequestSchema, { - providerSessionId: input.providerSessionId.trim(), + providerSessionId: input.providerSessionId?.trim() || null, threadId: input.threadId.trim(), checkpointId: input.checkpointId.trim(), checkpointRef: input.checkpointRef.trim(), @@ -264,14 +265,14 @@ const makeCheckpointRepository = Effect.gen(function* () { const listCheckpoints: CheckpointRepositoryShape["listCheckpoints"] = (input) => Effect.gen(function* () { const parsed = yield* decodeInput( - SessionRequestSchema, + ThreadRequestSchema, { - providerSessionId: input.providerSessionId.trim(), + threadId: input.threadId.trim(), }, "CheckpointRepository.listCheckpoints", ); - const entries = yield* listForSession(parsed.providerSessionId); + const entries = yield* listForThread(parsed.threadId); return entries.map(toProviderCheckpoint); }); @@ -280,7 +281,7 @@ const makeCheckpointRepository = Effect.gen(function* () { const parsed = yield* decodeInput( GetCheckpointRequestSchema, { - providerSessionId: input.providerSessionId.trim(), + threadId: input.threadId.trim(), turnCount: input.turnCount, }, "CheckpointRepository.getCheckpoint", @@ -291,7 +292,7 @@ const makeCheckpointRepository = Effect.gen(function* () { parsed.turnCount, ); - const entries = yield* listForSession(parsed.providerSessionId); + const entries = yield* listForThread(parsed.threadId); const checkpoint = entries.find((entry) => entry.turnCount === parsed.turnCount); return Option.fromNullishOr(checkpoint); }); @@ -301,7 +302,7 @@ const makeCheckpointRepository = Effect.gen(function* () { const parsed = yield* decodeInput( DeleteAfterTurnRequestSchema, { - providerSessionId: input.providerSessionId.trim(), + threadId: input.threadId.trim(), maxTurnCount: input.maxTurnCount, }, "CheckpointRepository.deleteAfterTurn", @@ -314,7 +315,7 @@ const makeCheckpointRepository = Effect.gen(function* () { yield* sql` DELETE FROM provider_checkpoints - WHERE provider_session_id = ${parsed.providerSessionId} + WHERE thread_id = ${parsed.threadId} AND turn_count > ${parsed.maxTurnCount} `.pipe( Effect.mapError((cause) => @@ -323,22 +324,22 @@ const makeCheckpointRepository = Effect.gen(function* () { ); }); - const deleteAllForSession: CheckpointRepositoryShape["deleteAllForSession"] = (input) => + const deleteAllForThread: CheckpointRepositoryShape["deleteAllForThread"] = (input) => Effect.gen(function* () { const parsed = yield* decodeInput( - SessionRequestSchema, + ThreadRequestSchema, { - providerSessionId: input.providerSessionId.trim(), + threadId: input.threadId.trim(), }, - "CheckpointRepository.deleteAllForSession", + "CheckpointRepository.deleteAllForThread", ); yield* sql` DELETE FROM provider_checkpoints - WHERE provider_session_id = ${parsed.providerSessionId} + WHERE thread_id = ${parsed.threadId} `.pipe( Effect.mapError((cause) => - toPersistenceError("CheckpointRepository.deleteAllForSession:query", cause), + toPersistenceError("CheckpointRepository.deleteAllForThread:query", cause), ), ); }); @@ -348,7 +349,7 @@ const makeCheckpointRepository = Effect.gen(function* () { listCheckpoints, getCheckpoint, deleteAfterTurn, - deleteAllForSession, + deleteAllForThread, } satisfies CheckpointRepositoryShape; }); diff --git a/apps/server/src/persistence/Layers/Projects.ts b/apps/server/src/persistence/Layers/Projects.ts index fd6426ecb26e2d7537d9d883c38b06faf670f283..91ea0b695af7a91019c39e4acf0adc6687ea52f4 100644 --- a/apps/server/src/persistence/Layers/Projects.ts +++ b/apps/server/src/persistence/Layers/Projects.ts @@ -333,7 +333,7 @@ const makeRepository = Effect.gen(function* () { const input = yield* decodeAddInput(rawInput); const normalizedCwd = normalizeCwd(input.cwd); if (!isDirectory(normalizedCwd)) { - return yield* Effect.fail(new ProjectPathMissingError({ cwd: normalizedCwd })); + return yield* new ProjectPathMissingError({ cwd: normalizedCwd }); } const existing = yield* findProjectByCwd({ cwd: normalizedCwd }).pipe( diff --git a/apps/server/src/persistence/Layers/Sqlite.ts b/apps/server/src/persistence/Layers/Sqlite.ts index 7a700c878d46039ae97433d257d55b295a6139bd..b12d4a9b73c5dda17f76ea7b71c4503915a1ad0e 100644 --- a/apps/server/src/persistence/Layers/Sqlite.ts +++ b/apps/server/src/persistence/Layers/Sqlite.ts @@ -1,11 +1,34 @@ import { Effect, Layer, FileSystem, Path } from "effect"; +import * as SqlClient from "effect/unstable/sql/SqlClient"; import { runMigrations } from "../Migrations.ts"; -import * as SqliteClient from "../SqliteClient.ts"; + +type RuntimeSqliteLayerConfig = { + readonly filename: string; + readonly disableWAL?: boolean | undefined; +}; + +type Loader = { + layer: (config: RuntimeSqliteLayerConfig) => Layer.Layer; +}; +const defaultSqliteClientLoaders = { + bun: () => import("@effect/sql-sqlite-bun/SqliteClient"), + node: () => import("../NodeSqliteClient.ts"), +} satisfies Record Promise>; + +const makeRuntimeSqliteLayer = ( + config: RuntimeSqliteLayerConfig, +): Layer.Layer => + Effect.gen(function* () { + const runtime = process.versions.bun !== undefined ? "bun" : "node"; + const loader = defaultSqliteClientLoaders[runtime]; + const clientModule = yield* Effect.promise(loader); + return clientModule.layer(config); + }).pipe(Layer.unwrap); const setup = Layer.effectDiscard( Effect.gen(function* () { - const sql = yield* SqliteClient.SqliteClient; + const sql = yield* SqlClient.SqlClient; yield* sql`PRAGMA journal_mode = WAL;`; yield* sql`PRAGMA foreign_keys = ON;`; yield* runMigrations; @@ -18,7 +41,13 @@ export const makeSqlitePersistenceLive = (dbPath: string) => const path = yield* Path.Path; yield* fs.makeDirectory(path.dirname(dbPath), { recursive: true }); - return Layer.provideMerge(setup, SqliteClient.layer({ filename: dbPath })); + return Layer.provideMerge(setup, makeRuntimeSqliteLayer({ filename: dbPath })); }).pipe(Layer.unwrap); -export const SqlitePersistenceMemory = Layer.provideMerge(setup, SqliteClient.layerMemory()); +export const SqlitePersistenceMemory = Layer.provideMerge( + setup, + makeRuntimeSqliteLayer({ + filename: ":memory:", + disableWAL: true, + }), +); diff --git a/apps/server/src/persistence/Migrations/003_Checkpoints.ts b/apps/server/src/persistence/Migrations/003_Checkpoints.ts index c7dc9ea3c7eaf4a7587404693c78fc9736f3d346..661bcb5ca158e97f5e876dece992b555002ea6da 100644 --- a/apps/server/src/persistence/Migrations/003_Checkpoints.ts +++ b/apps/server/src/persistence/Migrations/003_Checkpoints.ts @@ -6,7 +6,7 @@ export default Effect.gen(function* () { yield* sql` CREATE TABLE IF NOT EXISTS provider_checkpoints ( - provider_session_id TEXT NOT NULL, + provider_session_id TEXT, thread_id TEXT NOT NULL, checkpoint_id TEXT NOT NULL, checkpoint_ref TEXT NOT NULL, @@ -15,12 +15,17 @@ export default Effect.gen(function* () { label TEXT NOT NULL, preview TEXT, created_at TEXT NOT NULL, - PRIMARY KEY (provider_session_id, turn_count) + PRIMARY KEY (thread_id, turn_count) ) `; yield* sql` - CREATE INDEX IF NOT EXISTS idx_provider_checkpoints_session_turn - ON provider_checkpoints(provider_session_id, turn_count ASC) + CREATE INDEX IF NOT EXISTS idx_provider_checkpoints_thread_turn + ON provider_checkpoints(thread_id, turn_count ASC) + `; + + yield* sql` + CREATE INDEX IF NOT EXISTS idx_provider_checkpoints_session + ON provider_checkpoints(provider_session_id) `; }); diff --git a/apps/server/src/persistence/SqliteClient.test.ts b/apps/server/src/persistence/NodeSqliteClient.test.ts similarity index 94% rename from apps/server/src/persistence/SqliteClient.test.ts rename to apps/server/src/persistence/NodeSqliteClient.test.ts index 68902626cea3e99a23bf9291db8117c041b5b752..18c24ec64517d815b8832577442f9fd953a17509 100644 --- a/apps/server/src/persistence/SqliteClient.test.ts +++ b/apps/server/src/persistence/NodeSqliteClient.test.ts @@ -2,7 +2,7 @@ import { assert, it } from "@effect/vitest"; import { Effect } from "effect"; import * as SqlClient from "effect/unstable/sql/SqlClient"; -import * as SqliteClient from "./SqliteClient.ts"; +import * as SqliteClient from "./NodeSqliteClient.ts"; it.effect("runs prepared queries and returns positional values", () => Effect.gen(function* () { diff --git a/apps/server/src/persistence/SqliteClient.ts b/apps/server/src/persistence/NodeSqliteClient.ts similarity index 100% rename from apps/server/src/persistence/SqliteClient.ts rename to apps/server/src/persistence/NodeSqliteClient.ts diff --git a/apps/server/src/persistence/Services/Checkpoints.ts b/apps/server/src/persistence/Services/Checkpoints.ts index 954d1c8b74c04ba12415196bac14cce1f2ad3c92..f92bef6c16203c9252a11cae5addffdf0065568d 100644 --- a/apps/server/src/persistence/Services/Checkpoints.ts +++ b/apps/server/src/persistence/Services/Checkpoints.ts @@ -2,7 +2,7 @@ * CheckpointRepository - Repository interface for checkpoint metadata persistence. * * Stores and queries checkpoint metadata and checkpoint Git refs keyed by - * provider session + turn count. It does not read/write Git refs itself and it + * thread + turn count. It does not read/write Git refs itself and it * does not coordinate provider rollback workflows. * * @module CheckpointRepository @@ -14,7 +14,7 @@ import type { Effect } from "effect"; import type { CheckpointRepositoryError } from "../Errors.ts"; export interface UpsertCheckpointInput { - readonly providerSessionId: string; + readonly providerSessionId?: string; readonly threadId: string; readonly checkpointId: string; readonly checkpointRef: string; @@ -31,22 +31,22 @@ export interface CheckpointRepositoryEntry extends ProviderCheckpoint { readonly createdAt: string; } -export interface ListSessionCheckpointsInput { - readonly providerSessionId: string; +export interface ListThreadCheckpointsInput { + readonly threadId: string; } export interface GetCheckpointInput { - readonly providerSessionId: string; + readonly threadId: string; readonly turnCount: number; } export interface DeleteAfterTurnInput { - readonly providerSessionId: string; + readonly threadId: string; readonly maxTurnCount: number; } -export interface DeleteAllForSessionInput { - readonly providerSessionId: string; +export interface DeleteAllForThreadInput { + readonly threadId: string; } export interface CheckpointRepositoryShape { @@ -58,10 +58,10 @@ export interface CheckpointRepositoryShape { ) => Effect.Effect; /** - * List user-facing checkpoints for a provider session. + * List user-facing checkpoints for a thread. */ readonly listCheckpoints: ( - input: ListSessionCheckpointsInput, + input: ListThreadCheckpointsInput, ) => Effect.Effect, CheckpointRepositoryError>; /** @@ -79,10 +79,10 @@ export interface CheckpointRepositoryShape { ) => Effect.Effect; /** - * Delete all checkpoint metadata for a provider session. + * Delete all checkpoint metadata for a thread. */ - readonly deleteAllForSession: ( - input: DeleteAllForSessionInput, + readonly deleteAllForThread: ( + input: DeleteAllForThreadInput, ) => Effect.Effect; } diff --git a/apps/server/src/provider/Layers/ProviderService.test.ts b/apps/server/src/provider/Layers/ProviderService.test.ts index 205d2e2f4d4924326cc8d6eda955ad05c899f7af..2bd5f6b38d0711a8ce9a8e1a5db9a88d400d3b54 100644 --- a/apps/server/src/provider/Layers/ProviderService.test.ts +++ b/apps/server/src/provider/Layers/ProviderService.test.ts @@ -55,7 +55,7 @@ function makeFakeCodexAdapter() { sessionId: `sess-${nextSession}`, provider: "codex", status: "ready", - threadId: `thread-${nextSession}`, + threadId: input.resumeThreadId ?? `thread-${nextSession}`, cwd: input.cwd ?? process.cwd(), createdAt: now, updatedAt: now, @@ -281,7 +281,7 @@ function makeProviderServiceLayer() { } const routing = makeProviderServiceLayer(); -it.effect("ProviderServiceLive reconciles stale persisted sessions on startup", () => +it.effect("ProviderServiceLive keeps persisted resumable sessions on startup", () => Effect.gen(function* () { const tempDir = fs.mkdtempSync(path.join(os.tmpdir(), "t3-provider-service-")); const dbPath = path.join(tempDir, "orchestration.sqlite"); @@ -322,7 +322,100 @@ it.effect("ProviderServiceLive reconciles stale persisted sessions on startup", yield* ProviderService; }).pipe(Effect.provide(providerLayer)); - assert.deepEqual(checkpoint.releaseSession.mock.calls, [[{ providerSessionId: "sess-stale" }]]); + assert.deepEqual(checkpoint.releaseSession.mock.calls, []); + const persistedProvider = yield* Effect.gen(function* () { + const directory = yield* ProviderSessionDirectory; + return yield* directory.getProvider("sess-stale"); + }).pipe(Effect.provide(directoryLayer)); + assert.equal(persistedProvider, "codex"); + + fs.rmSync(tempDir, { recursive: true, force: true }); + }).pipe(Effect.provide(NodeServices.layer)), +); + +it.effect("ProviderServiceLive restores checkpoint RPC routing after restart using persisted thread mapping", () => + Effect.gen(function* () { + const tempDir = fs.mkdtempSync(path.join(os.tmpdir(), "t3-provider-service-restart-")); + const dbPath = path.join(tempDir, "orchestration.sqlite"); + const persistenceLayer = makeSqlitePersistenceLive(dbPath); + + const checkpoint = makeCheckpointServiceDouble(); + const firstCodex = makeFakeCodexAdapter(); + const firstRegistry: typeof ProviderAdapterRegistry.Service = { + getByProvider: (provider) => + provider === "codex" + ? Effect.succeed(firstCodex.adapter) + : Effect.fail(new ProviderUnsupportedError({ provider })), + listProviders: () => Effect.succeed(["codex"]), + }; + + const firstDirectoryLayer = ProviderSessionDirectoryLive.pipe( + Layer.provide(ProviderSessionRepositoryLive), + Layer.provide(persistenceLayer), + ); + const firstProviderLayer = ProviderServiceLive.pipe( + Layer.provide(Layer.succeed(ProviderAdapterRegistry, firstRegistry)), + Layer.provide(firstDirectoryLayer), + Layer.provide(Layer.succeed(CheckpointService, checkpoint.service)), + ); + + const startedSession = yield* Effect.gen(function* () { + const provider = yield* ProviderService; + return yield* provider.startSession({ + provider: "codex", + cwd: "/tmp/project", + }); + }).pipe(Effect.provide(firstProviderLayer)); + + yield* Effect.gen(function* () { + const provider = yield* ProviderService; + yield* provider.stopAll(); + }).pipe(Effect.provide(firstProviderLayer)); + + const secondCodex = makeFakeCodexAdapter(); + const secondRegistry: typeof ProviderAdapterRegistry.Service = { + getByProvider: (provider) => + provider === "codex" + ? Effect.succeed(secondCodex.adapter) + : Effect.fail(new ProviderUnsupportedError({ provider })), + listProviders: () => Effect.succeed(["codex"]), + }; + const secondDirectoryLayer = ProviderSessionDirectoryLive.pipe( + Layer.provide(ProviderSessionRepositoryLive), + Layer.provide(persistenceLayer), + ); + const secondProviderLayer = ProviderServiceLive.pipe( + Layer.provide(Layer.succeed(ProviderAdapterRegistry, secondRegistry)), + Layer.provide(secondDirectoryLayer), + Layer.provide(Layer.succeed(CheckpointService, checkpoint.service)), + ); + + checkpoint.getCheckpointDiff.mockClear(); + secondCodex.startSession.mockClear(); + + yield* Effect.gen(function* () { + const provider = yield* ProviderService; + yield* provider.getCheckpointDiff({ + sessionId: startedSession.sessionId, + fromTurnCount: 0, + toTurnCount: 0, + }); + }).pipe(Effect.provide(secondProviderLayer)); + + assert.deepEqual(secondCodex.startSession.mock.calls, [ + [ + { + provider: "codex", + resumeThreadId: startedSession.threadId, + }, + ], + ]); + assert.equal(checkpoint.getCheckpointDiff.mock.calls.length, 1); + const recoveredDiffInput = checkpoint.getCheckpointDiff.mock.calls[0]?.[0]; + assert.equal(typeof recoveredDiffInput?.sessionId, "string"); + assert.equal((recoveredDiffInput?.sessionId?.length ?? 0) > 0, true); + assert.equal(recoveredDiffInput?.fromTurnCount, 0); + assert.equal(recoveredDiffInput?.toTurnCount, 0); fs.rmSync(tempDir, { recursive: true, force: true }); }).pipe(Effect.provide(NodeServices.layer)), @@ -403,6 +496,51 @@ routing.layer("ProviderServiceLive routing", (it) => { }), ); + it.effect("recovers stale persisted sessions for checkpoint RPCs by resuming thread identity", () => + Effect.gen(function* () { + const provider = yield* ProviderService; + + const initial = yield* provider.startSession({ + provider: "codex", + cwd: "/tmp/project", + }); + yield* routing.codex.stopSession(initial.sessionId); + + routing.checkpoint.getCheckpointDiff.mockClear(); + routing.checkpoint.revertToCheckpoint.mockClear(); + routing.codex.startSession.mockClear(); + + yield* provider.getCheckpointDiff({ + sessionId: initial.sessionId, + fromTurnCount: 0, + toTurnCount: 0, + }); + yield* provider.revertToCheckpoint({ + sessionId: initial.sessionId, + turnCount: 0, + }); + + assert.deepEqual(routing.codex.startSession.mock.calls, [ + [ + { + provider: "codex", + resumeThreadId: initial.threadId, + }, + ], + ]); + assert.equal(routing.checkpoint.getCheckpointDiff.mock.calls.length, 1); + assert.equal(routing.checkpoint.revertToCheckpoint.mock.calls.length, 1); + + const diffInput = routing.checkpoint.getCheckpointDiff.mock.calls[0]?.[0]; + const revertInput = routing.checkpoint.revertToCheckpoint.mock.calls[0]?.[0]; + assert.equal(diffInput?.fromTurnCount, 0); + assert.equal(diffInput?.toTurnCount, 0); + assert.equal(diffInput?.sessionId === initial.sessionId, false); + assert.equal(revertInput?.sessionId, diffInput?.sessionId); + assert.equal(revertInput?.turnCount, 0); + }), + ); + it.effect("releases checkpoint session state when stopping all sessions", () => Effect.gen(function* () { const provider = yield* ProviderService; diff --git a/apps/server/src/provider/Layers/ProviderService.ts b/apps/server/src/provider/Layers/ProviderService.ts index a088c6a81d128a1c2c4f044f6ad8d9b7a8d3b0a6..f9a22e77395af920d4d441ecee5b67cd2a469e46 100644 --- a/apps/server/src/provider/Layers/ProviderService.ts +++ b/apps/server/src/provider/Layers/ProviderService.ts @@ -23,8 +23,9 @@ import { providerStopSessionInputSchema, type ProviderRuntimeEvent, type ProviderRuntimeTurnCompletedEvent, + type ProviderSession, } from "@t3tools/contracts"; -import { Effect, Layer, PubSub, Queue, Stream } from "effect"; +import { Effect, Layer, Option, PubSub, Queue, Ref, Stream } from "effect"; import { CheckpointService } from "../../checkpointing/Services/CheckpointService.ts"; import { ProviderValidationError } from "../Errors.ts"; @@ -97,25 +98,100 @@ const makeProviderService = (options?: ProviderServiceLiveOptions) => const runtimeEventQueue = yield* Queue.unbounded(); const runtimeEventPubSub = yield* PubSub.unbounded(); + const routedSessionAliasesRef = yield* Ref.make>(new Map()); + + const canonicalizeRuntimeEventSession = ( + event: ProviderRuntimeEvent, + ): Effect.Effect => + Ref.get(routedSessionAliasesRef).pipe( + Effect.map((aliases) => { + for (const [staleSessionId, liveSessionId] of aliases) { + if (liveSessionId === event.sessionId) { + return { + ...event, + sessionId: staleSessionId, + } satisfies ProviderRuntimeEvent; + } + } + return event; + }), + ); const publishRuntimeEvent = (event: ProviderRuntimeEvent): Effect.Effect => - Effect.sync(() => { - canonicalEventLogger?.write({ - observedAt: new Date().toISOString(), - event, - }); - }).pipe( - Effect.flatMap(() => PubSub.publish(runtimeEventPubSub, event)), + canonicalizeRuntimeEventSession(event).pipe( + Effect.tap((canonicalEvent) => + Effect.sync(() => { + canonicalEventLogger?.write({ + observedAt: new Date().toISOString(), + event: canonicalEvent, + }); + }), + ), + Effect.flatMap((canonicalEvent) => PubSub.publish(runtimeEventPubSub, canonicalEvent)), Effect.asVoid, ); + const upsertSessionBinding = ( + session: ProviderSession, + operation: string, + ) => + Effect.gen(function* () { + const threadId = session.threadId?.trim(); + if (!threadId) { + return yield* toValidationError( + operation, + `Provider '${session.provider}' returned a session without threadId. threadId is required for checkpoint initialization.`, + ); + } + + yield* directory.upsert({ + sessionId: session.sessionId, + provider: session.provider, + threadId, + }); + + return threadId; + }); + + const clearAliasKey = (staleSessionId: string) => + Ref.update(routedSessionAliasesRef, (current) => { + if (!current.has(staleSessionId)) { + return current; + } + const next = new Map(current); + next.delete(staleSessionId); + return next; + }); + + const clearAliasesReferencing = (sessionId: string) => + Ref.update(routedSessionAliasesRef, (current) => { + let changed = false; + const next = new Map(); + for (const [key, value] of current) { + if (key === sessionId || value === sessionId) { + changed = true; + continue; + } + next.set(key, value); + } + return changed ? next : current; + }); + + const setAlias = (staleSessionId: string, liveSessionId: string) => + Ref.update(routedSessionAliasesRef, (current) => { + const existing = current.get(staleSessionId); + if (existing === liveSessionId) { + return current; + } + const next = new Map(current); + next.set(staleSessionId, liveSessionId); + return next; + }); + const providers = yield* registry.listProviders(); - const adapters = yield* Effect.forEach(providers, (provider) => registry.getByProvider(provider)); - - const staleSessionIds = yield* directory.reconcileWithAdapters(adapters); - yield* Effect.forEach(staleSessionIds, (sessionId) => - checkpointService.releaseSession({ providerSessionId: sessionId }), - ).pipe(Effect.asVoid); + const adapters = yield* Effect.forEach(providers, (provider) => + registry.getByProvider(provider), + ); const onTurnCompleted = (event: ProviderRuntimeTurnCompletedEvent): Effect.Effect => checkpointService @@ -166,203 +242,361 @@ const makeProviderService = (options?: ProviderServiceLiveOptions) => ).pipe(Effect.forkScoped), ).pipe(Effect.asVoid); - const adapterForSession = (sessionId: string) => - directory - .getProvider(sessionId) - .pipe(Effect.flatMap((provider) => registry.getByProvider(provider))); - - const startSession: ProviderServiceShape["startSession"] = (rawInput) => - Effect.gen(function* () { - const parsed = providerSessionStartInputSchema.safeParse(rawInput); - if (!parsed.success) { - return yield* Effect.fail( - toValidationError( + const recoverSessionForThread = (input: { + readonly staleSessionId: string; + readonly provider: ProviderSession["provider"]; + readonly threadId: string; + readonly operation: string; + }) => + Effect.gen(function* () { + const adapter = yield* registry.getByProvider(input.provider); + const activeSessions = yield* adapter.listSessions(); + const existing = activeSessions.find( + (session) => session.threadId?.trim() === input.threadId, + ); + if (existing) { + const existingThreadId = yield* upsertSessionBinding( + existing, + `${input.operation}:upsertExistingSession`, + ); + yield* directory.upsert({ + sessionId: input.staleSessionId, + provider: existing.provider, + threadId: existingThreadId, + }); + if (existing.sessionId !== input.staleSessionId) { + yield* setAlias(input.staleSessionId, existing.sessionId); + } else { + yield* clearAliasKey(input.staleSessionId); + } + return { + adapter, + sessionId: existing.sessionId, + } as const; + } + + const resumed = yield* adapter.startSession({ + provider: input.provider, + resumeThreadId: input.threadId, + }); + if (resumed.provider !== adapter.provider) { + return yield* toValidationError( + input.operation, + `Adapter/provider mismatch while recovering stale session '${input.staleSessionId}'. Expected '${adapter.provider}', received '${resumed.provider}'.`, + ); + } + + const resumedThreadId = yield* upsertSessionBinding( + resumed, + `${input.operation}:upsertRecoveredSession`, + ); + if (resumedThreadId !== input.threadId) { + return yield* toValidationError( + input.operation, + `Recovered session thread '${resumedThreadId}' does not match expected thread '${input.threadId}'.`, + ); + } + + yield* directory.upsert({ + sessionId: input.staleSessionId, + provider: resumed.provider, + threadId: resumedThreadId, + }); + + const checkpointCwd = resumed.cwd ?? process.cwd(); + yield* checkpointService.initializeForSession({ + providerSessionId: resumed.sessionId, + cwd: checkpointCwd, + }); + + if (resumed.sessionId !== input.staleSessionId) { + yield* setAlias(input.staleSessionId, resumed.sessionId); + } else { + yield* clearAliasKey(input.staleSessionId); + } + + return { + adapter, + sessionId: resumed.sessionId, + } as const; + }); + + const resolveRoutableSession = (input: { + readonly sessionId: string; + readonly operation: string; + readonly allowRecovery: boolean; + }) => + Effect.gen(function* () { + const provider = yield* directory.getProvider(input.sessionId); + const adapter = yield* registry.getByProvider(provider); + + const hasRequestedSession = yield* adapter.hasSession(input.sessionId); + if (hasRequestedSession) { + yield* clearAliasKey(input.sessionId); + return { + adapter, + sessionId: input.sessionId, + isActive: true, + } as const; + } + + const alias = yield* Ref.get(routedSessionAliasesRef).pipe( + Effect.map((aliases) => aliases.get(input.sessionId)), + ); + if (alias) { + const aliasIsActive = yield* adapter.hasSession(alias); + if (aliasIsActive) { + return { + adapter, + sessionId: alias, + isActive: true, + } as const; + } + yield* clearAliasKey(input.sessionId); + } + + if (!input.allowRecovery) { + return { + adapter, + sessionId: input.sessionId, + isActive: false, + } as const; + } + + const threadIdOption = yield* directory.getThreadId(input.sessionId); + const threadId = Option.getOrUndefined(threadIdOption)?.trim(); + if (!threadId) { + return yield* toValidationError( + input.operation, + `Cannot recover stale session '${input.sessionId}' because no thread id is persisted.`, + ); + } + + const recovered = yield* recoverSessionForThread({ + staleSessionId: input.sessionId, + provider, + threadId, + operation: input.operation, + }); + + return { + adapter: recovered.adapter, + sessionId: recovered.sessionId, + isActive: true, + } as const; + }); + + const startSession: ProviderServiceShape["startSession"] = (rawInput) => + Effect.gen(function* () { + const parsed = providerSessionStartInputSchema.safeParse(rawInput); + if (!parsed.success) { + return yield* toValidationError( "ProviderService.startSession", parsed.error.message, parsed.error.cause, - ), - ); - } + ); + } - const input = parsed.data; - const adapter = yield* registry.getByProvider(input.provider); - const session = yield* adapter.startSession(input); + const input = parsed.data; + const adapter = yield* registry.getByProvider(input.provider); + const session = yield* adapter.startSession(input); - if (session.provider !== adapter.provider) { - return yield* Effect.fail( - toValidationError( + if (session.provider !== adapter.provider) { + return yield* toValidationError( "ProviderService.startSession", `Adapter/provider mismatch: requested '${adapter.provider}', received '${session.provider}'.`, - ), - ); - } + ); + } - const threadId = session.threadId?.trim(); - if (!threadId) { - return yield* Effect.fail( - toValidationError( - "ProviderService.startSession", - `Provider '${adapter.provider}' returned a session without threadId. threadId is required for checkpoint initialization.`, - ), - ); - } + yield* upsertSessionBinding(session, "ProviderService.startSession"); - yield* directory.upsert({ - sessionId: session.sessionId, - provider: session.provider, - threadId, - }); + const checkpointCwd = session.cwd ?? input.cwd ?? process.cwd(); + yield* checkpointService.initializeForSession({ + providerSessionId: session.sessionId, + cwd: checkpointCwd, + }); - const checkpointCwd = session.cwd ?? input.cwd ?? process.cwd(); - yield* checkpointService.initializeForSession({ - providerSessionId: session.sessionId, - cwd: checkpointCwd, + return session; }); - return session; - }); - - const sendTurn: ProviderServiceShape["sendTurn"] = (rawInput) => - Effect.gen(function* () { - const parsed = providerSendTurnInputSchema.safeParse(rawInput); - if (!parsed.success) { - return yield* Effect.fail( - toValidationError("ProviderService.sendTurn", parsed.error.message, parsed.error.cause), - ); - } + const sendTurn: ProviderServiceShape["sendTurn"] = (rawInput) => + Effect.gen(function* () { + const parsed = providerSendTurnInputSchema.safeParse(rawInput); + if (!parsed.success) { + return yield* toValidationError( + "ProviderService.sendTurn", + parsed.error.message, + parsed.error.cause, + ); + } - const input = parsed.data; - const adapter = yield* adapterForSession(input.sessionId); - return yield* adapter.sendTurn(input); - }); + const input = parsed.data; + const routed = yield* resolveRoutableSession({ + sessionId: input.sessionId, + operation: "ProviderService.sendTurn", + allowRecovery: true, + }); + return yield* routed.adapter.sendTurn({ + ...input, + sessionId: routed.sessionId, + }); + }); - const interruptTurn: ProviderServiceShape["interruptTurn"] = (rawInput) => - Effect.gen(function* () { - const parsed = providerInterruptTurnInputSchema.safeParse(rawInput); - if (!parsed.success) { - return yield* Effect.fail( - toValidationError( + const interruptTurn: ProviderServiceShape["interruptTurn"] = (rawInput) => + Effect.gen(function* () { + const parsed = providerInterruptTurnInputSchema.safeParse(rawInput); + if (!parsed.success) { + return yield* toValidationError( "ProviderService.interruptTurn", parsed.error.message, parsed.error.cause, - ), - ); - } + ); + } - const input = parsed.data; - const adapter = yield* adapterForSession(input.sessionId); - yield* adapter.interruptTurn(input.sessionId, input.turnId); - }); + const input = parsed.data; + const routed = yield* resolveRoutableSession({ + sessionId: input.sessionId, + operation: "ProviderService.interruptTurn", + allowRecovery: true, + }); + yield* routed.adapter.interruptTurn(routed.sessionId, input.turnId); + }); - const respondToRequest: ProviderServiceShape["respondToRequest"] = (rawInput) => - Effect.gen(function* () { - const parsed = providerRespondToRequestInputSchema.safeParse(rawInput); - if (!parsed.success) { - return yield* Effect.fail( - toValidationError( + const respondToRequest: ProviderServiceShape["respondToRequest"] = (rawInput) => + Effect.gen(function* () { + const parsed = providerRespondToRequestInputSchema.safeParse(rawInput); + if (!parsed.success) { + return yield* toValidationError( "ProviderService.respondToRequest", parsed.error.message, parsed.error.cause, - ), - ); - } + ); + } - const input = parsed.data; - const adapter = yield* adapterForSession(input.sessionId); - yield* adapter.respondToRequest(input.sessionId, input.requestId, input.decision); - }); + const input = parsed.data; + const routed = yield* resolveRoutableSession({ + sessionId: input.sessionId, + operation: "ProviderService.respondToRequest", + allowRecovery: true, + }); + yield* routed.adapter.respondToRequest(routed.sessionId, input.requestId, input.decision); + }); - const stopSession: ProviderServiceShape["stopSession"] = (rawInput) => - Effect.gen(function* () { - const parsed = providerStopSessionInputSchema.safeParse(rawInput); - if (!parsed.success) { - return yield* Effect.fail( - toValidationError( + const stopSession: ProviderServiceShape["stopSession"] = (rawInput) => + Effect.gen(function* () { + const parsed = providerStopSessionInputSchema.safeParse(rawInput); + if (!parsed.success) { + return yield* toValidationError( "ProviderService.stopSession", parsed.error.message, parsed.error.cause, - ), - ); - } + ); + } - 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); - }); + const input = parsed.data; + const routed = yield* resolveRoutableSession({ + sessionId: input.sessionId, + operation: "ProviderService.stopSession", + allowRecovery: false, + }); + if (routed.isActive) { + yield* routed.adapter.stopSession(routed.sessionId); + } + if (routed.sessionId !== input.sessionId) { + yield* checkpointService.releaseSession({ providerSessionId: routed.sessionId }); + yield* directory.remove(routed.sessionId); + yield* clearAliasesReferencing(routed.sessionId); + } + yield* checkpointService.releaseSession({ providerSessionId: input.sessionId }); + yield* directory.remove(input.sessionId); + yield* clearAliasesReferencing(input.sessionId); + }); - const listSessions: ProviderServiceShape["listSessions"] = () => - Effect.forEach(adapters, (adapter) => adapter.listSessions()).pipe( - Effect.map((sessionsByProvider) => sessionsByProvider.flatMap((sessions) => sessions)), - ); + const listSessions: ProviderServiceShape["listSessions"] = () => + Effect.forEach(adapters, (adapter) => adapter.listSessions()).pipe( + Effect.map((sessionsByProvider) => sessionsByProvider.flatMap((sessions) => sessions)), + ); - const listCheckpoints: ProviderServiceShape["listCheckpoints"] = (rawInput) => - Effect.gen(function* () { - const parsed = providerListCheckpointsInputSchema.safeParse(rawInput); - if (!parsed.success) { - return yield* Effect.fail( - toValidationError( + const listCheckpoints: ProviderServiceShape["listCheckpoints"] = (rawInput) => + Effect.gen(function* () { + const parsed = providerListCheckpointsInputSchema.safeParse(rawInput); + if (!parsed.success) { + return yield* toValidationError( "ProviderService.listCheckpoints", parsed.error.message, parsed.error.cause, - ), - ); - } + ); + } - const input = parsed.data; - yield* directory.getProvider(input.sessionId); - return yield* checkpointService.listCheckpoints(input); - }); + const input = parsed.data; + const routed = yield* resolveRoutableSession({ + sessionId: input.sessionId, + operation: "ProviderService.listCheckpoints", + allowRecovery: true, + }); + return yield* checkpointService.listCheckpoints({ + ...input, + sessionId: routed.sessionId, + }); + }); - const getCheckpointDiff: ProviderServiceShape["getCheckpointDiff"] = (rawInput) => - Effect.gen(function* () { - const parsed = providerGetCheckpointDiffInputSchema.safeParse(rawInput); - if (!parsed.success) { - return yield* Effect.fail( - toValidationError( + const getCheckpointDiff: ProviderServiceShape["getCheckpointDiff"] = (rawInput) => + Effect.gen(function* () { + const parsed = providerGetCheckpointDiffInputSchema.safeParse(rawInput); + if (!parsed.success) { + return yield* toValidationError( "ProviderService.getCheckpointDiff", parsed.error.message, parsed.error.cause, - ), - ); - } + ); + } - const input = parsed.data; - yield* directory.getProvider(input.sessionId); - return yield* checkpointService.getCheckpointDiff(input); - }); + const input = parsed.data; + const routed = yield* resolveRoutableSession({ + sessionId: input.sessionId, + operation: "ProviderService.getCheckpointDiff", + allowRecovery: true, + }); + return yield* checkpointService.getCheckpointDiff({ + ...input, + sessionId: routed.sessionId, + }); + }); - const revertToCheckpoint: ProviderServiceShape["revertToCheckpoint"] = (rawInput) => - Effect.gen(function* () { - const parsed = providerRevertToCheckpointInputSchema.safeParse(rawInput); - if (!parsed.success) { - return yield* Effect.fail( - toValidationError( + const revertToCheckpoint: ProviderServiceShape["revertToCheckpoint"] = (rawInput) => + Effect.gen(function* () { + const parsed = providerRevertToCheckpointInputSchema.safeParse(rawInput); + if (!parsed.success) { + return yield* toValidationError( "ProviderService.revertToCheckpoint", parsed.error.message, parsed.error.cause, - ), - ); - } + ); + } - const input = parsed.data; - yield* directory.getProvider(input.sessionId); - return yield* checkpointService.revertToCheckpoint(input); - }); + const input = parsed.data; + const routed = yield* resolveRoutableSession({ + sessionId: input.sessionId, + operation: "ProviderService.revertToCheckpoint", + allowRecovery: true, + }); + return yield* checkpointService.revertToCheckpoint({ + ...input, + sessionId: routed.sessionId, + }); + }); - const stopAll: ProviderServiceShape["stopAll"] = () => - Effect.gen(function* () { - const sessionIds = yield* directory.listSessionIds(); - 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 stopAll: ProviderServiceShape["stopAll"] = () => + Effect.gen(function* () { + const sessionIds = yield* directory.listSessionIds(); + yield* Effect.forEach(adapters, (adapter) => adapter.stopAll()).pipe(Effect.asVoid); + // Keep persisted session bindings so stale sessions can be resumed after + // process restart via resumeThreadId. + yield* Effect.forEach(sessionIds, (sessionId) => + checkpointService.releaseSession({ providerSessionId: sessionId }), + ).pipe(Effect.asVoid); + yield* Ref.set(routedSessionAliasesRef, new Map()); + }); return { startSession, diff --git a/apps/server/src/provider/Layers/ProviderSessionDirectory.test.ts b/apps/server/src/provider/Layers/ProviderSessionDirectory.test.ts index db4dec63d99992b540bb354a5cd3e44ce43f347f..bcdc5cad250436bcc499c459af62f4f60084976b 100644 --- a/apps/server/src/provider/Layers/ProviderSessionDirectory.test.ts +++ b/apps/server/src/provider/Layers/ProviderSessionDirectory.test.ts @@ -5,13 +5,12 @@ import path from "node:path"; import { NodeServices } from "@effect/platform-node"; import { it, assert } from "@effect/vitest"; import { assertFailure, assertSome } from "@effect/vitest/utils"; -import { Effect, Layer, Stream } from "effect"; +import { Effect, Layer } from "effect"; import * as SqlClient from "effect/unstable/sql/SqlClient"; import { makeSqlitePersistenceLive, SqlitePersistenceMemory } from "../../persistence/Layers/Sqlite.ts"; import { ProviderSessionRepositoryLive } from "../../persistence/Layers/ProviderSessions.ts"; -import { ProviderSessionNotFoundError, ProviderValidationError, type ProviderAdapterError } from "../Errors.ts"; -import type { ProviderAdapterShape } from "../Services/ProviderAdapter.ts"; +import { ProviderSessionNotFoundError, ProviderValidationError } from "../Errors.ts"; import { ProviderSessionDirectory } from "../Services/ProviderSessionDirectory.ts"; import { ProviderSessionDirectoryLive } from "./ProviderSessionDirectory.ts"; @@ -22,24 +21,6 @@ function makeDirectoryLayer(persistenceLayer: Layer.Layer): ProviderAdapterShape { - const unsupported = () => Effect.die(new Error("unsupported test operation")) as never; - return { - provider: "codex", - startSession: () => unsupported(), - sendTurn: () => unsupported(), - interruptTurn: () => unsupported(), - respondToRequest: () => unsupported(), - stopSession: () => unsupported(), - listSessions: () => Effect.succeed([]), - hasSession: (sessionId) => Effect.succeed(activeSessionIds.has(sessionId)), - readThread: () => unsupported(), - rollbackThread: () => unsupported(), - stopAll: () => Effect.void, - streamEvents: Stream.empty, - }; -} - const layer = it.layer(makeDirectoryLayer(SqlitePersistenceMemory)); layer("ProviderSessionDirectoryLive", (it) => { @@ -118,28 +99,4 @@ layer("ProviderSessionDirectoryLive", (it) => { fs.rmSync(tempDir, { recursive: true, force: true }); }).pipe(Effect.provide(NodeServices.layer))); - - it("reconciles stale sessions and keeps active mappings", () => - Effect.gen(function* () { - const directory = yield* ProviderSessionDirectory; - - yield* directory.upsert({ - sessionId: "sess-active", - provider: "codex", - threadId: "thread-active", - }); - yield* directory.upsert({ - sessionId: "sess-stale", - provider: "codex", - threadId: "thread-stale", - }); - - const pruned = yield* directory.reconcileWithAdapters([ - makeAdapter(new Set(["sess-active"])), - ]); - assert.deepEqual(pruned, ["sess-stale"]); - - const remaining = yield* directory.listSessionIds(); - assert.deepEqual(remaining, ["sess-active"]); - })); }); diff --git a/apps/server/src/provider/Layers/ProviderSessionDirectory.ts b/apps/server/src/provider/Layers/ProviderSessionDirectory.ts index 98de7eaaca4bb4693ea467ae7b6d4ab01115ce69..eae8a444746ce7e84da47a26177a094cd653eb89 100644 --- a/apps/server/src/provider/Layers/ProviderSessionDirectory.ts +++ b/apps/server/src/provider/Layers/ProviderSessionDirectory.ts @@ -109,45 +109,12 @@ const makeProviderSessionDirectory = Effect.gen(function* () { const listSessionIds: ProviderSessionDirectoryShape["listSessionIds"] = () => listBindings().pipe(Effect.map((bindings) => bindings.map((binding) => binding.sessionId))); - const reconcileWithAdapters: ProviderSessionDirectoryShape["reconcileWithAdapters"] = ( - adapters, - ) => - Effect.gen(function* () { - const byProvider = new Map(adapters.map((adapter) => [adapter.provider, adapter])); - const bindings = yield* listBindings(); - const staleSessionIds: string[] = []; - - for (const binding of bindings) { - const adapter = byProvider.get(binding.provider); - if (!adapter) { - staleSessionIds.push(binding.sessionId); - continue; - } - - const hasSession = yield* adapter.hasSession(binding.sessionId); - if (!hasSession) { - staleSessionIds.push(binding.sessionId); - } - } - - if (staleSessionIds.length === 0) { - return [] as ReadonlyArray; - } - - yield* Effect.forEach(staleSessionIds, (sessionId) => persistDelete(sessionId)).pipe( - Effect.asVoid, - ); - - return staleSessionIds as ReadonlyArray; - }); - return { upsert, getProvider, getThreadId, remove, listSessionIds, - reconcileWithAdapters, } satisfies ProviderSessionDirectoryShape; }); diff --git a/apps/server/src/provider/Services/ProviderAdapterRegistry.ts b/apps/server/src/provider/Services/ProviderAdapterRegistry.ts index 23066db7911a3164937c045c9faf413b06beb21f..420086a7c15b129c05315a5e29cdf3e32aa4b79a 100644 --- a/apps/server/src/provider/Services/ProviderAdapterRegistry.ts +++ b/apps/server/src/provider/Services/ProviderAdapterRegistry.ts @@ -35,3 +35,5 @@ export class ProviderAdapterRegistry extends ServiceMap.Service< ProviderAdapterRegistry, ProviderAdapterRegistryShape >()("provider/ProviderAdapterRegistry") {} + +// Dummy comment for workflow testing. diff --git a/apps/server/src/provider/Services/ProviderService.ts b/apps/server/src/provider/Services/ProviderService.ts index 3d5798b93bf2de3eb8c36c1646edf9f9f8219ed0..78fc1adf48c73e6f41c2705856b0d38859f463b1 100644 --- a/apps/server/src/provider/Services/ProviderService.ts +++ b/apps/server/src/provider/Services/ProviderService.ts @@ -95,7 +95,10 @@ export interface ProviderServiceShape { ) => Effect.Effect; /** - * Stop all provider sessions. + * Stop all active provider sessions. + * + * Persisted session bindings remain so stale sessions can be resumed after + * process restart. */ readonly stopAll: () => Effect.Effect; diff --git a/apps/server/src/provider/Services/ProviderSessionDirectory.ts b/apps/server/src/provider/Services/ProviderSessionDirectory.ts index 41009f45465495996645a76b9526380d7fc1cd1b..5563663e71ad50b43b8002a394c4edf2b1b8d6aa 100644 --- a/apps/server/src/provider/Services/ProviderSessionDirectory.ts +++ b/apps/server/src/provider/Services/ProviderSessionDirectory.ts @@ -12,12 +12,10 @@ import { Option, ServiceMap } from "effect"; import type { Effect } from "effect"; import type { - ProviderAdapterError, ProviderSessionDirectoryPersistenceError, ProviderSessionNotFoundError, ProviderValidationError, } from "../Errors.ts"; -import type { ProviderAdapterShape } from "./ProviderAdapter.ts"; export interface ProviderSessionBinding { readonly sessionId: string; @@ -70,17 +68,6 @@ export interface ProviderSessionDirectoryShape { ProviderSessionDirectoryPersistenceError >; - /** - * Remove stale persisted sessions that are no longer active in adapters. - * - * Returns pruned session ids. - */ - readonly reconcileWithAdapters: ( - adapters: ReadonlyArray>, - ) => Effect.Effect< - ReadonlyArray, - ProviderSessionDirectoryPersistenceError | ProviderAdapterError - >; } /** diff --git a/apps/server/tsconfig.json b/apps/server/tsconfig.json index 122fd028672c33e174ef76ef90764192910e1617..586f8c560c471b0c3f2f7b8b0bae3d295c725d07 100644 --- a/apps/server/tsconfig.json +++ b/apps/server/tsconfig.json @@ -5,7 +5,12 @@ "types": ["node"], "lib": ["ES2023", "esnext.disposable"], "noEmit": true, - "allowImportingTsExtensions": true + "allowImportingTsExtensions": true, + "plugins": [ + { + "name": "@effect/language-service" + } + ] }, "include": ["src", "integration", "tsdown.config.ts", "scripts"] } diff --git a/bun.lock b/bun.lock index f0be6120c43a2143ea1ea37288b388815f1bf298..6fc8349814a234c37a0ca23bd6b1b53bbb28cdad 100644 --- a/bun.lock +++ b/bun.lock @@ -35,6 +35,7 @@ }, "dependencies": { "@effect/platform-node": "4.0.0-beta.10", + "@effect/sql-sqlite-bun": "4.0.0-beta.10", "@pierre/diffs": "^1.1.0-beta.16", "effect": "4.0.0-beta.10", "node-pty": "^1.1.0", @@ -42,6 +43,7 @@ "ws": "^8.18.0", }, "devDependencies": { + "@effect/language-service": "0.75.1", "@effect/vitest": "4.0.0-beta.10", "@t3tools/contracts": "workspace:*", "@t3tools/web": "workspace:*", @@ -96,6 +98,7 @@ "zod": "^3.24.1", }, "devDependencies": { + "@effect/language-service": "0.75.1", "tsdown": "^0.20.3", "typescript": "^5.7.3", }, @@ -150,10 +153,14 @@ "@base-ui/utils": ["@base-ui/utils@0.2.5", "", { "dependencies": { "@babel/runtime": "^7.28.6", "@floating-ui/utils": "^0.2.10", "reselect": "^5.1.1", "use-sync-external-store": "^1.6.0" }, "peerDependencies": { "@types/react": "^17 || ^18 || ^19", "react": "^17 || ^18 || ^19", "react-dom": "^17 || ^18 || ^19" }, "optionalPeers": ["@types/react"] }, "sha512-oYC7w0gp76RI5MxprlGLV0wze0SErZaRl3AAkeP3OnNB/UBMb6RqNf6ZSIlxOc9Qp68Ab3C2VOcJQyRs7Xc7Vw=="], + "@effect/language-service": ["@effect/language-service@0.75.1", "", { "bin": { "effect-language-service": "cli.js" } }, "sha512-g9xD2tAQgRFpYC2YgpZq02VeSL5fBbFJ0B/g1o+14NuNmwtaYJc7SjiLWAA9eyhJHosNrn6h1Ye+Kx6j5mN0AA=="], + "@effect/platform-node": ["@effect/platform-node@4.0.0-beta.10", "", { "dependencies": { "@effect/platform-node-shared": "^4.0.0-beta.10", "mime": "^4.1.0", "undici": "^7.20.0" }, "peerDependencies": { "effect": "^4.0.0-beta.10", "ioredis": "^5.7.0" } }, "sha512-ikZ/uyos5okYAHE+wWOwZkHgriFUTY3aY8Kdqbj/ro8xw0z9YXWN4fbyQAZUWkBcieDKS36N+QfHRMPG4fEWRA=="], "@effect/platform-node-shared": ["@effect/platform-node-shared@4.0.0-beta.10", "", { "dependencies": { "@types/ws": "^8.18.1", "ws": "^8.19.0" }, "peerDependencies": { "effect": "^4.0.0-beta.10" } }, "sha512-gBSeOG2bfchXFdgdi5oipHKynOzSd7BlzTy5QcjFOoCUTtLP9igqohPB9hTkV3a6seWE2JJqsgsQ79VuLb7fAg=="], + "@effect/sql-sqlite-bun": ["@effect/sql-sqlite-bun@4.0.0-beta.10", "", { "peerDependencies": { "effect": "^4.0.0-beta.10" } }, "sha512-LY1jNu2uqGYsJCVjVU73nzcZ+kAYBDsG7O8x5ddmkIXj10RmnIhjYMStWBa0NZOFg4Hsl+CadxtsDy3edaOVeA=="], + "@effect/vitest": ["@effect/vitest@4.0.0-beta.10", "", { "peerDependencies": { "effect": "^4.0.0-beta.10", "vitest": "^3.0.0" } }, "sha512-WmE8vCNw+eDoAq9CwwymrAdIBw1f+8hDP3AUF7jyC7/Cj/kyAYOFwkbtKmDXIFaTlf7G/2fmulEXeCO5W8j0kQ=="], "@electron/get": ["@electron/get@2.0.3", "", { "dependencies": { "debug": "^4.1.1", "env-paths": "^2.2.0", "fs-extra": "^8.1.0", "got": "^11.8.5", "progress": "^2.0.3", "semver": "^6.2.0", "sumchecker": "^3.0.1" }, "optionalDependencies": { "global-agent": "^3.0.0" } }, "sha512-Qkzpg2s9GnVV2I2BjRksUi43U5e6+zaQMcjoJy0C+C5oxaKl+fmckGDQFtRpZpZV0NQekuZZ+tGz7EA9TVnQtQ=="], diff --git a/packages/contracts/package.json b/packages/contracts/package.json index c4843f4d79d6c841117479569d397ee48c4feceb..fe90f1d888cac61258af038b540e78c3f89e82e9 100644 --- a/packages/contracts/package.json +++ b/packages/contracts/package.json @@ -19,6 +19,7 @@ "scripts": { "dev": "tsdown src/index.ts --format esm,cjs --dts --watch --clean", "build": "tsdown src/index.ts --format esm,cjs --dts --clean", + "prepare": "effect-language-service patch", "typecheck": "tsc --noEmit", "test": "vitest run" }, @@ -27,6 +28,7 @@ "zod": "^3.24.1" }, "devDependencies": { + "@effect/language-service": "0.75.1", "tsdown": "^0.20.3", "typescript": "^5.7.3" } diff --git a/packages/contracts/tsconfig.json b/packages/contracts/tsconfig.json index 564a5990051fdb8cee49fbbaf99734c1a23cf60a..a7d84f36f53736c33ed6b4d1d36fccbcb02efb7d 100644 --- a/packages/contracts/tsconfig.json +++ b/packages/contracts/tsconfig.json @@ -1,4 +1,11 @@ { "extends": "../../tsconfig.base.json", + "compilerOptions": { + "plugins": [ + { + "name": "@effect/language-service" + } + ] + }, "include": ["src"] } -- 2.54.0