From f2e3f81f40ae84bb8af25225fda4a503dd21d4fa Mon Sep 17 00:00:00 2001 From: Tom Kaczocha Date: Thu, 15 Oct 2026 20:44:26 +1000 Subject: [PATCH 1/5] test(factory): sessions and paths adapter cases Ports softfac's find_project and status cases from src/t3.rs, adds launch and thread cases, and a contract test over the real orchestration layers with the Claude replay transcript. Red: paths.ts and sessions.ts do not exist yet, so none of the three files compile. Issue: #5 --- .../t3/fixtures/t3_thread_snapshot.json | 153 +++++ apps/server/src/factory/t3/paths.test.ts | 30 + .../src/factory/t3/sessions.contract.test.ts | 286 ++++++++++ apps/server/src/factory/t3/sessions.test.ts | 527 ++++++++++++++++++ 4 files changed, 996 insertions(+) create mode 100644 apps/server/src/factory/t3/fixtures/t3_thread_snapshot.json create mode 100644 apps/server/src/factory/t3/paths.test.ts create mode 100644 apps/server/src/factory/t3/sessions.contract.test.ts create mode 100644 apps/server/src/factory/t3/sessions.test.ts diff --git a/apps/server/src/factory/t3/fixtures/t3_thread_snapshot.json b/apps/server/src/factory/t3/fixtures/t3_thread_snapshot.json new file mode 100644 index 000000000000..3ec1c0e7ebca --- /dev/null +++ b/apps/server/src/factory/t3/fixtures/t3_thread_snapshot.json @@ -0,0 +1,153 @@ +{ + "snapshotSequence": 38, + "projection": { + "thread": { + "createdBy": "user", + "creationSource": "web", + "id": "thread:project:e87a2ca6-3572-4958-bb1d-d91ca88c9070:79e52860-a72a-4aa1-8042-6ed913b9a511", + "projectId": "e87a2ca6-3572-4958-bb1d-d91ca88c9070", + "title": "factory spike", + "providerInstanceId": "claudeAgent", + "modelSelection": { + "instanceId": "claudeAgent", + "model": "claude-sonnet-5-5" + }, + "runtimeMode": "approval-required", + "interactionMode": "default", + "branch": "t3code/618ff943", + "worktreePath": "/tmp/t3-spike-home/worktrees/t3-spike-repo/t3code-618ff943", + "activeProviderThreadId": "provider-thread:provider:claudeAgent:native-thread:pending%3Arun%3Athread%3Athread%253Aproject%253Ae87a2ca6-3572-4958-bb1d-d91ca88c9070%253A79e52860-a72a-4aa1-8042-6ed913b9a511%3Aordinal%3A1", + "lineage": { + "parentThreadId": null, + "relationshipToParent": null, + "rootThreadId": "thread:project:e87a2ca6-3572-4958-bb1d-d91ca88c9070:79e52860-a72a-4aa1-8042-6ed913b9a511" + }, + "forkedFrom": null, + "createdAt": "2026-10-04T09:36:15.108Z", + "updatedAt": "2026-10-04T09:36:19.470Z", + "archivedAt": null, + "settledOverride": null, + "settledAt": null, + "snoozedUntil": null, + "snoozedAt": null, + "lastVisitedAt": null, + "deletedAt": null + }, + "runs": [ + { + "id": "run:thread:thread%3Aproject%3Ae87a2ca6-3572-4958-bb1d-d91ca88c9070%3A79e52860-a72a-4aa1-8042-6ed913b9a511:ordinal:1", + "threadId": "thread:project:e87a2ca6-3572-4958-bb1d-d91ca88c9070:79e52860-a72a-4aa1-8042-6ed913b9a511", + "ordinal": 1, + "providerInstanceId": "claudeAgent", + "modelSelection": { + "instanceId": "claudeAgent", + "model": "claude-sonnet-5-5" + }, + "providerThreadId": "provider-thread:provider:claudeAgent:native-thread:pending%3Arun%3Athread%3Athread%253Aproject%253Ae87a2ca6-3572-4958-bb1d-d91ca88c9070%253A79e52860-a72a-4aa1-8042-6ed913b9a511%3Aordinal%3A1", + "userMessageId": "message:thread:thread%3Aproject%3Ae87a2ca6-3572-4958-bb1d-d91ca88c9070%3A79e52860-a72a-4aa1-8042-6ed913b9a511:ordinal:1:e61e49ab-97fb-4e08-a54f-8c5409ae7982", + "rootNodeId": "node:run:run%3Athread%3Athread%253Aproject%253Ae87a2ca6-3572-4958-bb1d-d91ca88c9070%253A79e52860-a72a-4aa1-8042-6ed913b9a511%3Aordinal%3A1:root", + "activeAttemptId": "run-attempt:run:run%3Athread%3Athread%253Aproject%253Ae87a2ca6-3572-4958-bb1d-d91ca88c9070%253A79e52860-a72a-4aa1-8042-6ed913b9a511%3Aordinal%3A1:attempt:1", + "status": "completed", + "queuePosition": null, + "requestedAt": "2026-10-04T09:36:15.113Z", + "startedAt": "2026-10-04T09:36:15.242Z", + "completedAt": "2026-10-04T09:36:19.470Z", + "checkpointId": "checkpoint:scope:checkpoint-scope%3Athread%3Athread%253Aproject%253Ae87a2ca6-3572-4958-bb1d-d91ca88c9070%253A79e52860-a72a-4aa1-8042-6ed913b9a511%3Aname%3Aroot:name:1", + "contextHandoffId": null, + "workspacePreparation": { + "type": "worktree", + "baseRef": "main", + "startFromOrigin": false + } + } + ], + "attempts": [ + { + "id": "run-attempt:run:run%3Athread%3Athread%253Aproject%253Ae87a2ca6-3572-4958-bb1d-d91ca88c9070%253A79e52860-a72a-4aa1-8042-6ed913b9a511%3Aordinal%3A1:attempt:1", + "nativeThreadId": "913ad7b4-ea27-4a7c-957e-de45ae9e96ff", + "runId": "run:thread:thread%3Aproject%3Ae87a2ca6-3572-4958-bb1d-d91ca88c9070%3A79e52860-a72a-4aa1-8042-6ed913b9a511:ordinal:1", + "attemptOrdinal": 1, + "rootNodeId": "node:run:run%3Athread%3Athread%253Aproject%253Ae87a2ca6-3572-4958-bb1d-d91ca88c9070%253A79e52860-a72a-4aa1-8042-6ed913b9a511%3Aordinal%3A1:root", + "providerInstanceId": "claudeAgent", + "providerThreadId": "provider-thread:provider:claudeAgent:native-thread:pending%3Arun%3Athread%3Athread%253Aproject%253Ae87a2ca6-3572-4958-bb1d-d91ca88c9070%253A79e52860-a72a-4aa1-8042-6ed913b9a511%3Aordinal%3A1", + "providerTurnId": null, + "reason": "initial", + "status": "completed", + "startedAt": "2026-10-04T09:36:15.242Z", + "completedAt": "2026-10-04T09:36:19.466Z" + } + ], + "nodes": [ + { + "id": "node:run:run%3Athread%3Athread%253Aproject%253Ae87a2ca6-3572-4958-bb1d-d91ca88c9070%253A79e52860-a72a-4aa1-8042-6ed913b9a511%3Aordinal%3A1:root", + "threadId": "thread:project:e87a2ca6-3572-4958-bb1d-d91ca88c9070:79e52860-a72a-4aa1-8042-6ed913b9a511", + "runId": "run:thread:thread%3Aproject%3Ae87a2ca6-3572-4958-bb1d-d91ca88c9070%3A79e52860-a72a-4aa1-8042-6ed913b9a511:ordinal:1", + "parentNodeId": null, + "rootNodeId": "node:run:run%3Athread%3Athread%253Aproject%253Ae87a2ca6-3572-4958-bb1d-d91ca88c9070%253A79e52860-a72a-4aa1-8042-6ed913b9a511%3Aordinal%3A1:root", + "kind": "root_turn", + "status": "completed", + "countsForRun": true, + "providerThreadId": "provider-thread:provider:claudeAgent:native-thread:pending%3Arun%3Athread%3Athread%253Aproject%253Ae87a2ca6-3572-4958-bb1d-d91ca88c9070%253A79e52860-a72a-4aa1-8042-6ed913b9a511%3Aordinal%3A1", + "providerTurnId": null, + "nativeItemRef": null, + "runtimeRequestId": null, + "checkpointScopeId": "checkpoint-scope:thread:thread%3Aproject%3Ae87a2ca6-3572-4958-bb1d-d91ca88c9070%3A79e52860-a72a-4aa1-8042-6ed913b9a511:name:root", + "startedAt": "2026-10-04T09:36:15.242Z", + "completedAt": "2026-10-04T09:36:19.470Z" + }, + { + "id": "node:provider:claudeAgent:native-item:32c394dd-48e9-4e54-9fa5-26c8c2789600", + "threadId": "thread:project:e87a2ca6-3572-4958-bb1d-d91ca88c9070:79e52860-a72a-4aa1-8042-6ed913b9a511", + "runId": "run:thread:thread%3Aproject%3Ae87a2ca6-3572-4958-bb1d-d91ca88c9070%3A79e52860-a72a-4aa1-8042-6ed913b9a511:ordinal:1", + "parentNodeId": "node:run:run%3Athread%3Athread%253Aproject%253Ae87a2ca6-3572-4958-bb1d-d91ca88c9070%253A79e52860-a72a-4aa1-8042-6ed913b9a511%3Aordinal%3A1:root", + "rootNodeId": "node:run:run%3Athread%3Athread%253Aproject%253Ae87a2ca6-3572-4958-bb1d-d91ca88c9070%253A79e52860-a72a-4aa1-8042-6ed913b9a511%3Aordinal%3A1:root", + "kind": "assistant_message", + "status": "completed", + "countsForRun": false, + "providerThreadId": "provider-thread:provider:claudeAgent:native-thread:pending%3Arun%3Athread%3Athread%253Aproject%253Ae87a2ca6-3572-4958-bb1d-d91ca88c9070%253A79e52860-a72a-4aa1-8042-6ed913b9a511%3Aordinal%3A1", + "providerTurnId": "provider-turn:provider:claudeAgent:native-turn:turn%3Arun-attempt%3Arun%3Arun%253Athread%253Athread%25253Aproject%25253Ae87a2ca6-3572-4958-bb1d-d91ca88c9070%25253A79e52860-a72a-4aa1-8042-6ed913b9a511%253Aordinal%253A1%3Aattempt%3A1", + "nativeItemRef": { + "driver": "claudeAgent", + "nativeId": "32c394dd-48e9-4e54-9fa5-26c8c2789600", + "strength": "strong" + }, + "runtimeRequestId": null, + "checkpointScopeId": null, + "startedAt": "2026-10-04T09:36:19.429Z", + "completedAt": "2026-10-04T09:36:19.429Z" + } + ], + "messages": [ + { + "createdBy": "user", + "creationSource": "web", + "id": "message:thread:thread%3Aproject%3Ae87a2ca6-3572-4958-bb1d-d91ca88c9070%3A79e52860-a72a-4aa1-8042-6ed913b9a511:ordinal:1:e61e49ab-97fb-4e08-a54f-8c5409ae7982", + "threadId": "thread:project:e87a2ca6-3572-4958-bb1d-d91ca88c9070:79e52860-a72a-4aa1-8042-6ed913b9a511", + "runId": "run:thread:thread%3Aproject%3Ae87a2ca6-3572-4958-bb1d-d91ca88c9070%3A79e52860-a72a-4aa1-8042-6ed913b9a511:ordinal:1", + "nodeId": "node:run:run%3Athread%3Athread%253Aproject%253Ae87a2ca6-3572-4958-bb1d-d91ca88c9070%253A79e52860-a72a-4aa1-8042-6ed913b9a511%3Aordinal%3A1:root", + "role": "user", + "text": "Reply with the single word READY. Do not use any tools.", + "attachments": [], + "streaming": false, + "createdAt": "2026-10-04T09:36:15.113Z", + "updatedAt": "2026-10-04T09:36:15.113Z" + }, + { + "createdBy": "agent", + "creationSource": "provider", + "id": "message:provider:claudeAgent:native-item:32c394dd-48e9-4e54-9fa5-26c8c2789600", + "threadId": "thread:project:e87a2ca6-3572-4958-bb1d-d91ca88c9070:79e52860-a72a-4aa1-8042-6ed913b9a511", + "runId": "run:thread:thread%3Aproject%3Ae87a2ca6-3572-4958-bb1d-d91ca88c9070%3A79e52860-a72a-4aa1-8042-6ed913b9a511:ordinal:1", + "nodeId": "node:provider:claudeAgent:native-item:32c394dd-48e9-4e54-9fa5-26c8c2789600", + "role": "assistant", + "text": "READY", + "attachments": [], + "streaming": false, + "createdAt": "2026-10-04T09:36:19.429Z", + "updatedAt": "2026-10-04T09:36:19.429Z" + } + ], + "plans": [], + "runtimeRequests": [] + } +} diff --git a/apps/server/src/factory/t3/paths.test.ts b/apps/server/src/factory/t3/paths.test.ts new file mode 100644 index 000000000000..39193a787e69 --- /dev/null +++ b/apps/server/src/factory/t3/paths.test.ts @@ -0,0 +1,30 @@ +import { assert, it } from "@effect/vitest"; +import * as NodeServices from "@effect/platform-node/NodeServices"; +import * as Effect from "effect/Effect"; +import * as Layer from "effect/Layer"; + +import * as ServerConfig from "../../config.ts"; +import * as Paths from "./paths.ts"; + +it.effect( + "takes the base dir and the worktrees dir from ServerConfig and puts factory data under /factory", + () => + Effect.gen(function* () { + const paths = yield* Paths.FactoryPaths; + assert.strictEqual(paths.baseDir, "/srv/t3fac"); + assert.strictEqual(paths.worktreesDir, "/srv/t3fac/worktrees"); + assert.strictEqual(paths.factoryDir, "/srv/t3fac/factory"); + }).pipe( + Effect.provide( + Paths.layer.pipe( + Layer.provide( + Layer.mock(ServerConfig.ServerConfig)({ + baseDir: "/srv/t3fac", + worktreesDir: "/srv/t3fac/worktrees", + }), + ), + Layer.provide(NodeServices.layer), + ), + ), + ), +); diff --git a/apps/server/src/factory/t3/sessions.contract.test.ts b/apps/server/src/factory/t3/sessions.contract.test.ts new file mode 100644 index 000000000000..f33e33f65dae --- /dev/null +++ b/apps/server/src/factory/t3/sessions.contract.test.ts @@ -0,0 +1,286 @@ +/** + * Runs the adapter against T3's real orchestration layers: the orchestrator, + * its stores, the effect worker, the launch service and a real git worktree, + * with the Claude replay transcript from provider-testing standing in for the + * provider. It answers one question on every upstream sync: does the adapter + * still work against this upstream? + */ +import { assert, it } from "@effect/vitest"; +import * as NodeServices from "@effect/platform-node/NodeServices"; +import { EventId, ProjectId, ThreadId, type Project } from "@t3tools/contracts"; +import * as IdAllocator from "@t3tools/provider-core/server/IdAllocator"; +import { readProviderReplayTranscript } from "@t3tools/provider-testing/replayTranscript"; +import * as Effect from "effect/Effect"; +import * as FileSystem from "effect/FileSystem"; +import * as Layer from "effect/Layer"; +import * as Option from "effect/Option"; +import * as Path from "effect/Path"; +import * as Stream from "effect/Stream"; + +import * as GitWorkflow from "../../git/GitWorkflowService.ts"; +import { ClaudeOrchestratorReplayHarness } from "../../orchestration-v2/Adapters/ClaudeAdapterV2.testkit.ts"; +import * as CommandReceiptStore from "../../orchestration-v2/CommandReceiptStore.ts"; +import * as EventSink from "../../orchestration-v2/EventSink.ts"; +import * as ProjectionStore from "../../orchestration-v2/ProjectionStore.ts"; +import * as ProjectStore from "../../orchestration-v2/ProjectStore.ts"; +import * as ProviderReplayHarness from "../../orchestration-v2/testkit/ProviderReplayHarness.ts"; +import * as ThreadLaunch from "../../orchestration-v2/ThreadLaunchService.ts"; +import * as ThreadManagement from "../../orchestration-v2/ThreadManagementService.ts"; +import * as SqlitePersistence from "../../persistence/Sqlite.ts"; +import * as ManagedProjectFolders from "../../project/ManagedProjectFolders.ts"; +import * as ProjectCloneTracker from "../../project/ProjectCloneTracker.ts"; +import * as ProjectService from "../../project/ProjectService.ts"; +import * as ProjectSetupScriptRunner from "../../project/ProjectSetupScriptRunner.ts"; +import * as WorktreeSetupTracker from "../../project/WorktreeSetupTracker.ts"; +import * as ProviderRegistryMock from "../../provider/testUtils/providerRegistryMock.ts"; +import * as ServerSettings from "../../serverSettings.ts"; +import * as TerminalManager from "../../terminal/Manager.ts"; +import * as TextGeneration from "../../textGeneration/TextGeneration.ts"; +import * as GitVcsDriver from "../../vcs/GitVcsDriver.ts"; +import * as VcsProcess from "../../vcs/VcsProcess.ts"; +import * as Sessions from "./sessions.ts"; + +const SIMPLE_TRANSCRIPT = new URL( + "../../orchestration-v2/testkit/fixtures/simple/claude_transcript.ndjson", + import.meta.url, +); +/** The prompt the transcript was recorded with; the reply is "fixture simple ok". */ +const SIMPLE_PROMPT = "Respond with the following text: fixture simple ok"; + +const projectId = ProjectId.make("project:factory-sessions-contract"); +const createdAt = "2026-10-15T00:00:00.000Z"; + +const TERMINAL_RUN_STATUSES: ReadonlySet = new Set([ + "completed", + "interrupted", + "failed", + "cancelled", + "rolled_back", +]); + +const git = (cwd: string, args: ReadonlyArray) => + Effect.gen(function* () { + const driver = yield* GitVcsDriver.GitVcsDriver; + const result = yield* driver.execute({ + operation: "factory.sessions.contract.git", + cwd, + args, + timeoutMs: 10_000, + }); + return result.stdout.trim(); + }); + +const initRepoWithCommit = (cwd: string) => + Effect.gen(function* () { + const fs = yield* FileSystem.FileSystem; + const path = yield* Path.Path; + const driver = yield* GitVcsDriver.GitVcsDriver; + yield* driver.initRepo({ cwd }); + yield* git(cwd, ["config", "user.email", "factory@test.invalid"]); + yield* git(cwd, ["config", "user.name", "Factory"]); + yield* fs.writeFileString(path.join(cwd, "README.md"), "# factory contract\n"); + yield* git(cwd, ["add", "README.md"]); + yield* git(cwd, ["commit", "-m", "initial commit"]); + return yield* git(cwd, ["branch", "--show-current"]); + }); + +/** The real orchestration runtime plus the launch service over a real repo. */ +const layerRuntime = (input: { readonly repo: string; readonly worktreesDir: string }) => + Layer.unwrap( + Effect.gen(function* () { + const transcript = yield* ClaudeOrchestratorReplayHarness.decodeTranscript( + yield* readProviderReplayTranscript(SIMPLE_TRANSCRIPT), + ); + const project: Project = { + id: projectId, + title: "Factory contract", + workspaceRoot: input.repo, + repositoryIdentity: null, + faviconPath: null, + defaultModelSelection: null, + defaultThreadEnvMode: null, + scripts: [], + createdAt, + updatedAt: createdAt, + deletedAt: null, + }; + + const layerDatabase = SqlitePersistence.layerMemory; + const layerOrchestrator = ProviderReplayHarness.layerWithRegistry( + { name: "factory-sessions-contract" }, + ClaudeOrchestratorReplayHarness.makeProviderAdapterRegistryLayer(transcript), + { databaseLayer: layerDatabase }, + ); + const layerServerSettings = ServerSettings.layerTest(); + const layerReceipts = CommandReceiptStore.layer.pipe(Layer.provide(layerDatabase)); + const layerThreadManagement = ThreadManagement.layer.pipe( + Layer.provide(layerOrchestrator), + Layer.provide(layerServerSettings), + Layer.provide(layerReceipts), + ); + // Real worktrees through the git driver; the rest of the workflow + // service is not on the launch path. + const layerGitWorkflow = Layer.unwrap( + Effect.gen(function* () { + const driver = yield* GitVcsDriver.GitVcsDriver; + return Layer.mock(GitWorkflow.GitWorkflowService)({ + createWorktree: (worktree, options) => + driver.createWorktree(worktree, { + ...options, + worktreesDirectory: input.worktreesDir, + }), + hasCommit: () => Effect.succeed(false), + remoteExists: (remote) => driver.remoteExists(remote), + removeWorktree: (worktree) => driver.removeWorktree(worktree), + }); + }), + ); + const layerExternalServices = Layer.mergeAll( + WorktreeSetupTracker.layer, + Layer.mock(ProjectCloneTracker.ProjectCloneTracker)({ get: () => Effect.succeed(null) }), + Layer.mock(TerminalManager.TerminalManager)({ close: () => Effect.void }), + Layer.mock(ProjectService.ProjectService)({ + getById: (id) => Effect.succeed(id === projectId ? Option.some(project) : Option.none()), + }), + layerGitWorkflow, + Layer.mock(ProjectSetupScriptRunner.ProjectSetupScriptRunner)({ + runForThread: () => Effect.succeed({ status: "no-script" as const }), + }), + // No branch names in this test: the temporary worktree branch sticks. + Layer.mock(TextGeneration.TextGeneration)({ + generateBranchName: () => Effect.die("branch names are not generated here"), + }), + layerServerSettings, + ProviderRegistryMock.layer(), + Layer.mock(ManagedProjectFolders.ManagedProjectFolders)({ + folderForThread: () => Effect.succeed(Option.none()), + }), + ); + const layerLaunch = ThreadLaunch.layer.pipe( + Layer.provide( + Layer.mergeAll( + layerExternalServices, + layerThreadManagement, + layerReceipts, + IdAllocator.layer, + ), + ), + ); + const layerStores = Layer.merge(ProjectionStore.layer, ProjectStore.layer).pipe( + Layer.provide(layerDatabase), + ); + return Layer.mergeAll( + Sessions.layer.pipe(Layer.provide(Layer.merge(layerLaunch, layerStores))), + layerStores, + layerOrchestrator, + ); + }), + ); + +const layerGit = GitVcsDriver.layer.pipe( + Layer.provide(VcsProcess.layer), + Layer.provide(NodeServices.layer), +); + +it.effect( + "launches a thread in a new worktree, reads its status to completion, and starts one thread per command id", + () => + Effect.scoped( + Effect.gen(function* () { + const fs = yield* FileSystem.FileSystem; + const path = yield* Path.Path; + const base = yield* fs.makeTempDirectoryScoped({ prefix: "t3-factory-sessions-" }); + const repo = path.join(base, "repo"); + const worktreesDir = path.join(base, "worktrees"); + yield* fs.makeDirectory(repo); + yield* fs.makeDirectory(worktreesDir); + const baseRef = yield* initRepoWithCommit(repo); + + yield* Effect.gen(function* () { + const projects = yield* ProjectStore.ProjectStoreV2; + yield* projects.apply({ + sequence: 1, + eventId: EventId.make("event:factory-sessions-contract:project-created"), + aggregateKind: "project", + aggregateId: projectId, + occurredAt: createdAt, + commandId: null, + causationEventId: null, + correlationId: null, + metadata: {}, + type: "project.created", + payload: { + projectId, + title: "Factory contract", + workspaceRoot: repo, + defaultModelSelection: null, + scripts: [], + createdAt, + updatedAt: createdAt, + }, + }); + + const sessions = yield* Sessions.FactorySessions; + assert.strictEqual(yield* sessions.findProject(repo), projectId); + assert.strictEqual(yield* sessions.findProject(`${repo}/`), null); + + const request: Sessions.LaunchRequest = { + commandId: "softfac:factory-sessions-contract:attempt-1", + projectId, + title: "Factory contract", + baseRef, + startFromOrigin: false, + message: SIMPLE_PROMPT, + model: { instanceId: "claudeAgent", model: "claude-sonnet-4-6" }, + }; + const launched = yield* sessions.launch(request); + const threadId = ThreadId.make(launched.threadId); + + // The worktree is prepared and the provider turn run after launch + // returns; the run's own events say when it is over. + const sink = yield* EventSink.EventSinkV2; + const terminal = yield* sink.stream({ threadId, eventType: "run.updated" }).pipe( + Stream.filterMap((stored) => + stored.event.type === "run.updated" && + TERMINAL_RUN_STATUSES.has(stored.event.payload.status) + ? Option.some(stored.event.payload) + : Option.none(), + ), + Stream.runHead, + ); + assert.isTrue(Option.isSome(terminal)); + assert.strictEqual(Option.getOrThrow(terminal).status, "completed"); + + const status = yield* sessions.status(launched.threadId); + assert.strictEqual(status.run, "completed"); + assert.strictEqual(status.runOrdinal, 1); + assert.include(status.lastAssistantMessage ?? "", "fixture simple ok"); + assert.isFalse(status.archived); + assert.isNotNull(status.branch); + assert.isNotNull(status.worktreePath); + const worktreePath = status.worktreePath!; + assert.isTrue(yield* fs.exists(path.join(worktreePath, ".git"))); + assert.strictEqual(path.dirname(path.dirname(worktreePath)), worktreesDir); + assert.strictEqual(yield* git(worktreePath, ["branch", "--show-current"]), status.branch); + assert.strictEqual( + yield* git(worktreePath, ["rev-parse", "HEAD"]), + yield* git(repo, ["rev-parse", baseRef]), + ); + + assert.deepStrictEqual(yield* sessions.thread(launched.threadId), { + worktreePath, + branch: status.branch, + archived: false, + }); + + // The same command id again is the same launch: the thread it made, + // and no second run. + const again = yield* sessions.launch(request); + assert.strictEqual(again.threadId, launched.threadId); + const after = yield* sessions.status(launched.threadId); + assert.strictEqual(after.runOrdinal, 1); + assert.strictEqual(after.run, "completed"); + }).pipe(Effect.provide(layerRuntime({ repo, worktreesDir }))); + }), + ).pipe(Effect.provide(Layer.merge(layerGit, NodeServices.layer))), +); diff --git a/apps/server/src/factory/t3/sessions.test.ts b/apps/server/src/factory/t3/sessions.test.ts new file mode 100644 index 000000000000..8e584b0d172f --- /dev/null +++ b/apps/server/src/factory/t3/sessions.test.ts @@ -0,0 +1,527 @@ +// @effect-diagnostics nodeBuiltinImport:off preferSchemaOverJson:off - the recorded snapshot is read as the wire bytes softfac saw. +import * as NodeFS from "node:fs"; +import * as NodePath from "node:path"; + +import { assert, describe, it } from "@effect/vitest"; +import { + EventId, + ProjectId, + ThreadId, + type OrchestrationV2AppThread, + type OrchestrationV2ThreadProjection, +} from "@t3tools/contracts"; +import * as Effect from "effect/Effect"; +import * as Layer from "effect/Layer"; + +import * as ProjectionStore from "../../orchestration-v2/ProjectionStore.ts"; +import * as ProjectStore from "../../orchestration-v2/ProjectStore.ts"; +import * as ThreadLaunch from "../../orchestration-v2/ThreadLaunchService.ts"; +import * as SqlitePersistence from "../../persistence/Sqlite.ts"; +import * as Sessions from "./sessions.ts"; + +// ---- status: the run, its message and the thread fields, read from a projection ---- + +const thread = ( + branch: string | null = null, + worktreePath: string | null = null, + archivedAt: string | null = null, +) => ({ branch, worktreePath, archivedAt }); + +const run = (ordinal: number, status: Sessions.RunStatus) => ({ + id: `run-${ordinal}`, + ordinal, + status, + startedAt: "2026-10-05T06:39:38.376Z", +}); + +/** A run that was queued and never started: T3 gives it no `startedAt`. */ +const unstartedRun = (ordinal: number, status: Sessions.RunStatus) => ({ + id: `run-${ordinal}`, + ordinal, + status, + startedAt: null, +}); + +/** A message of the run with `ordinal`, which T3 references by the run's `id`. */ +const message = (ordinal: number | null, role: string, text: string) => ({ + role, + text, + runId: ordinal === null ? null : `run-${ordinal}`, +}); + +const statusOf = ( + runs: ReadonlyArray, + messages: ReadonlyArray = [], + fields = thread(), +) => Sessions.threadStatusOf({ thread: fields, runs, messages }); + +describe("threadStatusOf", () => { + it("reads a real recorded snapshot and ignores unknown fields", () => { + const snapshot = JSON.parse( + NodeFS.readFileSync( + NodePath.join(import.meta.dirname, "fixtures", "t3_thread_snapshot.json"), + "utf8", + ), + ) as { readonly projection: Sessions.ThreadSnapshot }; + assert.deepStrictEqual(Sessions.threadStatusOf(snapshot.projection), { + run: "completed", + runOrdinal: 1, + lastAssistantMessage: "READY", + worktreePath: "/tmp/t3-spike-home/worktrees/t3-spike-repo/t3code-618ff943", + branch: "t3code/618ff943", + archived: false, + }); + }); + + it("takes the run with the highest ordinal whatever the order", () => { + // The highest ordinal is in the middle, so first, last and lowest all get it wrong. + const status = statusOf([run(1, "completed"), run(3, "running"), run(2, "failed")]); + assert.strictEqual(status.run, "running"); + }); + + it("skips a cancelled run that never started", () => { + // Run 1 is working; the person's queued message became run 2 and was cancelled. + const active: ReadonlyArray = [ + "preparing", + "queued", + "starting", + "running", + "waiting", + ]; + for (const expected of active) { + // The cancelled run comes first, so "last in the array" gets it wrong too. + const status = statusOf([unstartedRun(2, "cancelled"), run(1, expected)]); + assert.strictEqual(status.run, expected, expected); + assert.strictEqual(status.runOrdinal, 1, expected); + } + }); + + it("keeps the finished run and its message behind a cancelled queued run", () => { + const status = statusOf( + [run(1, "completed"), unstartedRun(2, "cancelled")], + [ + message(1, "user", "do the work"), + message(1, "assistant", "all done. FACTORY-DONE: ready"), + message(2, "user", "continue the work"), + ], + ); + assert.strictEqual(status.run, "completed"); + assert.strictEqual(status.runOrdinal, 1); + assert.strictEqual(status.lastAssistantMessage, "all done. FACTORY-DONE: ready"); + }); + + it("of a sole cancelled run that never started is cancelled", () => { + const status = statusOf([unstartedRun(1, "cancelled")]); + assert.strictEqual(status.run, "cancelled"); + assert.strictEqual(status.runOrdinal, 1); + }); + + it("of a sole cancelled run that started is cancelled", () => { + const status = statusOf([run(1, "cancelled")]); + assert.strictEqual(status.run, "cancelled"); + assert.strictEqual(status.runOrdinal, 1); + }); + + it("lets a cancelled run that started decide", () => { + const status = statusOf([run(1, "completed"), run(2, "cancelled")]); + assert.strictEqual(status.run, "cancelled"); + assert.strictEqual(status.runOrdinal, 2); + }); + + it("skips every cancelled run that never started", () => { + const status = statusOf([ + unstartedRun(3, "cancelled"), + run(1, "running"), + unstartedRun(2, "cancelled"), + ]); + assert.strictEqual(status.run, "running"); + assert.strictEqual(status.runOrdinal, 1); + }); + + it("takes the highest ordinal when every run is cancelled and never started", () => { + // The highest ordinal comes first, so "last in the array" gets it wrong too. + const status = statusOf([unstartedRun(2, "cancelled"), unstartedRun(1, "cancelled")]); + assert.strictEqual(status.run, "cancelled"); + assert.strictEqual(status.runOrdinal, 2); + }); + + it("keeps a queued run that has not started yet", () => { + // Only a cancelled run is left out: a message still waiting in the queue is the + // thread's run. + const status = statusOf([unstartedRun(2, "queued"), run(1, "completed")]); + assert.strictEqual(status.run, "queued"); + assert.strictEqual(status.runOrdinal, 2); + }); + + it("treats a cancelled run without startedAt as never started", () => { + const status = statusOf([{ id: "run-2", ordinal: 2, status: "cancelled" }, run(1, "running")]); + assert.strictEqual(status.run, "running"); + assert.strictEqual(status.runOrdinal, 1); + }); + + it("reports the ordinal of the run it describes", () => { + // Out of order, with the highest in the middle: the status and the ordinal are of the + // same run. + const status = statusOf([run(4, "completed"), run(9, "running"), run(2, "failed")]); + assert.strictEqual(status.run, "running"); + assert.strictEqual(status.runOrdinal, 9); + }); + + it("without runs has no run and no run ordinal", () => { + const status = statusOf([]); + assert.strictEqual(status.run, null); + assert.strictEqual(status.runOrdinal, null); + assert.strictEqual(status.lastAssistantMessage, null); + }); + + it("maps every known run status", () => { + for (const status of Sessions.RUN_STATUSES) { + assert.strictEqual(statusOf([run(1, status)]).run, status, status); + } + }); + + it("reports only the reported run's own last assistant message", () => { + // A run that ended quietly must not be read as ending with an older run's result + // line, and a message with no run reference is no run's. + const status = statusOf( + [run(1, "completed"), run(2, "completed")], + [ + message(1, "assistant", "FACTORY-DONE: old"), + message(2, "user", "go again"), + message(null, "assistant", "orphan"), + ], + ); + assert.strictEqual(status.runOrdinal, 2); + assert.strictEqual(status.lastAssistantMessage, null); + }); + + it("reports the thread's branch, worktree path and whether it is archived", () => { + const status = statusOf([], [], thread("t3/abc", "/tmp/wt", "2026-10-05T06:39:38.376Z")); + assert.strictEqual(status.branch, "t3/abc"); + assert.strictEqual(status.worktreePath, "/tmp/wt"); + assert.strictEqual(status.archived, true); + assert.strictEqual(statusOf([], [], thread()).archived, false); + }); +}); + +// ---- the service over T3's stores ---- + +const projectId = ProjectId.make("project:factory-sessions-test"); +const threadId = ThreadId.make("thread:project:factory-sessions-test:1"); + +const projectCreated = (id: string, workspaceRoot: string, sequence: number) => + ({ + sequence, + eventId: EventId.make(`event:${id}:created`), + aggregateKind: "project", + aggregateId: ProjectId.make(id), + occurredAt: "2026-10-03T00:00:00.000Z", + commandId: null, + causationEventId: null, + correlationId: null, + metadata: {}, + type: "project.created", + payload: { + projectId: ProjectId.make(id), + title: id, + workspaceRoot, + defaultModelSelection: null, + scripts: [], + createdAt: "2026-10-03T00:00:00.000Z", + updatedAt: "2026-10-03T00:00:00.000Z", + }, + }) as const; + +const projectDeleted = (id: string, sequence: number) => + ({ + sequence, + eventId: EventId.make(`event:${id}:deleted`), + aggregateKind: "project", + aggregateId: ProjectId.make(id), + occurredAt: "2026-10-03T00:00:00.000Z", + commandId: null, + causationEventId: null, + correlationId: null, + metadata: {}, + type: "project.deleted", + payload: { projectId: ProjectId.make(id), deletedAt: "2026-10-03T00:00:00.000Z" }, + }) as const; + +const layerProjects = ProjectStore.layer.pipe(Layer.provide(SqlitePersistence.layerMemory)); + +const layerSessions = ( + projections: Partial = {}, + launches: Partial = {}, +) => + Layer.mergeAll( + Sessions.layer.pipe( + Layer.provide( + Layer.mergeAll( + layerProjects, + Layer.mock(ProjectionStore.ProjectionStoreV2)(projections), + Layer.mock(ThreadLaunch.ThreadLaunchService)(launches), + ), + ), + ), + layerProjects, + ); + +describe("FactorySessions.findProject", () => { + it.effect("returns the id of the matching workspace root", () => + Effect.gen(function* () { + const projects = yield* ProjectStore.ProjectStoreV2; + yield* projects.apply(projectCreated("p-other", "/tmp/other", 1)); + yield* projects.apply(projectCreated("p-spike", "/tmp/t3-spike-repo", 2)); + const sessions = yield* Sessions.FactorySessions; + assert.strictEqual(yield* sessions.findProject("/tmp/t3-spike-repo"), "p-spike"); + }).pipe(Effect.provide(layerSessions())), + ); + + it.effect("without a match is null", () => + Effect.gen(function* () { + const projects = yield* ProjectStore.ProjectStoreV2; + yield* projects.apply(projectCreated("p-other", "/tmp/other", 1)); + const sessions = yield* Sessions.FactorySessions; + assert.strictEqual(yield* sessions.findProject("/tmp/t3-spike-repo"), null); + }).pipe(Effect.provide(layerSessions())), + ); + + it.effect("ignores deleted projects", () => + Effect.gen(function* () { + const projects = yield* ProjectStore.ProjectStoreV2; + yield* projects.apply(projectCreated("p-old", "/tmp/t3-spike-repo", 1)); + yield* projects.apply(projectDeleted("p-old", 2)); + const sessions = yield* Sessions.FactorySessions; + assert.strictEqual(yield* sessions.findProject("/tmp/t3-spike-repo"), null); + yield* projects.apply(projectCreated("p-new", "/tmp/t3-spike-repo", 3)); + assert.strictEqual(yield* sessions.findProject("/tmp/t3-spike-repo"), "p-new"); + }).pipe(Effect.provide(layerSessions())), + ); + + it.effect("matches the whole root, not a prefix or a trailing slash", () => + Effect.gen(function* () { + const projects = yield* ProjectStore.ProjectStoreV2; + yield* projects.apply(projectCreated("p-a", "/tmp/t3-spike-repo-2", 1)); + yield* projects.apply(projectCreated("p-b", "/tmp/t3-spike-repo/", 2)); + const sessions = yield* Sessions.FactorySessions; + assert.strictEqual(yield* sessions.findProject("/tmp/t3-spike-repo"), null); + }).pipe(Effect.provide(layerSessions())), + ); +}); + +describe("FactorySessions.launch", () => { + const request: Sessions.LaunchRequest = { + commandId: "softfac:issue-5:attempt-1", + projectId, + title: "Build issue 5", + baseRef: "main", + startFromOrigin: true, + message: "/factory 5", + model: { instanceId: "claudeAgent", model: "claude-sonnet-5-5" }, + }; + + it.effect( + "starts a thread in a new worktree, full-access, default interaction, under the caller's command id", + () => + Effect.gen(function* () { + const inputs: Array = []; + const launched = yield* Effect.gen(function* () { + const sessions = yield* Sessions.FactorySessions; + return yield* sessions.launch(request); + }).pipe( + Effect.provide( + layerSessions( + {}, + { + launch: (input) => + Effect.sync(() => { + inputs.push(input); + return { + threadId, + resumed: false, + projection: { + thread: { branch: null, worktreePath: null }, + } as OrchestrationV2ThreadProjection, + }; + }), + }, + ), + ), + ); + assert.deepStrictEqual(launched, { threadId, worktreePath: null, branch: null }); + assert.strictEqual(inputs.length, 1); + const input = inputs[0]!; + assert.strictEqual(input.commandId, "softfac:issue-5:attempt-1"); + assert.strictEqual(input.projectId, projectId); + assert.strictEqual(input.title, "Build issue 5"); + assert.deepStrictEqual(input.modelSelection, { + instanceId: "claudeAgent", + model: "claude-sonnet-5-5", + }); + assert.strictEqual(input.runtimeMode, "full-access"); + assert.strictEqual(input.interactionMode, "default"); + assert.deepStrictEqual(input.workspaceStrategy, { + type: "worktree", + baseRef: "main", + startFromOrigin: true, + }); + assert.strictEqual(input.initialMessage?.text, "/factory 5"); + assert.deepStrictEqual(input.initialMessage?.attachments, []); + assert.strictEqual(input.threadId, undefined); + assert.strictEqual(input.reuseExistingThread, undefined); + }), + ); + + it.effect("reports the branch and worktree path once T3 has them", () => + Effect.gen(function* () { + const sessions = yield* Sessions.FactorySessions; + const launched = yield* sessions.launch(request); + assert.deepStrictEqual(launched, { + threadId, + worktreePath: "/tmp/wt/t3-abc", + branch: "t3/abc", + }); + }).pipe( + Effect.provide( + layerSessions( + {}, + { + launch: () => + Effect.succeed({ + threadId, + resumed: true, + projection: { + thread: { branch: "t3/abc", worktreePath: "/tmp/wt/t3-abc" }, + } as OrchestrationV2ThreadProjection, + }), + }, + ), + ), + ), + ); + + it.effect("fails as a sessions error when T3 rejects the launch", () => + Effect.gen(function* () { + const sessions = yield* Sessions.FactorySessions; + const error = yield* Effect.flip(sessions.launch(request)); + assert.strictEqual(error._tag, "FactorySessionsError"); + assert.strictEqual(error.operation, "launch"); + }).pipe( + Effect.provide( + layerSessions( + {}, + { + launch: (input) => + Effect.fail( + new ThreadLaunch.ThreadLaunchError({ + operation: "resolve-project", + commandId: input.commandId, + projectId: input.projectId, + cause: "Project not found.", + }), + ), + }, + ), + ), + ), + ); +}); + +describe("FactorySessions.status and thread", () => { + const appThread = (fields: ReturnType) => + ({ + ...fields, + archivedAt: fields.archivedAt === null ? null : new Date(fields.archivedAt), + }) as unknown as OrchestrationV2AppThread; + + it.effect("status reads the run, its last assistant message and the thread fields", () => + Effect.gen(function* () { + const sessions = yield* Sessions.FactorySessions; + assert.deepStrictEqual(yield* sessions.status(threadId), { + run: "completed", + runOrdinal: 2, + lastAssistantMessage: "FACTORY-DONE: https://example.test/pr/1", + worktreePath: "/tmp/wt/t3-abc", + branch: "t3/abc", + archived: false, + }); + }).pipe( + Effect.provide( + layerSessions({ + getThreadProjection: (id) => + Effect.succeed({ + thread: appThread(thread("t3/abc", "/tmp/wt/t3-abc")), + runs: [run(1, "completed"), run(2, "completed")].map((r) => ({ + ...r, + threadId: id, + })), + messages: [ + message(1, "assistant", "first"), + message(2, "assistant", "FACTORY-DONE: https://example.test/pr/1"), + ], + } as unknown as OrchestrationV2ThreadProjection), + }), + ), + ), + ); + + it.effect("status of a thread T3 does not know is not found", () => + Effect.gen(function* () { + const sessions = yield* Sessions.FactorySessions; + const error = yield* Effect.flip(sessions.status(threadId)); + assert.strictEqual(error._tag, "FactoryThreadNotFoundError"); + if (error._tag === "FactoryThreadNotFoundError") { + assert.strictEqual(error.threadId, threadId); + } + }).pipe( + Effect.provide( + layerSessions({ + getThreadProjection: (id) => + Effect.fail(new ProjectionStore.ProjectionStoreThreadNotFoundError({ threadId: id })), + }), + ), + ), + ); + + it.effect("thread reads the thread's own fields and works when the runs cannot be read", () => + Effect.gen(function* () { + const sessions = yield* Sessions.FactorySessions; + assert.deepStrictEqual(yield* sessions.thread(threadId), { + worktreePath: "/tmp/wt/t3-abc", + branch: "t3/abc", + archived: true, + }); + const error = yield* Effect.flip(sessions.status(threadId)); + assert.strictEqual(error._tag, "FactorySessionsError"); + }).pipe( + Effect.provide( + layerSessions({ + getThread: () => + Effect.succeed(appThread(thread("t3/abc", "/tmp/wt/t3-abc", "2026-10-05T00:00:00Z"))), + getThreadProjection: (id) => + Effect.fail( + new ProjectionStore.ProjectionStoreReadError({ + threadId: id, + cause: new Error("a run row did not decode"), + }), + ), + }), + ), + ), + ); + + it.effect("thread of a thread T3 does not know is not found", () => + Effect.gen(function* () { + const sessions = yield* Sessions.FactorySessions; + const error = yield* Effect.flip(sessions.thread(threadId)); + assert.strictEqual(error._tag, "FactoryThreadNotFoundError"); + }).pipe( + Effect.provide( + layerSessions({ + getThread: (id) => + Effect.fail(new ProjectionStore.ProjectionStoreThreadNotFoundError({ threadId: id })), + }), + ), + ), + ); +}); From 53d91d5b87900a074f181f29e654b4b8334c69f8 Mon Sep 17 00:00:00 2001 From: Tom Kaczocha Date: Thu, 15 Oct 2026 21:00:42 +1000 Subject: [PATCH 2/5] feat(factory): sessions adapter over T3's launch and projection services FactorySessions ports softfac's Sessions trait in-process: findProject over ProjectStoreV2, launch over ThreadLaunchService (new worktree from baseRef or the origin tip, full-access, default interaction, the caller's command id so a retry is the same launch), status and thread over ProjectionStoreV2 with softfac's run selection. FactoryPaths gives the base dir, the worktrees dir and /factory from ServerConfig. The contract test waits for the run through EventSinkV2.stream with take(1): Stream.filterMap takes a Filter in Effect 4, not an Option, so the first draft never matched, and the test runs on the live clock because the orchestration runtime does not advance a TestClock. Issue: #5 --- apps/server/src/factory/t3/paths.test.ts | 4 +- apps/server/src/factory/t3/paths.ts | 34 +++ .../src/factory/t3/sessions.contract.test.ts | 24 +- apps/server/src/factory/t3/sessions.test.ts | 6 +- apps/server/src/factory/t3/sessions.ts | 268 ++++++++++++++++++ 5 files changed, 322 insertions(+), 14 deletions(-) create mode 100644 apps/server/src/factory/t3/paths.ts create mode 100644 apps/server/src/factory/t3/sessions.ts diff --git a/apps/server/src/factory/t3/paths.test.ts b/apps/server/src/factory/t3/paths.test.ts index 39193a787e69..d94dd2aac03e 100644 --- a/apps/server/src/factory/t3/paths.test.ts +++ b/apps/server/src/factory/t3/paths.test.ts @@ -18,10 +18,10 @@ it.effect( Effect.provide( Paths.layer.pipe( Layer.provide( - Layer.mock(ServerConfig.ServerConfig)({ + Layer.succeed(ServerConfig.ServerConfig, { baseDir: "/srv/t3fac", worktreesDir: "/srv/t3fac/worktrees", - }), + } as ServerConfig.ServerConfig["Service"]), ), Layer.provide(NodeServices.layer), ), diff --git a/apps/server/src/factory/t3/paths.ts b/apps/server/src/factory/t3/paths.ts new file mode 100644 index 000000000000..d7df8e5060c7 --- /dev/null +++ b/apps/server/src/factory/t3/paths.ts @@ -0,0 +1,34 @@ +import * as Context from "effect/Context"; +import * as Effect from "effect/Effect"; +import * as Layer from "effect/Layer"; +import * as Path from "effect/Path"; + +import * as ServerConfig from "../../config.ts"; + +/** + * Where the factory keeps and finds things on this machine, from the server's + * base dir: the worktrees T3 makes for agent threads, and the factory's own + * data under `/factory` (the same layout as softfac's data dir, so its + * retro can read both). + */ +export class FactoryPaths extends Context.Service< + FactoryPaths, + { + readonly baseDir: string; + readonly worktreesDir: string; + readonly factoryDir: string; + } +>()("t3/factory/t3/paths/FactoryPaths") {} + +export const layer = Layer.effect( + FactoryPaths, + Effect.gen(function* () { + const config = yield* ServerConfig.ServerConfig; + const path = yield* Path.Path; + return { + baseDir: config.baseDir, + worktreesDir: config.worktreesDir, + factoryDir: path.join(config.baseDir, "factory"), + }; + }), +); diff --git a/apps/server/src/factory/t3/sessions.contract.test.ts b/apps/server/src/factory/t3/sessions.contract.test.ts index f33e33f65dae..46c3f7ad265c 100644 --- a/apps/server/src/factory/t3/sessions.contract.test.ts +++ b/apps/server/src/factory/t3/sessions.contract.test.ts @@ -17,6 +17,7 @@ import * as Option from "effect/Option"; import * as Path from "effect/Path"; import * as Stream from "effect/Stream"; +import * as ServerConfig from "../../config.ts"; import * as GitWorkflow from "../../git/GitWorkflowService.ts"; import { ClaudeOrchestratorReplayHarness } from "../../orchestration-v2/Adapters/ClaudeAdapterV2.testkit.ts"; import * as CommandReceiptStore from "../../orchestration-v2/CommandReceiptStore.ts"; @@ -88,6 +89,7 @@ const initRepoWithCommit = (cwd: string) => const layerRuntime = (input: { readonly repo: string; readonly worktreesDir: string }) => Layer.unwrap( Effect.gen(function* () { + const path = yield* Path.Path; const transcript = yield* ClaudeOrchestratorReplayHarness.decodeTranscript( yield* readProviderReplayTranscript(SIMPLE_TRANSCRIPT), ); @@ -153,7 +155,8 @@ const layerRuntime = (input: { readonly repo: string; readonly worktreesDir: str layerServerSettings, ProviderRegistryMock.layer(), Layer.mock(ManagedProjectFolders.ManagedProjectFolders)({ - folderForThread: () => Effect.succeed(Option.none()), + namedProjectsRoot: path.join(input.worktreesDir, "..", "projects"), + folderForThread: () => Effect.succeedNone, }), ); const layerLaunch = ThreadLaunch.layer.pipe( @@ -179,10 +182,11 @@ const layerRuntime = (input: { readonly repo: string; readonly worktreesDir: str const layerGit = GitVcsDriver.layer.pipe( Layer.provide(VcsProcess.layer), + Layer.provide(ServerConfig.layerTest(process.cwd(), { prefix: "t3-factory-sessions-config-" })), Layer.provide(NodeServices.layer), ); -it.effect( +it.live( "launches a thread in a new worktree, reads its status to completion, and starts one thread per command id", () => Effect.scoped( @@ -240,16 +244,16 @@ it.effect( // returns; the run's own events say when it is over. const sink = yield* EventSink.EventSinkV2; const terminal = yield* sink.stream({ threadId, eventType: "run.updated" }).pipe( - Stream.filterMap((stored) => - stored.event.type === "run.updated" && - TERMINAL_RUN_STATUSES.has(stored.event.payload.status) - ? Option.some(stored.event.payload) - : Option.none(), + Stream.filter( + (stored) => + stored.event.type === "run.updated" && + TERMINAL_RUN_STATUSES.has(stored.event.payload.status), ), - Stream.runHead, + Stream.map((stored) => stored.event.payload as { readonly status: Sessions.RunStatus }), + Stream.take(1), + Stream.runCollect, ); - assert.isTrue(Option.isSome(terminal)); - assert.strictEqual(Option.getOrThrow(terminal).status, "completed"); + assert.strictEqual(terminal[0]?.status, "completed"); const status = yield* sessions.status(launched.threadId); assert.strictEqual(status.run, "completed"); diff --git a/apps/server/src/factory/t3/sessions.test.ts b/apps/server/src/factory/t3/sessions.test.ts index 8e584b0d172f..bae0bcc311d9 100644 --- a/apps/server/src/factory/t3/sessions.test.ts +++ b/apps/server/src/factory/t3/sessions.test.ts @@ -6,10 +6,12 @@ import { assert, describe, it } from "@effect/vitest"; import { EventId, ProjectId, + ProviderInstanceId, ThreadId, type OrchestrationV2AppThread, type OrchestrationV2ThreadProjection, } from "@t3tools/contracts"; +import * as DateTime from "effect/DateTime"; import * as Effect from "effect/Effect"; import * as Layer from "effect/Layer"; @@ -355,7 +357,7 @@ describe("FactorySessions.launch", () => { assert.strictEqual(input.projectId, projectId); assert.strictEqual(input.title, "Build issue 5"); assert.deepStrictEqual(input.modelSelection, { - instanceId: "claudeAgent", + instanceId: ProviderInstanceId.make("claudeAgent"), model: "claude-sonnet-5-5", }); assert.strictEqual(input.runtimeMode, "full-access"); @@ -431,7 +433,7 @@ describe("FactorySessions.status and thread", () => { const appThread = (fields: ReturnType) => ({ ...fields, - archivedAt: fields.archivedAt === null ? null : new Date(fields.archivedAt), + archivedAt: fields.archivedAt === null ? null : DateTime.makeUnsafe(fields.archivedAt), }) as unknown as OrchestrationV2AppThread; it.effect("status reads the run, its last assistant message and the thread fields", () => diff --git a/apps/server/src/factory/t3/sessions.ts b/apps/server/src/factory/t3/sessions.ts new file mode 100644 index 000000000000..c77072533928 --- /dev/null +++ b/apps/server/src/factory/t3/sessions.ts @@ -0,0 +1,268 @@ +import { CommandId, ProjectId, ProviderInstanceId, ThreadId } from "@t3tools/contracts"; +import * as Context from "effect/Context"; +import * as Effect from "effect/Effect"; +import * as Layer from "effect/Layer"; +import * as Option from "effect/Option"; +import * as Schema from "effect/Schema"; + +import * as ProjectionStore from "../../orchestration-v2/ProjectionStore.ts"; +import * as ProjectStore from "../../orchestration-v2/ProjectStore.ts"; +import * as ThreadLaunch from "../../orchestration-v2/ThreadLaunchService.ts"; + +/** + * Agent sessions as the factory sees them: softfac's `Sessions` trait + * (`src/t3.rs`) over T3's services, called in-process instead of over the + * wire. Everything above this file uses these types, never T3's. + */ + +/** Where a run is in its life, as T3 Code names it. */ +export const RUN_STATUSES = [ + "preparing", + "queued", + "starting", + "running", + "waiting", + "completed", + "interrupted", + "failed", + "cancelled", + "rolled_back", +] as const; +export type RunStatus = (typeof RUN_STATUSES)[number]; + +/** The model a thread runs on, as T3 Code names it. */ +export interface Model { + /** The configured provider instance, e.g. `claudeAgent`. */ + readonly instanceId: string; + /** The model id, e.g. `claude-sonnet-5-5`. */ + readonly model: string; +} + +/** Everything `launch` needs. */ +export interface LaunchRequest { + /** + * Names this command. Sent again after a failure, the same id lets T3 tell a + * retry from a new launch, so callers derive it from what they are launching + * (issue and attempt). + */ + readonly commandId: string; + /** A T3 project id, from `findProject`. */ + readonly projectId: string; + readonly title: string; + /** The ref the thread's worktree starts from. */ + readonly baseRef: string; + /** + * Fetch `origin` first and start from `baseRef`'s tip there, not from the + * clone's local ref: for a resume from a branch an earlier session pushed. + */ + readonly startFromOrigin: boolean; + /** The first message, which starts the first run. */ + readonly message: string; + readonly model: Model; +} + +/** What `launch` returns. */ +export interface Launched { + readonly threadId: string; + /** `null` until T3 Code has prepared the worktree. */ + readonly worktreePath: string | null; + /** `null` until T3 Code has prepared the worktree. */ + readonly branch: string | null; +} + +/** Where a thread works and whether it is archived. */ +export interface Thread { + /** `null` until T3 Code has prepared the worktree. */ + readonly worktreePath: string | null; + /** `null` until T3 Code has prepared the worktree. */ + readonly branch: string | null; + readonly archived: boolean; +} + +/** What `status` reports about a thread. */ +export interface ThreadStatus extends Thread { + /** + * The run with the highest ordinal, if the thread has any run yet. A + * cancelled run that never started is left out, unless every run is one. + */ + readonly run: RunStatus | null; + /** The `ordinal` of that same run; `null` when the thread has no run. */ + readonly runOrdinal: number | null; + /** The text of the last assistant message of that same run; `null` when the run has none. */ + readonly lastAssistantMessage: string | null; +} + +export class ThreadNotFoundError extends Schema.TaggedError()( + "FactoryThreadNotFoundError", + { threadId: Schema.String }, +) { + override get message(): string { + return `T3 does not know thread ${this.threadId}.`; + } +} + +export class SessionsError extends Schema.TaggedError()("FactorySessionsError", { + operation: Schema.Literals(["find-project", "launch", "status", "thread"]), + cause: Schema.Defect(), +}) { + override get message(): string { + return `Factory sessions ${this.operation} failed.`; + } +} + +/** The fields `status` and `thread` read from a thread projection. */ +export interface ThreadSnapshot { + readonly thread: ThreadFields; + readonly runs: ReadonlyArray<{ + /** What each of the run's messages carries as its `runId`. */ + readonly id: string; + readonly ordinal: number; + readonly status: RunStatus; + /** Absent for a run that never started. */ + readonly startedAt?: unknown; + }>; + readonly messages: ReadonlyArray<{ + readonly role: string; + readonly text: string; + /** The `id` of the run the message belongs to. */ + readonly runId: string | null; + }>; +} + +export interface ThreadFields { + readonly branch: string | null; + readonly worktreePath: string | null; + readonly archivedAt: unknown; +} + +export const threadOf = (thread: ThreadFields): Thread => ({ + worktreePath: thread.worktreePath, + branch: thread.branch, + archived: thread.archivedAt !== null && thread.archivedAt !== undefined, +}); + +/** A message queued behind a working run and cancelled before it started. */ +const isCancelledUnstarted = (run: ThreadSnapshot["runs"][number]) => + run.status === "cancelled" && (run.startedAt === null || run.startedAt === undefined); + +const highestOrdinal = (runs: ReadonlyArray) => + runs.reduce( + (best, run) => (best === undefined || run.ordinal >= best.ordinal ? run : best), + undefined, + ); + +export const threadStatusOf = (snapshot: ThreadSnapshot): ThreadStatus => { + // A cancelled run that never started is not the thread's run. When every run + // is one, the highest ordinal among them decides. + const latest = + highestOrdinal(snapshot.runs.filter((run) => !isCancelledUnstarted(run))) ?? + highestOrdinal(snapshot.runs); + // Only the reported run's own message: a run that ended quietly must not be + // read as ending with an older run's result line. A message with no run + // reference is no run's. + const lastAssistantMessage = + latest === undefined + ? null + : (snapshot.messages.findLast( + (message) => message.role === "assistant" && message.runId === latest.id, + )?.text ?? null); + return { + run: latest?.status ?? null, + runOrdinal: latest?.ordinal ?? null, + lastAssistantMessage, + ...threadOf(snapshot.thread), + }; +}; + +export class FactorySessions extends Context.Service< + FactorySessions, + { + /** The id of the live T3 project whose workspace root is exactly `workspaceRoot`. */ + readonly findProject: (workspaceRoot: string) => Effect.Effect; + /** + * Starts a thread in its own worktree and runs `message` in it. The + * worktree is prepared asynchronously, so the answer may not carry a + * branch or path yet; `status` will. The same `commandId` again is the + * same launch: T3 answers with the thread it made and starts nothing. + */ + readonly launch: (request: LaunchRequest) => Effect.Effect; + /** The status of the thread's run, the last assistant message and where the thread works. */ + readonly status: ( + threadId: string, + ) => Effect.Effect; + /** + * Where the thread works and whether it is archived. Only the thread's own + * fields are read, so this answers for a thread whose runs or messages + * `status` cannot read: the stop path uses it, so a session can always be + * stopped. + */ + readonly thread: ( + threadId: string, + ) => Effect.Effect; + } +>()("t3/factory/t3/sessions/FactorySessions") {} + +const isThreadNotFound = Schema.is(ProjectionStore.ProjectionStoreThreadNotFoundError); + +const mapReadError = + (operation: "status" | "thread", threadId: string) => + (cause: ProjectionStore.ProjectionStoreV2Error) => + isThreadNotFound(cause) + ? new ThreadNotFoundError({ threadId }) + : new SessionsError({ operation, cause }); + +export const make = Effect.gen(function* () { + const launches = yield* ThreadLaunch.ThreadLaunchService; + const projections = yield* ProjectionStore.ProjectionStoreV2; + const projects = yield* ProjectStore.ProjectStoreV2; + + const findProject: FactorySessions["Service"]["findProject"] = (workspaceRoot) => + projects.findActiveByWorkspaceRoot(workspaceRoot).pipe( + Effect.map(Option.match({ onNone: () => null, onSome: (row) => row.projectId as string })), + Effect.mapError((cause) => new SessionsError({ operation: "find-project", cause })), + ); + + const launch: FactorySessions["Service"]["launch"] = (request) => + launches + .launch({ + commandId: CommandId.make(request.commandId), + projectId: ProjectId.make(request.projectId), + title: request.title, + modelSelection: { + instanceId: ProviderInstanceId.make(request.model.instanceId), + model: request.model.model, + }, + runtimeMode: "full-access", + interactionMode: "default", + workspaceStrategy: { + type: "worktree", + baseRef: request.baseRef, + startFromOrigin: request.startFromOrigin, + }, + initialMessage: { text: request.message, attachments: [] }, + createdBy: "user", + creationSource: "server", + }) + .pipe( + Effect.map((result) => ({ + threadId: result.threadId as string, + worktreePath: result.projection.thread.worktreePath, + branch: result.projection.thread.branch, + })), + Effect.mapError((cause) => new SessionsError({ operation: "launch", cause })), + ); + + const status: FactorySessions["Service"]["status"] = (threadId) => + projections + .getThreadProjection(ThreadId.make(threadId)) + .pipe(Effect.map(threadStatusOf), Effect.mapError(mapReadError("status", threadId))); + + const thread: FactorySessions["Service"]["thread"] = (threadId) => + projections + .getThread(ThreadId.make(threadId)) + .pipe(Effect.map(threadOf), Effect.mapError(mapReadError("thread", threadId))); + + return FactorySessions.of({ findProject, launch, status, thread }); +}); + +export const layer = Layer.effect(FactorySessions, make); From 55551d4a5e866ba034b4c7486746877bb56e4fe9 Mon Sep 17 00:00:00 2001 From: Tom Kaczocha Date: Thu, 15 Oct 2026 21:07:24 +1000 Subject: [PATCH 3/5] test(factory): blank ids fail typed, last assistant message is the last From review cycle 1. Red: a blank commandId or threadId escapes as a schema defect instead of SessionsError / ThreadNotFoundError. The two-assistant-messages case is green already; it pins findLast against find, which every earlier test let through. Issue: #5 --- apps/server/src/factory/t3/sessions.test.ts | 66 +++++++++++++++++++++ 1 file changed, 66 insertions(+) diff --git a/apps/server/src/factory/t3/sessions.test.ts b/apps/server/src/factory/t3/sessions.test.ts index bae0bcc311d9..5b3ae81a4c68 100644 --- a/apps/server/src/factory/t3/sessions.test.ts +++ b/apps/server/src/factory/t3/sessions.test.ts @@ -197,6 +197,19 @@ describe("threadStatusOf", () => { assert.strictEqual(status.lastAssistantMessage, null); }); + it("reports the last of the run's assistant messages, not the first", () => { + // A real run says something on the way ("looking") before its result line. + const status = statusOf( + [run(1, "completed")], + [ + message(1, "assistant", "looking"), + message(1, "user", "carry on"), + message(1, "assistant", "FACTORY-DONE: ready"), + ], + ); + assert.strictEqual(status.lastAssistantMessage, "FACTORY-DONE: ready"); + }); + it("reports the thread's branch, worktree path and whether it is archived", () => { const status = statusOf([], [], thread("t3/abc", "/tmp/wt", "2026-10-05T06:39:38.376Z")); assert.strictEqual(status.branch, "t3/abc"); @@ -402,6 +415,38 @@ describe("FactorySessions.launch", () => { ), ); + it.effect("rejects a blank command id as a sessions error without reaching T3", () => + Effect.gen(function* () { + let launches = 0; + const error = yield* Effect.gen(function* () { + const sessions = yield* Sessions.FactorySessions; + return yield* Effect.flip(sessions.launch({ ...request, commandId: " " })); + }).pipe( + Effect.provide( + layerSessions( + {}, + { + launch: () => + Effect.sync(() => { + launches += 1; + return { + threadId, + resumed: false, + projection: { + thread: { branch: null, worktreePath: null }, + } as OrchestrationV2ThreadProjection, + }; + }), + }, + ), + ), + ); + assert.strictEqual(error._tag, "FactorySessionsError"); + assert.strictEqual(error.operation, "launch"); + assert.strictEqual(launches, 0); + }), + ); + it.effect("fails as a sessions error when T3 rejects the launch", () => Effect.gen(function* () { const sessions = yield* Sessions.FactorySessions; @@ -512,6 +557,27 @@ describe("FactorySessions.status and thread", () => { ), ); + it.effect("a blank thread id names no thread", () => + Effect.gen(function* () { + const sessions = yield* Sessions.FactorySessions; + assert.strictEqual( + (yield* Effect.flip(sessions.status("")))._tag, + "FactoryThreadNotFoundError", + ); + assert.strictEqual( + (yield* Effect.flip(sessions.thread(" ")))._tag, + "FactoryThreadNotFoundError", + ); + }).pipe( + Effect.provide( + layerSessions({ + getThread: () => Effect.die("a blank id must not reach the store"), + getThreadProjection: () => Effect.die("a blank id must not reach the store"), + }), + ), + ), + ); + it.effect("thread of a thread T3 does not know is not found", () => Effect.gen(function* () { const sessions = yield* Sessions.FactorySessions; From 24a0161ce9f5de2d98b3c38a92e163dc0ac32a6e Mon Sep 17 00:00:00 2001 From: Tom Kaczocha Date: Thu, 15 Oct 2026 21:08:19 +1000 Subject: [PATCH 4/5] fix(factory): blank ids fail as the declared sessions errors CommandId.make and ThreadId.make throw on a blank id. Decode the ids inside the Effect instead: a blank command, project or instance id is a launch SessionsError, and a blank thread id names no thread. Issue: #5 --- apps/server/src/factory/t3/sessions.ts | 63 ++++++++++++++++---------- 1 file changed, 40 insertions(+), 23 deletions(-) diff --git a/apps/server/src/factory/t3/sessions.ts b/apps/server/src/factory/t3/sessions.ts index c77072533928..b14d137d1db4 100644 --- a/apps/server/src/factory/t3/sessions.ts +++ b/apps/server/src/factory/t3/sessions.ts @@ -211,6 +211,17 @@ const mapReadError = ? new ThreadNotFoundError({ threadId }) : new SessionsError({ operation, cause }); +// Ids are T3's branded non-blank strings. A blank one is a caller's mistake +// and fails as the declared error, never as a thrown schema defect. +const decodeCommandId = Schema.decodeUnknownEffect(CommandId); +const decodeProjectId = Schema.decodeUnknownEffect(ProjectId); +const decodeInstanceId = Schema.decodeUnknownEffect(ProviderInstanceId); +const decodeThreadId = Schema.decodeUnknownEffect(ThreadId); + +/** A blank thread id names no thread. */ +const requireThreadId = (threadId: string) => + decodeThreadId(threadId).pipe(Effect.mapError(() => new ThreadNotFoundError({ threadId }))); + export const make = Effect.gen(function* () { const launches = yield* ThreadLaunch.ThreadLaunchService; const projections = yield* ProjectionStore.ProjectionStoreV2; @@ -223,15 +234,15 @@ export const make = Effect.gen(function* () { ); const launch: FactorySessions["Service"]["launch"] = (request) => - launches - .launch({ - commandId: CommandId.make(request.commandId), - projectId: ProjectId.make(request.projectId), + Effect.gen(function* () { + const commandId = yield* decodeCommandId(request.commandId); + const projectId = yield* decodeProjectId(request.projectId); + const instanceId = yield* decodeInstanceId(request.model.instanceId); + const result = yield* launches.launch({ + commandId, + projectId, title: request.title, - modelSelection: { - instanceId: ProviderInstanceId.make(request.model.instanceId), - model: request.model.model, - }, + modelSelection: { instanceId, model: request.model.model }, runtimeMode: "full-access", interactionMode: "default", workspaceStrategy: { @@ -242,25 +253,31 @@ export const make = Effect.gen(function* () { initialMessage: { text: request.message, attachments: [] }, createdBy: "user", creationSource: "server", - }) - .pipe( - Effect.map((result) => ({ - threadId: result.threadId as string, - worktreePath: result.projection.thread.worktreePath, - branch: result.projection.thread.branch, - })), - Effect.mapError((cause) => new SessionsError({ operation: "launch", cause })), - ); + }); + return { + threadId: result.threadId as string, + worktreePath: result.projection.thread.worktreePath, + branch: result.projection.thread.branch, + }; + }).pipe(Effect.mapError((cause) => new SessionsError({ operation: "launch", cause }))); const status: FactorySessions["Service"]["status"] = (threadId) => - projections - .getThreadProjection(ThreadId.make(threadId)) - .pipe(Effect.map(threadStatusOf), Effect.mapError(mapReadError("status", threadId))); + requireThreadId(threadId).pipe( + Effect.flatMap((id) => + projections + .getThreadProjection(id) + .pipe(Effect.map(threadStatusOf), Effect.mapError(mapReadError("status", threadId))), + ), + ); const thread: FactorySessions["Service"]["thread"] = (threadId) => - projections - .getThread(ThreadId.make(threadId)) - .pipe(Effect.map(threadOf), Effect.mapError(mapReadError("thread", threadId))); + requireThreadId(threadId).pipe( + Effect.flatMap((id) => + projections + .getThread(id) + .pipe(Effect.map(threadOf), Effect.mapError(mapReadError("thread", threadId))), + ), + ); return FactorySessions.of({ findProject, launch, status, thread }); }); From a282f3b668c7085bad983acb2f7a5b06980cf191 Mon Sep 17 00:00:00 2001 From: Tom Kaczocha Date: Thu, 15 Oct 2026 21:37:18 +1000 Subject: [PATCH 5/5] fix(factory): status reads only runs and assistant messages status is polled per session, so it now reads the runs and assistant messages through getThreadRecords instead of loading the whole thread projection. Adds tests for blank project and provider instance ids, thread read errors and project store failures failing as sessions errors. --- apps/server/src/factory/t3/sessions.test.ts | 75 ++++++++++++++++++--- apps/server/src/factory/t3/sessions.ts | 4 +- 2 files changed, 70 insertions(+), 9 deletions(-) diff --git a/apps/server/src/factory/t3/sessions.test.ts b/apps/server/src/factory/t3/sessions.test.ts index 5b3ae81a4c68..f38dd836af1a 100644 --- a/apps/server/src/factory/t3/sessions.test.ts +++ b/apps/server/src/factory/t3/sessions.test.ts @@ -322,6 +322,35 @@ describe("FactorySessions.findProject", () => { assert.strictEqual(yield* sessions.findProject("/tmp/t3-spike-repo"), null); }).pipe(Effect.provide(layerSessions())), ); + + it.effect("fails as a sessions error when the project store fails, not as no project", () => + Effect.gen(function* () { + const sessions = yield* Sessions.FactorySessions; + const error = yield* Effect.flip(sessions.findProject("/tmp/t3-spike-repo")); + assert.strictEqual(error._tag, "FactorySessionsError"); + assert.strictEqual(error.operation, "find-project"); + }).pipe( + Effect.provide( + Sessions.layer.pipe( + Layer.provide( + Layer.mergeAll( + Layer.mock(ProjectStore.ProjectStoreV2)({ + findActiveByWorkspaceRoot: () => + Effect.fail( + new ProjectStore.ProjectStoreV2Error({ + operation: "findActiveByWorkspaceRoot", + cause: new Error("database is locked"), + }), + ), + }), + Layer.mock(ProjectionStore.ProjectionStoreV2)({}), + Layer.mock(ThreadLaunch.ThreadLaunchService)({}), + ), + ), + ), + ), + ), + ); }); describe("FactorySessions.launch", () => { @@ -415,12 +444,12 @@ describe("FactorySessions.launch", () => { ), ); - it.effect("rejects a blank command id as a sessions error without reaching T3", () => + const rejectsWithoutLaunching = (blank: Sessions.LaunchRequest) => Effect.gen(function* () { let launches = 0; const error = yield* Effect.gen(function* () { const sessions = yield* Sessions.FactorySessions; - return yield* Effect.flip(sessions.launch({ ...request, commandId: " " })); + return yield* Effect.flip(sessions.launch(blank)); }).pipe( Effect.provide( layerSessions( @@ -444,7 +473,14 @@ describe("FactorySessions.launch", () => { assert.strictEqual(error._tag, "FactorySessionsError"); assert.strictEqual(error.operation, "launch"); assert.strictEqual(launches, 0); - }), + }); + + it.effect.each([ + { name: "command id", blank: { ...request, commandId: " " } }, + { name: "project id", blank: { ...request, projectId: " " } }, + { name: "provider instance id", blank: { ...request, model: { instanceId: "", model: "m" } } }, + ])("rejects a blank $name as a sessions error without reaching T3", ({ blank }) => + rejectsWithoutLaunching(blank), ); it.effect("fails as a sessions error when T3 rejects the launch", () => @@ -495,7 +531,7 @@ describe("FactorySessions.status and thread", () => { }).pipe( Effect.provide( layerSessions({ - getThreadProjection: (id) => + getThreadRecords: (id) => Effect.succeed({ thread: appThread(thread("t3/abc", "/tmp/wt/t3-abc")), runs: [run(1, "completed"), run(2, "completed")].map((r) => ({ @@ -506,7 +542,7 @@ describe("FactorySessions.status and thread", () => { message(1, "assistant", "first"), message(2, "assistant", "FACTORY-DONE: https://example.test/pr/1"), ], - } as unknown as OrchestrationV2ThreadProjection), + } as never), }), ), ), @@ -523,7 +559,7 @@ describe("FactorySessions.status and thread", () => { }).pipe( Effect.provide( layerSessions({ - getThreadProjection: (id) => + getThreadRecords: (id) => Effect.fail(new ProjectionStore.ProjectionStoreThreadNotFoundError({ threadId: id })), }), ), @@ -545,7 +581,7 @@ describe("FactorySessions.status and thread", () => { layerSessions({ getThread: () => Effect.succeed(appThread(thread("t3/abc", "/tmp/wt/t3-abc", "2026-10-05T00:00:00Z"))), - getThreadProjection: (id) => + getThreadRecords: (id) => Effect.fail( new ProjectionStore.ProjectionStoreReadError({ threadId: id, @@ -572,7 +608,30 @@ describe("FactorySessions.status and thread", () => { Effect.provide( layerSessions({ getThread: () => Effect.die("a blank id must not reach the store"), - getThreadProjection: () => Effect.die("a blank id must not reach the store"), + getThreadRecords: () => Effect.die("a blank id must not reach the store"), + }), + ), + ), + ); + + it.effect("thread fails as a sessions error when the thread cannot be read", () => + Effect.gen(function* () { + const sessions = yield* Sessions.FactorySessions; + const error = yield* Effect.flip(sessions.thread(threadId)); + assert.strictEqual(error._tag, "FactorySessionsError"); + if (error._tag === "FactorySessionsError") { + assert.strictEqual(error.operation, "thread"); + } + }).pipe( + Effect.provide( + layerSessions({ + getThread: (id) => + Effect.fail( + new ProjectionStore.ProjectionStoreReadError({ + threadId: id, + cause: new Error("the thread row did not decode"), + }), + ), }), ), ), diff --git a/apps/server/src/factory/t3/sessions.ts b/apps/server/src/factory/t3/sessions.ts index b14d137d1db4..4d9747edbb29 100644 --- a/apps/server/src/factory/t3/sessions.ts +++ b/apps/server/src/factory/t3/sessions.ts @@ -261,11 +261,13 @@ export const make = Effect.gen(function* () { }; }).pipe(Effect.mapError((cause) => new SessionsError({ operation: "launch", cause }))); + // The factory polls this per session, so it reads only the rows + // `threadStatusOf` uses, not the whole projection of a long thread. const status: FactorySessions["Service"]["status"] = (threadId) => requireThreadId(threadId).pipe( Effect.flatMap((id) => projections - .getThreadProjection(id) + .getThreadRecords(id, ["runs", "messages"], { messageRoles: ["assistant"] }) .pipe(Effect.map(threadStatusOf), Effect.mapError(mapReadError("status", threadId))), ), );