diff --git a/apps/mobile/src/features/threads/thread-settings-options.ts b/apps/mobile/src/features/threads/thread-settings-options.ts index b678154f83b..b741c2fd555 100644 --- a/apps/mobile/src/features/threads/thread-settings-options.ts +++ b/apps/mobile/src/features/threads/thread-settings-options.ts @@ -14,6 +14,11 @@ export const RUNTIME_MODE_CHOICES: ReadonlyArray<{ readonly label: string; readonly description: string; }> = [ + { + mode: "read-only", + label: "Read only", + description: "Allow inspection but deny commands and file changes that need write access.", + }, { mode: "approval-required", label: "Supervised", diff --git a/apps/server/src/mcp/OrchestratorMcpService.ts b/apps/server/src/mcp/OrchestratorMcpService.ts index 7d7bd24f02d..db11e83f313 100644 --- a/apps/server/src/mcp/OrchestratorMcpService.ts +++ b/apps/server/src/mcp/OrchestratorMcpService.ts @@ -373,6 +373,8 @@ function pageIncludesTerminalTaskResult(input: { function runtimeModeRank(mode: RuntimeMode): number { switch (mode) { + case "read-only": + return -1; case "approval-required": return 0; case "auto-accept-edits": diff --git a/apps/server/src/orchestration-v2/Adapters/ClaudeAdapterV2.ts b/apps/server/src/orchestration-v2/Adapters/ClaudeAdapterV2.ts index a52f220ee4b..e496563b283 100644 --- a/apps/server/src/orchestration-v2/Adapters/ClaudeAdapterV2.ts +++ b/apps/server/src/orchestration-v2/Adapters/ClaudeAdapterV2.ts @@ -1154,6 +1154,9 @@ const isClaudeRuntimeReadOnlyFullAccessSandboxPolicy = Schema.is( function sandboxPolicyKindForClaudeRuntimePolicy( runtimePolicy: ProviderAdapterV2RuntimePolicy, ): ClaudeRuntimeSandboxPolicyKindName | undefined { + if (runtimePolicy.sandboxPolicy === undefined && runtimePolicy.runtimeMode === "read-only") { + return "readOnly"; + } return runtimePolicy.sandboxPolicy !== undefined && isClaudeRuntimeSandboxPolicyKind(runtimePolicy.sandboxPolicy) ? runtimePolicy.sandboxPolicy.type @@ -1205,6 +1208,8 @@ function permissionModeForClaudeRuntimePolicy( } switch (runtimePolicy.runtimeMode) { + case "read-only": + return "dontAsk"; case "approval-required": return "default"; case "auto-accept-edits": diff --git a/apps/server/src/orchestration-v2/Adapters/CodexAdapterV2.test.ts b/apps/server/src/orchestration-v2/Adapters/CodexAdapterV2.test.ts index 5f272c21dea..fdb095307f3 100644 --- a/apps/server/src/orchestration-v2/Adapters/CodexAdapterV2.test.ts +++ b/apps/server/src/orchestration-v2/Adapters/CodexAdapterV2.test.ts @@ -266,7 +266,12 @@ describe("CodexAdapterV2 runtime policy", () => { it.effect("derives concrete Codex turn policies from every T3 runtime mode", () => Effect.gen(function* () { const build = ( - runtimeMode: "approval-required" | "auto-accept-edits" | "auto" | "full-access", + runtimeMode: + | "read-only" + | "approval-required" + | "auto-accept-edits" + | "auto" + | "full-access", ) => buildCodexTurnStartParams({ nativeThreadId: `native-${runtimeMode}`, @@ -282,11 +287,15 @@ describe("CodexAdapterV2 runtime policy", () => { }, }); + const readOnly = yield* build("read-only"); const approvalRequired = yield* build("approval-required"); const autoAcceptEdits = yield* build("auto-accept-edits"); const auto = yield* build("auto"); const fullAccess = yield* build("full-access"); + assert.equal(readOnly.approvalPolicy, "never"); + assert.equal(readOnly.approvalsReviewer, "user"); + assert.equal(readOnly.sandboxPolicy?.type, "readOnly"); assert.equal(approvalRequired.approvalPolicy, "untrusted"); assert.equal(approvalRequired.approvalsReviewer, "user"); assert.equal(approvalRequired.sandboxPolicy?.type, "readOnly"); diff --git a/apps/server/src/orchestration-v2/Adapters/CodexAdapterV2.ts b/apps/server/src/orchestration-v2/Adapters/CodexAdapterV2.ts index 1bbe97fd8c6..b7c7e719bee 100644 --- a/apps/server/src/orchestration-v2/Adapters/CodexAdapterV2.ts +++ b/apps/server/src/orchestration-v2/Adapters/CodexAdapterV2.ts @@ -537,6 +537,14 @@ function codexRuntimeModeTurnDefaults(runtimeMode: RuntimeMode): { readonly sandboxPolicy: CodexSchema.V2TurnStartParams__SandboxPolicy; } { switch (runtimeMode) { + case "read-only": + return { + approvalPolicy: "never", + approvalsReviewer: "user", + sandboxPolicy: { + type: "readOnly", + }, + }; case "approval-required": return { approvalPolicy: "untrusted", diff --git a/apps/server/src/orchestration-v2/EffectWorker.test.ts b/apps/server/src/orchestration-v2/EffectWorker.test.ts index b39105f36f8..56da249d465 100644 --- a/apps/server/src/orchestration-v2/EffectWorker.test.ts +++ b/apps/server/src/orchestration-v2/EffectWorker.test.ts @@ -117,6 +117,7 @@ function makeExecutorLayer(input: { Layer.succeed( ProviderTurnStartServiceV2, ProviderTurnStartServiceV2.of({ + fail: () => Effect.void, start: () => Effect.gen(function* () { yield* record("start"); diff --git a/apps/server/src/orchestration-v2/EffectWorker.ts b/apps/server/src/orchestration-v2/EffectWorker.ts index 1139d2fea3b..06bab360234 100644 --- a/apps/server/src/orchestration-v2/EffectWorker.ts +++ b/apps/server/src/orchestration-v2/EffectWorker.ts @@ -66,6 +66,10 @@ export interface OrchestrationEffectExecutorV2Shape { readonly execute: ( effect: OrchestrationEffectV2, ) => Effect.Effect; + readonly terminalize?: ( + effect: OrchestrationEffectV2, + error: string, + ) => Effect.Effect; } export class OrchestrationEffectExecutorV2 extends Context.Service< @@ -95,6 +99,25 @@ export const executorLayer: Layer.Layer< const runtimeRequests = yield* RuntimeRequestServiceV2; const threadTitleRegeneration = yield* ThreadTitleRegenerationService; return OrchestrationEffectExecutorV2.of({ + terminalize: (effect, failure) => { + if ( + effect.request.type !== "provider-turn.start" && + effect.request.type !== "provider-turn.restart" + ) + return Effect.void; + return providerTurnStart + .fail({ threadId: effect.threadId, runId: effect.request.runId, cause: failure }) + .pipe( + Effect.mapError( + (cause) => + new OrchestrationEffectExecutionError({ + effectId: effect.id, + effectType: effect.request.type, + cause, + }), + ), + ); + }, execute: (effect) => { switch (effect.request.type) { case "provider-session.detach": @@ -532,9 +555,12 @@ export const layerWithOptions = ( .succeed({ effectId: effect.id, workerId }) .pipe(Effect.onError((cause) => terminalizeClaim(effect, cause))) : effect.attemptCount >= maxAttempts - ? yield* outbox - .fail({ effectId: effect.id, workerId, error }) - .pipe(Effect.onError((cause) => terminalizeClaim(effect, cause))) + ? yield* Effect.gen(function* () { + yield* executor.terminalize?.(effect, error) ?? Effect.void; + return yield* outbox + .fail({ effectId: effect.id, workerId, error }) + .pipe(Effect.onError((cause) => terminalizeClaim(effect, cause))); + }) : yield* outbox .retry({ effectId: effect.id, diff --git a/apps/server/src/orchestration-v2/Orchestrator.ts b/apps/server/src/orchestration-v2/Orchestrator.ts index 186eda17ea3..b25e501fa26 100644 --- a/apps/server/src/orchestration-v2/Orchestrator.ts +++ b/apps/server/src/orchestration-v2/Orchestrator.ts @@ -5522,7 +5522,12 @@ const makeOrchestrator = Effect.fn("orchestrationV2.Orchestrator.layer")(functio occurredAt: now, payload: { ...state.preparationItem, - title: command.phase === "worktree" ? "Preparing worktree" : "Starting setup script", + title: + command.phase === "worktree" + ? "Preparing worktree" + : command.phase === "verification" + ? "Verifying prepared worktree" + : "Starting setup script", updatedAt: now, }, }); diff --git a/apps/server/src/orchestration-v2/PreparedWorktreeVerifier.test.ts b/apps/server/src/orchestration-v2/PreparedWorktreeVerifier.test.ts new file mode 100644 index 00000000000..a6b568783d4 --- /dev/null +++ b/apps/server/src/orchestration-v2/PreparedWorktreeVerifier.test.ts @@ -0,0 +1,73 @@ +import * as NodeServices from "@effect/platform-node/NodeServices"; +import { assert, it } from "@effect/vitest"; +import * as Effect from "effect/Effect"; +import * as Layer from "effect/Layer"; +import { ChildProcessSpawner } from "effect/unstable/process"; + +import * as GitVcsDriver from "../vcs/GitVcsDriver.ts"; +import * as PreparedWorktreeVerifier from "./PreparedWorktreeVerifier.ts"; + +const checkout = { + repositoryRoot: process.cwd(), + gitCommonDir: process.cwd(), + worktreePath: process.cwd(), + branch: "prepared", + startingCommit: "abc123", +} as const; + +function makeLayer(status = "") { + const execute: GitVcsDriver.GitVcsDriver["Service"]["execute"] = (input) => { + const command = input.args.join(" "); + let stdout = ""; + let exitCode = 0; + if (command === "rev-parse --show-toplevel") { + stdout = + input.cwd === checkout.repositoryRoot ? checkout.repositoryRoot : checkout.worktreePath; + } else if (command === "rev-parse --path-format=absolute --git-common-dir") { + stdout = checkout.gitCommonDir; + } else if (command === "symbolic-ref --quiet --short HEAD") { + stdout = checkout.branch; + } else if (command === "rev-parse HEAD") { + stdout = checkout.startingCommit; + } else if (command === "status --porcelain=v1 -z") { + stdout = status; + } else if (command === "worktree list --porcelain -z") { + stdout = [ + `worktree ${checkout.worktreePath}`, + `HEAD ${checkout.startingCommit}`, + `branch refs/heads/${checkout.branch}`, + "", + ].join("\0"); + } else { + exitCode = 1; + } + return Effect.succeed({ + exitCode: ChildProcessSpawner.ExitCode(exitCode), + stdout, + stderr: "", + stdoutTruncated: false, + stderrTruncated: false, + stdoutInvalidUtf8: false, + stderrInvalidUtf8: false, + }); + }; + return PreparedWorktreeVerifier.layer.pipe( + Layer.provide(Layer.mock(GitVcsDriver.GitVcsDriver)({ execute })), + Layer.provideMerge(NodeServices.layer), + ); +} + +it.effect("proves an exact clean registered worktree", () => + Effect.gen(function* () { + const verifier = yield* PreparedWorktreeVerifier.PreparedWorktreeVerifier; + assert.deepEqual(yield* verifier.verify(checkout, checkout.repositoryRoot), checkout); + }).pipe(Effect.provide(makeLayer())), +); + +it.effect("rejects a prepared worktree that became dirty", () => + Effect.gen(function* () { + const verifier = yield* PreparedWorktreeVerifier.PreparedWorktreeVerifier; + const failure = yield* Effect.flip(verifier.verify(checkout, checkout.repositoryRoot)); + assert.equal(failure.reason, "dirty_worktree"); + }).pipe(Effect.provide(makeLayer("?? changed.txt\0"))), +); diff --git a/apps/server/src/orchestration-v2/PreparedWorktreeVerifier.ts b/apps/server/src/orchestration-v2/PreparedWorktreeVerifier.ts new file mode 100644 index 00000000000..ab9c5134bde --- /dev/null +++ b/apps/server/src/orchestration-v2/PreparedWorktreeVerifier.ts @@ -0,0 +1,278 @@ +import type { PreparedWorktreeCheckout } from "@t3tools/contracts"; +import * as Context from "effect/Context"; +import * as Effect from "effect/Effect"; +import * as FileSystem from "effect/FileSystem"; +import * as Layer from "effect/Layer"; +import * as Path from "effect/Path"; +import * as Schema from "effect/Schema"; + +import * as GitVcsDriver from "../vcs/GitVcsDriver.ts"; + +export class PreparedWorktreeVerificationError extends Schema.TaggedErrorClass()( + "PreparedWorktreeVerificationError", + { + reason: Schema.Literals([ + "path_unavailable", + "repository_root_mismatch", + "git_common_dir_mismatch", + "worktree_not_registered", + "worktree_root_mismatch", + "detached_head", + "branch_mismatch", + "commit_mismatch", + "dirty_worktree", + "git_command_failed", + ]), + detail: Schema.String, + cause: Schema.optional(Schema.Defect()), + }, +) {} + +export interface PreparedWorktreeVerification { + readonly repositoryRoot: string; + readonly gitCommonDir: string; + readonly worktreePath: string; + readonly branch: string; + readonly startingCommit: string; +} + +export class PreparedWorktreeVerifier extends Context.Service< + PreparedWorktreeVerifier, + { + readonly verify: ( + checkout: PreparedWorktreeCheckout, + projectWorkspaceRoot: string, + ) => Effect.Effect; + } +>()("t3/orchestration-v2/PreparedWorktreeVerifier") {} + +interface RegisteredWorktree { + readonly path: string; + readonly head: string | null; + readonly branch: string | null; +} + +function parseRegisteredWorktrees(output: string): ReadonlyArray { + const worktrees: RegisteredWorktree[] = []; + let current: { path?: string; head?: string; branch?: string } = {}; + for (const field of output.split("\0")) { + if (field.length === 0) { + if (current.path !== undefined) { + worktrees.push({ + path: current.path, + head: current.head ?? null, + branch: current.branch?.replace(/^refs\/heads\//, "") ?? null, + }); + } + current = {}; + continue; + } + const separator = field.indexOf(" "); + const key = separator === -1 ? field : field.slice(0, separator); + const value = separator === -1 ? "" : field.slice(separator + 1); + if (key === "worktree") current.path = value; + if (key === "HEAD") current.head = value; + if (key === "branch") current.branch = value; + } + return worktrees; +} + +export const layer = Layer.effect( + PreparedWorktreeVerifier, + Effect.gen(function* () { + const fs = yield* FileSystem.FileSystem; + const path = yield* Path.Path; + const git = yield* GitVcsDriver.GitVcsDriver; + + const fail = ( + reason: PreparedWorktreeVerificationError["reason"], + detail: string, + cause?: unknown, + ) => + Effect.fail( + new PreparedWorktreeVerificationError({ + reason, + detail, + ...(cause === undefined ? {} : { cause }), + }), + ); + + const realPath = (label: string, value: string) => + fs.realPath(value).pipe( + Effect.map(path.normalize), + Effect.mapError( + (cause) => + new PreparedWorktreeVerificationError({ + reason: "path_unavailable", + detail: `${label} is unavailable: ${value}`, + cause, + }), + ), + ); + + const run = Effect.fn("PreparedWorktreeVerifier.runGit")(function* ( + cwd: string, + args: ReadonlyArray, + ) { + return yield* git + .execute({ + operation: "PreparedWorktreeVerifier.verify", + cwd, + args, + allowNonZeroExit: true, + timeoutMs: 5_000, + maxOutputBytes: 16 * 1024 * 1024, + }) + .pipe( + Effect.mapError( + (cause) => + new PreparedWorktreeVerificationError({ + reason: "git_command_failed", + detail: `Git failed while checking ${args.join(" ")}.`, + cause, + }), + ), + ); + }); + + const verify: PreparedWorktreeVerifier["Service"]["verify"] = Effect.fn( + "PreparedWorktreeVerifier.verify", + )(function* (checkout, projectWorkspaceRoot) { + const [repositoryRoot, projectRoot, expectedCommonDir, worktreePath] = yield* Effect.all( + [ + realPath("repository root", checkout.repositoryRoot), + realPath("project workspace root", projectWorkspaceRoot), + realPath("Git common directory", checkout.gitCommonDir), + realPath("worktree", checkout.worktreePath), + ], + { concurrency: 4 }, + ); + if (repositoryRoot !== projectRoot) { + return yield* fail( + "repository_root_mismatch", + `Expected project workspace root ${projectRoot}, received ${repositoryRoot}.`, + ); + } + const [repositoryTop, repositoryCommon, worktreeTop, worktreeCommon, branch, head, status] = + yield* Effect.all( + [ + run(repositoryRoot, ["rev-parse", "--show-toplevel"]), + run(repositoryRoot, ["rev-parse", "--path-format=absolute", "--git-common-dir"]), + run(worktreePath, ["rev-parse", "--show-toplevel"]), + run(worktreePath, ["rev-parse", "--path-format=absolute", "--git-common-dir"]), + run(worktreePath, ["symbolic-ref", "--quiet", "--short", "HEAD"]), + run(worktreePath, ["rev-parse", "HEAD"]), + run(worktreePath, ["status", "--porcelain=v1", "-z"]), + ], + { concurrency: 7 }, + ); + if (repositoryTop.exitCode !== 0) { + return yield* fail( + "repository_root_mismatch", + "The repository root is not a Git worktree.", + ); + } + const actualRepositoryRoot = yield* realPath( + "resolved repository root", + repositoryTop.stdout.trim(), + ); + if (actualRepositoryRoot !== repositoryRoot) { + return yield* fail( + "repository_root_mismatch", + `Expected repository root ${repositoryRoot}, received ${actualRepositoryRoot}.`, + ); + } + if (repositoryCommon.exitCode !== 0 || worktreeCommon.exitCode !== 0) { + return yield* fail( + "git_common_dir_mismatch", + "Git could not resolve the common directory.", + ); + } + const actualRepositoryCommon = yield* realPath( + "repository Git common directory", + repositoryCommon.stdout.trim(), + ); + const actualWorktreeCommon = yield* realPath( + "worktree Git common directory", + worktreeCommon.stdout.trim(), + ); + if ( + actualRepositoryCommon !== expectedCommonDir || + actualWorktreeCommon !== expectedCommonDir + ) { + return yield* fail( + "git_common_dir_mismatch", + `Expected Git common directory ${expectedCommonDir}.`, + ); + } + if (worktreeTop.exitCode !== 0) { + return yield* fail("worktree_root_mismatch", "The prepared path is not a Git worktree."); + } + const actualWorktreeRoot = yield* realPath( + "resolved worktree root", + worktreeTop.stdout.trim(), + ); + if (actualWorktreeRoot !== worktreePath) { + return yield* fail( + "worktree_root_mismatch", + `Expected worktree root ${worktreePath}, received ${actualWorktreeRoot}.`, + ); + } + const listed = yield* run(repositoryRoot, ["worktree", "list", "--porcelain", "-z"]); + if (listed.exitCode !== 0) { + return yield* fail("worktree_not_registered", "Git could not list registered worktrees."); + } + const registeredWorktrees = yield* Effect.forEach( + parseRegisteredWorktrees(listed.stdout), + (entry) => + fs.realPath(entry.path).pipe( + Effect.map((resolvedPath) => ({ entry, resolvedPath: path.normalize(resolvedPath) })), + Effect.catchCause(() => Effect.succeed({ entry, resolvedPath: null })), + ), + { concurrency: "unbounded" }, + ); + const registered = registeredWorktrees.find( + (candidate) => candidate.resolvedPath === worktreePath, + )?.entry; + if (registered === undefined) { + return yield* fail( + "worktree_not_registered", + `${worktreePath} is not a registered worktree.`, + ); + } + if (branch.exitCode !== 0) { + return yield* fail("detached_head", `${worktreePath} does not have a symbolic branch.`); + } + const actualBranch = branch.stdout.trim(); + if (actualBranch !== checkout.branch || registered.branch !== checkout.branch) { + return yield* fail( + "branch_mismatch", + `Expected branch ${checkout.branch}, received ${actualBranch}.`, + ); + } + const actualHead = head.stdout.trim(); + if ( + head.exitCode !== 0 || + actualHead !== checkout.startingCommit || + registered.head !== checkout.startingCommit + ) { + return yield* fail( + "commit_mismatch", + `Expected commit ${checkout.startingCommit}, received ${actualHead}.`, + ); + } + if (status.exitCode !== 0 || status.stdout.length > 0) { + return yield* fail("dirty_worktree", `${worktreePath} is not clean.`); + } + return { + repositoryRoot, + gitCommonDir: expectedCommonDir, + worktreePath, + branch: actualBranch, + startingCommit: actualHead, + }; + }); + + return PreparedWorktreeVerifier.of({ verify }); + }), +); diff --git a/apps/server/src/orchestration-v2/ProgramAttemptService.test.ts b/apps/server/src/orchestration-v2/ProgramAttemptService.test.ts new file mode 100644 index 00000000000..5acecdf24d6 --- /dev/null +++ b/apps/server/src/orchestration-v2/ProgramAttemptService.test.ts @@ -0,0 +1,356 @@ +import { assert, it, vi } from "@effect/vitest"; +import * as NodeServices from "@effect/platform-node/NodeServices"; +import { + MessageId, + ProgramAttemptId, + ProgramAttemptRequestId, + ProjectId, + ProviderInstanceId, + RunId, + ThreadId, + TurnItemId, + type OrchestrationV2RunStatus, + type OrchestrationV2ThreadProjection, +} from "@t3tools/contracts"; +import * as DateTime from "effect/DateTime"; +import * as Effect from "effect/Effect"; +import * as Exit from "effect/Exit"; +import * as Layer from "effect/Layer"; +import * as Ref from "effect/Ref"; + +import { SqlitePersistenceMemory } from "../persistence/Layers/Sqlite.ts"; +import * as ProgramAttemptService from "./ProgramAttemptService.ts"; +import * as ThreadLaunchService from "./ThreadLaunchService.ts"; +import * as ThreadManagementService from "./ThreadManagementService.ts"; + +const attemptId = ProgramAttemptId.make("attempt:s1"); +const projectId = ProjectId.make("project:s1"); +const threadId = ThreadId.make("thread:s1"); +const runId = RunId.make("run:s1"); +const providerInstanceId = ProviderInstanceId.make("codex"); +const modelSelection = { instanceId: providerInstanceId, model: "gpt-5.6-sol" } as const; +const now = DateTime.makeUnsafe("2026-08-19T00:00:00.000Z"); + +it.effect("retries the launch receipt gap without remounting the thread panel", () => + Effect.gen(function* () { + let requests = 0; + const result = yield* ProgramAttemptService.retryProgramAttemptReceipt( + () => Effect.sync(() => (++requests === 1 ? null : "snapshot")), + { attempts: 3, delay: Effect.void }, + ); + assert.strictEqual(result, "snapshot"); + assert.strictEqual(requests, 2); + }), +); + +function makeProjection(status: OrchestrationV2RunStatus): OrchestrationV2ThreadProjection { + const terminal = ThreadManagementService.isTerminalRunStatus(status); + return { + thread: { + createdBy: "system", + creationSource: "server", + id: threadId, + projectId, + title: "S1 disposable task", + providerInstanceId, + modelSelection, + runtimeMode: "full-access", + interactionMode: "default", + branch: "prepared", + worktreePath: "/repo-worktrees/prepared", + activeProviderThreadId: null, + lineage: { parentThreadId: null, relationshipToParent: null, rootThreadId: threadId }, + forkedFrom: null, + createdAt: now, + updatedAt: now, + archivedAt: null, + settledOverride: null, + settledAt: terminal ? now : null, + lastVisitedAt: null, + deletedAt: null, + }, + runs: [ + { + id: runId, + threadId, + ordinal: 1, + providerInstanceId, + modelSelection, + providerThreadId: null, + userMessageId: MessageId.make("message:s1:user"), + rootNodeId: null, + activeAttemptId: null, + status, + requestedAt: now, + startedAt: status === "preparing" ? null : now, + completedAt: terminal ? now : null, + checkpointId: null, + contextHandoffId: null, + }, + ], + attempts: [], + nodes: [], + subagents: [], + providerSessions: [], + providerThreads: [], + providerTurns: [], + runtimeRequests: [], + messages: [], + plans: [], + turnItems: + status === "completed" + ? [ + { + id: TurnItemId.make("turn-item:s1:assistant"), + threadId, + runId, + nodeId: null, + providerThreadId: null, + providerTurnId: null, + nativeItemRef: null, + parentItemId: null, + ordinal: 1, + status: "completed", + title: null, + startedAt: now, + completedAt: now, + updatedAt: now, + type: "assistant_message", + messageId: MessageId.make("message:s1:assistant"), + text: "Disposable task finished.", + streaming: false, + }, + ] + : [], + checkpointScopes: [], + checkpoints: [], + contextHandoffs: [], + contextTransfers: [], + visibleTurnItems: [], + updatedAt: now, + }; +} + +const launchInput = { + attemptId, + requestId: ProgramAttemptRequestId.make("request:s1:launch"), + programId: "agents-dlr", + taskId: "agents-dlr.2", + projectId, + title: "S1 disposable task", + prompt: "Reply once, then stop.", + checkout: { + repositoryRoot: "/repo", + gitCommonDir: "/repo/.git", + worktreePath: "/repo-worktrees/prepared", + branch: "prepared", + startingCommit: "abc123", + }, + providerPolicy: { + modelSelection, + runtimeMode: "full-access" as const, + interactionMode: "default" as const, + }, +}; + +function makeHarness() { + return Effect.gen(function* () { + const projection = yield* Ref.make(makeProjection("preparing")); + const launch = vi.fn(() => + Ref.get(projection).pipe( + Effect.map((current) => ({ threadId, projection: current, resumed: false })), + ), + ); + const interruptThread = vi.fn( + (_input: ThreadManagementService.ThreadManagementInterruptInput) => + Effect.succeed({ type: "no_active_run" as const }), + ); + const services = Layer.mergeAll( + Layer.succeed(ThreadLaunchService.ThreadLaunchService, { launch }), + Layer.mock(ThreadManagementService.ThreadManagementService)({ + getThreadProjection: (requestedThreadId) => + requestedThreadId === threadId ? Ref.get(projection) : Effect.die("missing test thread"), + interruptThread, + }), + ); + const layer = ProgramAttemptService.layer.pipe( + Layer.provide(services), + Layer.provideMerge(SqlitePersistenceMemory), + Layer.provideMerge(NodeServices.layer), + ); + return { layer, projection, launch, interruptThread }; + }); +} + +it.effect("replays one launch and retains one terminal result until acknowledgement", () => + Effect.gen(function* () { + const harness = yield* makeHarness(); + yield* Effect.gen(function* () { + const attempts = yield* ProgramAttemptService.ProgramAttemptService; + const first = yield* attempts.launch(launchInput); + const replay = yield* attempts.launch(launchInput); + const observedByThread = yield* attempts.observeThread(threadId); + const unrelatedThread = yield* attempts.observeThread(ThreadId.make("thread:unrelated")); + assert.equal(first.threadId, threadId); + assert.equal(first.runId, runId); + assert.equal(first.programId, "agents-dlr"); + assert.equal(first.taskId, "agents-dlr.2"); + assert.equal(first.checkout.startingCommit, "abc123"); + assert.equal(replay.threadId, threadId); + assert.equal(observedByThread?.attemptId, attemptId); + assert.isNull(unrelatedThread); + assert.equal(harness.launch.mock.calls.length, 2); + + yield* Ref.set(harness.projection, makeProjection("completed")); + const terminal = yield* attempts.observe(attemptId); + const terminalReplay = yield* attempts.observe(attemptId); + assert.deepEqual(terminal.terminalResult, terminalReplay.terminalResult); + assert.equal(terminal.terminalResult?.output, "Disposable task finished."); + + const acknowledged = yield* attempts.acknowledge({ + attemptId, + requestId: ProgramAttemptRequestId.make("request:s1:ack"), + }); + const acknowledgementReplay = yield* attempts.acknowledge({ + attemptId, + requestId: ProgramAttemptRequestId.make("request:s1:ack"), + }); + assert.isTrue(acknowledged.terminalAcknowledged); + assert.isNull(acknowledged.terminalResult); + assert.deepEqual(acknowledgementReplay, acknowledged); + }).pipe(Effect.provide(harness.layer)); + }), +); + +it.effect("retains live Program Attempts as restart interruptions before runtime recovery", () => + Effect.gen(function* () { + const harness = yield* makeHarness(); + yield* Effect.gen(function* () { + const attempts = yield* ProgramAttemptService.ProgramAttemptService; + yield* attempts.launch(launchInput); + yield* Ref.set(harness.projection, makeProjection("running")); + + assert.equal(yield* attempts.retainProcessInterruptions, 1); + yield* Ref.set(harness.projection, makeProjection("cancelled")); + + const recovered = yield* attempts.observe(attemptId); + assert.equal(recovered.state, "terminal"); + assert.equal(recovered.terminalResult?.status, "interrupted"); + assert.equal(recovered.terminalResult?.failure?.code, "t3_restart_interrupted"); + assert.isTrue(recovered.terminalResult?.failure?.retryable); + assert.equal(yield* attempts.retainProcessInterruptions, 0); + }).pipe(Effect.provide(harness.layer)); + }), +); + +it.effect("makes repeated cancellation harmless", () => + Effect.gen(function* () { + const harness = yield* makeHarness(); + yield* Effect.gen(function* () { + const attempts = yield* ProgramAttemptService.ProgramAttemptService; + yield* attempts.launch(launchInput); + yield* Ref.set(harness.projection, makeProjection("running")); + const cancel = { + attemptId, + requestId: ProgramAttemptRequestId.make("request:s1:cancel"), + reason: "operator stop", + }; + yield* attempts.cancel(cancel); + yield* attempts.cancel(cancel); + assert.equal(harness.interruptThread.mock.calls.length, 2); + assert.equal( + harness.interruptThread.mock.calls[0]?.[0].commandId, + `program-attempt:${attemptId}:cancel`, + ); + }).pipe(Effect.provide(harness.layer)); + }), +); + +it.effect("rejects cancellation request or payload mismatches", () => + Effect.gen(function* () { + const harness = yield* makeHarness(); + yield* Effect.gen(function* () { + const attempts = yield* ProgramAttemptService.ProgramAttemptService; + yield* attempts.launch(launchInput); + yield* Ref.set(harness.projection, makeProjection("running")); + const cancel = { + attemptId, + requestId: ProgramAttemptRequestId.make("request:s1:cancel:bound"), + reason: "operator stop", + }; + yield* attempts.cancel(cancel); + + const requestConflict = yield* Effect.flip( + attempts.cancel({ + ...cancel, + requestId: ProgramAttemptRequestId.make("request:s1:cancel:other"), + }), + ); + const payloadConflict = yield* Effect.flip( + attempts.cancel({ ...cancel, reason: "different reason" }), + ); + + assert.equal(requestConflict.reason, "request_conflict"); + assert.equal(payloadConflict.reason, "request_conflict"); + assert.equal(harness.interruptThread.mock.calls.length, 1); + }).pipe(Effect.provide(harness.layer)); + }), +); + +it.effect("allows only one of two concurrent cancellation requests to take effect", () => + Effect.gen(function* () { + const harness = yield* makeHarness(); + yield* Effect.gen(function* () { + const attempts = yield* ProgramAttemptService.ProgramAttemptService; + yield* attempts.launch(launchInput); + yield* Ref.set(harness.projection, makeProjection("running")); + + const results = yield* Effect.all( + [ + Effect.exit( + attempts.cancel({ + attemptId, + requestId: ProgramAttemptRequestId.make("request:s1:cancel:concurrent:a"), + reason: "first request", + }), + ), + Effect.exit( + attempts.cancel({ + attemptId, + requestId: ProgramAttemptRequestId.make("request:s1:cancel:concurrent:b"), + reason: "second request", + }), + ), + ], + { concurrency: 2 }, + ); + + assert.equal(results.filter(Exit.isSuccess).length, 1); + assert.equal(harness.interruptThread.mock.calls.length, 1); + }).pipe(Effect.provide(harness.layer)); + }), +); + +it.effect("rejects a different acknowledgement request after binding the first", () => + Effect.gen(function* () { + const harness = yield* makeHarness(); + yield* Effect.gen(function* () { + const attempts = yield* ProgramAttemptService.ProgramAttemptService; + yield* attempts.launch(launchInput); + yield* Ref.set(harness.projection, makeProjection("completed")); + yield* attempts.acknowledge({ + attemptId, + requestId: ProgramAttemptRequestId.make("request:s1:ack:bound"), + }); + + const conflict = yield* Effect.flip( + attempts.acknowledge({ + attemptId, + requestId: ProgramAttemptRequestId.make("request:s1:ack:other"), + }), + ); + + assert.equal(conflict.reason, "request_conflict"); + }).pipe(Effect.provide(harness.layer)); + }), +); diff --git a/apps/server/src/orchestration-v2/ProgramAttemptService.ts b/apps/server/src/orchestration-v2/ProgramAttemptService.ts new file mode 100644 index 00000000000..6ed4cd73552 --- /dev/null +++ b/apps/server/src/orchestration-v2/ProgramAttemptService.ts @@ -0,0 +1,599 @@ +import { + CommandId, + type OrchestrationV2ThreadProjection, + type OrchestrationV2ProviderFailure, + ProgramAttemptId, + ProgramAttemptCancelInput as ProgramAttemptCancelInputSchema, + type ProgramAttemptCancelInput, + ProgramAttemptEffectInput as ProgramAttemptEffectInputSchema, + type ProgramAttemptEffectInput, + ProgramAttemptLaunchInput as ProgramAttemptLaunchInputSchema, + type ProgramAttemptLaunchInput, + type ProgramAttemptSnapshot, + ProgramAttemptTerminalResult as ProgramAttemptTerminalResultSchema, + type ProgramAttemptTerminalResult, + ProjectId, + RunId, + ThreadId, +} from "@t3tools/contracts"; +import * as Context from "effect/Context"; +import * as DateTime from "effect/DateTime"; +import * as Effect from "effect/Effect"; +import * as Exit from "effect/Exit"; +import * as Layer from "effect/Layer"; +import * as Schema from "effect/Schema"; +import * as SqlClient from "effect/unstable/sql/SqlClient"; + +import * as ThreadLaunchService from "./ThreadLaunchService.ts"; +import * as ThreadManagementService from "./ThreadManagementService.ts"; + +interface ProgramAttemptRow { + readonly attempt_id: string; + readonly launch_request_id: string; + readonly launch_input_json: string; + readonly project_id: string; + readonly thread_id: string | null; + readonly run_id: string | null; + readonly cancel_input_json: string | null; + readonly acknowledge_input_json: string | null; + readonly terminal_result_json: string | null; + readonly terminal_acknowledged_at: string | null; + readonly created_at: string; + readonly updated_at: string; +} + +export class ProgramAttemptError extends Schema.TaggedErrorClass()( + "ProgramAttemptError", + { + reason: Schema.Literals([ + "not_found", + "request_conflict", + "launch_incomplete", + "run_missing", + "not_terminal", + "persistence_failed", + "launch_failed", + "projection_failed", + "cancel_failed", + "invalid_record", + ]), + attemptId: ProgramAttemptId, + detail: Schema.String, + cause: Schema.optional(Schema.Defect()), + }, +) {} + +export class ProgramAttemptService extends Context.Service< + ProgramAttemptService, + { + readonly launch: ( + input: ProgramAttemptLaunchInput, + ) => Effect.Effect; + readonly observe: ( + attemptId: ProgramAttemptId, + ) => Effect.Effect; + readonly observeThread: ( + threadId: ThreadId, + ) => Effect.Effect; + readonly cancel: ( + input: ProgramAttemptCancelInput, + ) => Effect.Effect; + readonly acknowledge: ( + input: ProgramAttemptEffectInput, + ) => Effect.Effect; + readonly retainProcessInterruptions: Effect.Effect; + } +>()("t3/orchestration-v2/ProgramAttemptService") {} + +const decodeTerminalResult = Schema.decodeUnknownEffect( + Schema.fromJsonString(ProgramAttemptTerminalResultSchema), +); +const encodeTerminalResult = Schema.encodeEffect( + Schema.fromJsonString(ProgramAttemptTerminalResultSchema), +); +const encodeLaunchInput = Schema.encodeEffect( + Schema.fromJsonString(ProgramAttemptLaunchInputSchema), +); +const decodeLaunchInput = Schema.decodeUnknownEffect( + Schema.fromJsonString(ProgramAttemptLaunchInputSchema), +); +const encodeCancelInput = Schema.encodeEffect( + Schema.fromJsonString(ProgramAttemptCancelInputSchema), +); +const encodeAcknowledgeInput = Schema.encodeEffect( + Schema.fromJsonString(ProgramAttemptEffectInputSchema), +); + +function error( + attemptId: ProgramAttemptId, + reason: ProgramAttemptError["reason"], + detail: string, + cause?: unknown, +) { + return new ProgramAttemptError({ + attemptId, + reason, + detail, + ...(cause === undefined ? {} : { cause }), + }); +} + +function terminalResult( + projection: OrchestrationV2ThreadProjection, + runId: RunId, +): ProgramAttemptTerminalResult { + const run = projection.runs.find((candidate) => candidate.id === runId); + if (run === undefined || !ThreadManagementService.isTerminalRunStatus(run.status)) { + throw new Error(`Run ${runId} is not terminal.`); + } + const items = projection.turnItems.filter((item) => item.runId === runId); + const output = items + .filter((item) => item.type === "assistant_message" && item.status === "completed") + .toSorted((left, right) => right.ordinal - left.ordinal)[0]; + const failure = items + .filter((item) => item.type === "error") + .toSorted((left, right) => right.ordinal - left.ordinal)[0]; + return { + status: run.status, + output: output?.type === "assistant_message" ? output.text : null, + failure: + failure?.type === "error" ? (failure.failure satisfies OrchestrationV2ProviderFailure) : null, + completedAt: run.completedAt === null ? null : DateTime.formatIso(run.completedAt), + }; +} + +export const layer = Layer.effect( + ProgramAttemptService, + Effect.gen(function* () { + const sql = yield* SqlClient.SqlClient; + const launches = yield* ThreadLaunchService.ThreadLaunchService; + const threads = yield* ThreadManagementService.ThreadManagementService; + + const now = DateTime.now.pipe(Effect.map((value) => DateTime.formatIso(value))); + + const load = Effect.fn("ProgramAttemptService.load")(function* (attemptId: ProgramAttemptId) { + const rows = yield* sql` + SELECT * FROM program_attempts WHERE attempt_id = ${attemptId} + `.pipe( + Effect.mapError((cause) => + error(attemptId, "persistence_failed", "Could not load the Program Attempt.", cause), + ), + ); + const row = rows[0]; + if (row === undefined) { + return yield* error(attemptId, "not_found", `Program Attempt ${attemptId} was not found.`); + } + return row; + }); + + const persistTerminal = Effect.fn("ProgramAttemptService.persistTerminal")(function* ( + row: ProgramAttemptRow, + result: ProgramAttemptTerminalResult, + ) { + const updatedAt = yield* now; + const encoded = yield* encodeTerminalResult(result).pipe( + Effect.mapError((cause) => + error( + ProgramAttemptId.make(row.attempt_id), + "invalid_record", + "Could not encode the terminal result.", + cause, + ), + ), + ); + yield* sql` + UPDATE program_attempts + SET terminal_result_json = COALESCE(terminal_result_json, ${encoded}), + updated_at = ${updatedAt} + WHERE attempt_id = ${row.attempt_id} + `.pipe( + Effect.mapError((cause) => + error( + ProgramAttemptId.make(row.attempt_id), + "persistence_failed", + "Could not retain the terminal result.", + cause, + ), + ), + ); + return yield* load(ProgramAttemptId.make(row.attempt_id)); + }); + + const snapshot = Effect.fn("ProgramAttemptService.snapshot")(function* ( + initialRow: ProgramAttemptRow, + ) { + const attemptId = ProgramAttemptId.make(initialRow.attempt_id); + const launchInput = yield* decodeLaunchInput(initialRow.launch_input_json).pipe( + Effect.mapError((cause) => + error(attemptId, "invalid_record", "The retained launch request is invalid.", cause), + ), + ); + if (initialRow.thread_id === null || initialRow.run_id === null) { + return yield* error( + attemptId, + "launch_incomplete", + "The launch intent exists but the thread and run receipt are not recorded yet.", + ); + } + const threadId = ThreadId.make(initialRow.thread_id); + const runId = RunId.make(initialRow.run_id); + const projection = yield* threads + .getThreadProjection(threadId) + .pipe( + Effect.mapError((cause) => + error(attemptId, "projection_failed", "Could not load the Attempt thread.", cause), + ), + ); + const run = projection.runs.find((candidate) => candidate.id === runId); + if (run === undefined) { + return yield* error(attemptId, "run_missing", `Run ${runId} is missing from the thread.`); + } + let row = initialRow; + if ( + ThreadManagementService.isTerminalRunStatus(run.status) && + row.terminal_result_json === null + ) { + row = yield* persistTerminal(row, terminalResult(projection, runId)); + } + const retained = + row.terminal_result_json === null + ? null + : yield* decodeTerminalResult(row.terminal_result_json).pipe( + Effect.mapError((cause) => + error( + attemptId, + "invalid_record", + "The retained terminal result is invalid.", + cause, + ), + ), + ); + return { + attemptId, + programId: launchInput.programId ?? null, + taskId: launchInput.taskId ?? null, + attemptKind: launchInput.attemptKind ?? null, + candidateId: launchInput.candidateId ?? null, + reviewId: launchInput.reviewId ?? null, + reviewKind: launchInput.reviewKind ?? null, + title: launchInput.title, + checkout: launchInput.checkout, + projectId: ProjectId.make(row.project_id), + threadId, + runId, + state: retained !== null ? "terminal" : run.status === "preparing" ? "preparing" : "active", + runStatus: run.status, + terminalResult: + row.terminal_acknowledged_at === null + ? (retained as ProgramAttemptTerminalResult | null) + : null, + terminalAcknowledged: row.terminal_acknowledged_at !== null, + } satisfies ProgramAttemptSnapshot; + }); + + const observe: ProgramAttemptService["Service"]["observe"] = Effect.fn( + "ProgramAttemptService.observe", + )(function* (attemptId) { + return yield* snapshot(yield* load(attemptId)); + }); + + const observeThread: ProgramAttemptService["Service"]["observeThread"] = Effect.fn( + "ProgramAttemptService.observeThread", + )(function* (threadId) { + const lookupId = ProgramAttemptId.make(`program-attempt:thread:${threadId}`); + const visible = yield* threads + .getThreadProjection(threadId) + .pipe(Effect.exit, Effect.map(Exit.isSuccess)); + if (!visible) return null; + return yield* retryProgramAttemptReceipt(() => + sql` + SELECT * FROM program_attempts + WHERE thread_id = ${threadId} + ORDER BY created_at DESC + LIMIT 1 + `.pipe( + Effect.mapError((cause) => + error( + lookupId, + "persistence_failed", + "Could not load the Program Attempt for this thread.", + cause, + ), + ), + Effect.flatMap((rows) => + rows[0] === undefined ? Effect.succeed(null) : snapshot(rows[0]), + ), + ), + ); + }); + + const retainProcessInterruptions: ProgramAttemptService["Service"]["retainProcessInterruptions"] = + Effect.gen(function* () { + const recoveryId = ProgramAttemptId.make("program-attempt:process-recovery"); + const rows = yield* sql` + SELECT * FROM program_attempts + WHERE terminal_result_json IS NULL AND thread_id IS NOT NULL AND run_id IS NOT NULL + `.pipe( + Effect.mapError((cause) => + error(recoveryId, "persistence_failed", "Could not load live Program Attempts.", cause), + ), + ); + let retained = 0; + for (const row of rows) { + const projection = yield* threads + .getThreadProjection(ThreadId.make(row.thread_id!)) + .pipe( + Effect.mapError((cause) => + error( + ProgramAttemptId.make(row.attempt_id), + "projection_failed", + "Could not load a live Program Attempt before process recovery.", + cause, + ), + ), + ); + const run = projection.runs.find((candidate) => candidate.id === row.run_id); + if (run === undefined || ThreadManagementService.isTerminalRunStatus(run.status)) + continue; + const completedAt = yield* now; + yield* persistTerminal(row, { + status: "interrupted", + output: null, + failure: { + class: "transport_error", + message: "T3 restarted before the Program Attempt completed.", + code: "t3_restart_interrupted", + retryable: true, + }, + completedAt, + }); + retained += 1; + } + return retained; + }); + + const launch: ProgramAttemptService["Service"]["launch"] = Effect.fn( + "ProgramAttemptService.launch", + )(function* (input) { + const inputJson = yield* encodeLaunchInput(input).pipe( + Effect.mapError((cause) => + error(input.attemptId, "invalid_record", "Could not encode the launch request.", cause), + ), + ); + const timestamp = yield* now; + yield* sql` + INSERT INTO program_attempts ( + attempt_id, launch_request_id, launch_input_json, + project_id, created_at, updated_at + ) VALUES ( + ${input.attemptId}, ${input.requestId}, ${inputJson}, + ${input.projectId}, ${timestamp}, ${timestamp} + ) ON CONFLICT(attempt_id) DO NOTHING + `.pipe( + Effect.mapError((cause) => + error( + input.attemptId, + "persistence_failed", + "Could not persist the launch intent.", + cause, + ), + ), + ); + const row = yield* load(input.attemptId); + if (row.launch_input_json !== inputJson) { + return yield* error( + input.attemptId, + "request_conflict", + "This Attempt ID is already bound to a different launch request.", + ); + } + const launched = yield* launches + .launch({ + commandId: CommandId.make(`program-attempt:${input.attemptId}:launch`), + projectId: input.projectId, + title: input.title, + generateTitle: false, + modelSelection: input.providerPolicy.modelSelection, + runtimeMode: input.providerPolicy.runtimeMode, + interactionMode: input.providerPolicy.interactionMode, + workspaceStrategy: { type: "prepared_worktree", ...input.checkout }, + initialMessage: { text: input.prompt, attachments: [] }, + createdBy: "system", + creationSource: "server", + }) + .pipe( + Effect.mapError((cause) => + error( + input.attemptId, + "launch_failed", + "T3 could not launch the Program Attempt.", + cause, + ), + ), + ); + const run = launched.projection.runs.toSorted( + (left, right) => right.ordinal - left.ordinal, + )[0]; + if (run === undefined) { + return yield* error( + input.attemptId, + "run_missing", + "T3 accepted the Program Attempt without a durable run.", + ); + } + const updatedAt = yield* now; + yield* sql` + UPDATE program_attempts + SET thread_id = COALESCE(thread_id, ${launched.threadId}), + run_id = COALESCE(run_id, ${run.id}), + updated_at = ${updatedAt} + WHERE attempt_id = ${input.attemptId} + `.pipe( + Effect.mapError((cause) => + error( + input.attemptId, + "persistence_failed", + "Could not persist the launch receipt.", + cause, + ), + ), + ); + const persisted = yield* load(input.attemptId); + if (persisted.thread_id !== launched.threadId || persisted.run_id !== run.id) { + return yield* error( + input.attemptId, + "invalid_record", + "The durable launch receipt does not match T3's idempotent launch receipt.", + ); + } + return yield* snapshot(persisted); + }); + + const bindEffectInput = Effect.fn("ProgramAttemptService.bindEffectInput")(function* ( + attemptId: ProgramAttemptId, + column: "cancel_input_json" | "acknowledge_input_json", + inputJson: string, + ) { + const updatedAt = yield* now; + const query = + column === "cancel_input_json" + ? sql` + UPDATE program_attempts + SET cancel_input_json = COALESCE(cancel_input_json, ${inputJson}), updated_at = ${updatedAt} + WHERE attempt_id = ${attemptId} + ` + : sql` + UPDATE program_attempts + SET acknowledge_input_json = COALESCE(acknowledge_input_json, ${inputJson}), updated_at = ${updatedAt} + WHERE attempt_id = ${attemptId} + `; + yield* query.pipe( + Effect.mapError((cause) => + error(attemptId, "persistence_failed", "Could not persist the effect intent.", cause), + ), + ); + const row = yield* load(attemptId); + const bound = + column === "cancel_input_json" ? row.cancel_input_json : row.acknowledge_input_json; + if (bound !== inputJson) { + return yield* error( + attemptId, + "request_conflict", + "This Attempt effect is already bound to a different request.", + ); + } + return row; + }); + + const cancel: ProgramAttemptService["Service"]["cancel"] = Effect.fn( + "ProgramAttemptService.cancel", + )(function* (input) { + let row = yield* load(input.attemptId); + if (row.thread_id === null || row.run_id === null) { + return yield* error( + input.attemptId, + "launch_incomplete", + "The Attempt has no run to cancel.", + ); + } + const inputJson = yield* encodeCancelInput(input).pipe( + Effect.mapError((cause) => + error(input.attemptId, "invalid_record", "Could not encode the cancel request.", cause), + ), + ); + row = yield* bindEffectInput(input.attemptId, "cancel_input_json", inputJson); + if (row.thread_id === null || row.run_id === null) { + return yield* error( + input.attemptId, + "launch_incomplete", + "The Attempt has no run to cancel.", + ); + } + yield* threads + .interruptThread({ + projectId: ProjectId.make(row.project_id), + commandId: CommandId.make(`program-attempt:${input.attemptId}:cancel`), + threadId: ThreadId.make(row.thread_id), + runId: RunId.make(row.run_id), + ...(input.reason === undefined ? {} : { reason: input.reason }), + }) + .pipe( + Effect.mapError((cause) => + error( + input.attemptId, + "cancel_failed", + "T3 could not cancel the Program Attempt.", + cause, + ), + ), + ); + return yield* snapshot(yield* load(input.attemptId)); + }); + + const acknowledge: ProgramAttemptService["Service"]["acknowledge"] = Effect.fn( + "ProgramAttemptService.acknowledge", + )(function* (input) { + let row = yield* load(input.attemptId); + const before = yield* snapshot(row); + if (before.state !== "terminal") { + return yield* error( + input.attemptId, + "not_terminal", + "A Program Attempt can be acknowledged only after it reaches a terminal state.", + ); + } + const inputJson = yield* encodeAcknowledgeInput(input).pipe( + Effect.mapError((cause) => + error( + input.attemptId, + "invalid_record", + "Could not encode the acknowledgement request.", + cause, + ), + ), + ); + row = yield* bindEffectInput(input.attemptId, "acknowledge_input_json", inputJson); + const acknowledgedAt = yield* now; + yield* sql` + UPDATE program_attempts + SET terminal_acknowledged_at = COALESCE(terminal_acknowledged_at, ${acknowledgedAt}), + updated_at = ${acknowledgedAt} + WHERE attempt_id = ${input.attemptId} + `.pipe( + Effect.mapError((cause) => + error( + input.attemptId, + "persistence_failed", + "Could not acknowledge the terminal result.", + cause, + ), + ), + ); + return yield* snapshot(yield* load(input.attemptId)); + }); + + return ProgramAttemptService.of({ + launch, + observe, + observeThread, + cancel, + acknowledge, + retainProcessInterruptions, + }); + }), +); +export function retryProgramAttemptReceipt( + lookup: () => Effect.Effect, + options: { readonly attempts?: number; readonly delay?: Effect.Effect } = {}, +): Effect.Effect { + const attempts = options.attempts ?? 12; + const delay = options.delay ?? Effect.sleep("250 millis"); + return Effect.gen(function* () { + for (let index = 0; index < attempts; index += 1) { + const result = yield* lookup(); + if (result !== null || index === attempts - 1) return result; + yield* delay; + } + return null; + }); +} diff --git a/apps/server/src/orchestration-v2/ProviderTurnStartService.test.ts b/apps/server/src/orchestration-v2/ProviderTurnStartService.test.ts index 95c9a03187a..4b54f5cef53 100644 --- a/apps/server/src/orchestration-v2/ProviderTurnStartService.test.ts +++ b/apps/server/src/orchestration-v2/ProviderTurnStartService.test.ts @@ -101,3 +101,74 @@ it("does not commit running state when inherited background routing cannot be re expect(startRootRun).not.toHaveBeenCalled(); }).pipe(Effect.provide(layer), Effect.runPromise); }); + +it("terminalizes a starting run when provider startup exhausts its retries", async () => { + const threadId = ThreadId.make("thread_provider_turn_start_exhausted"); + const runId = RunId.make("run_provider_turn_start_exhausted"); + const attemptId = RunAttemptId.make("attempt_provider_turn_start_exhausted"); + const rootNodeId = NodeId.make("node_provider_turn_start_exhausted"); + const providerThreadId = ProviderThreadId.make("provider_thread_provider_turn_start_exhausted"); + const projection = { + thread: { id: threadId }, + runs: [ + { + id: runId, + status: "starting", + rootNodeId, + activeAttemptId: attemptId, + providerThreadId, + providerInstanceId: "codex", + }, + ], + nodes: [{ id: rootNodeId, status: "pending" }], + attempts: [{ id: attemptId, status: "pending" }], + providerThreads: [{ id: providerThreadId, driver: "codex" }], + turnItems: [], + } as unknown as OrchestrationV2ThreadProjection; + type WriteInput = Parameters[0]; + let written: WriteInput | undefined; + let writeCount = 0; + const writeIfRunCurrent: EventSink.EventSinkV2Shape["writeIfRunCurrent"] = (input) => { + written = input; + writeCount += 1; + return Effect.succeed({ committed: true, storedEvents: [] }); + }; + const layer = ProviderTurnStart.layer.pipe( + Layer.provide( + Layer.mergeAll( + Layer.mock(ContextHandoffService.ContextHandoffServiceV2)({}), + Layer.mock(EventSink.EventSinkV2)({ writeIfRunCurrent }), + IdAllocator.layer, + Layer.mock(ProjectionStore.ProjectionStoreV2)({ + getThreadProjection: () => Effect.succeed(projection), + }), + Layer.mock(ProviderSessionManager.ProviderSessionManagerV2)({}), + Layer.mock(RunExecutionService.RunExecutionServiceV2)({}), + Layer.mock(RuntimePolicy.RuntimePolicyV2)({}), + ), + ), + ); + + await ProviderTurnStart.ProviderTurnStartServiceV2.pipe( + Effect.flatMap((service) => + service.fail({ threadId, runId, cause: "invalid provider configuration" }), + ), + Effect.provide(layer), + Effect.runPromise, + ); + + expect(writeCount).toBe(1); + const input = written as WriteInput; + expect(input?.expectedStatus).toBe("starting"); + expect( + input.events.map((event) => [ + event.type, + (event.payload as { readonly status?: string }).status, + ]), + ).toEqual([ + ["turn-item.updated", "failed"], + ["run-attempt.updated", "failed"], + ["node.updated", "failed"], + ["run.updated", "failed"], + ]); +}); diff --git a/apps/server/src/orchestration-v2/ProviderTurnStartService.ts b/apps/server/src/orchestration-v2/ProviderTurnStartService.ts index 15f7a537436..3e6a6872215 100644 --- a/apps/server/src/orchestration-v2/ProviderTurnStartService.ts +++ b/apps/server/src/orchestration-v2/ProviderTurnStartService.ts @@ -21,6 +21,7 @@ import { } from "./ContextHandoffService.ts"; import { IdAllocatorV2 } from "./IdAllocator.ts"; import { ProjectionStoreV2 } from "./ProjectionStore.ts"; +import { makeProviderFailure, makeProviderFailureTurnItem } from "./ProviderFailure.ts"; import { ProviderSessionManagerV2 } from "./ProviderSessionManager.ts"; import { canRouteRelatedSubagent, @@ -44,6 +45,11 @@ export interface ProviderTurnStartServiceV2Shape { readonly threadId: ThreadId; readonly runId: RunId; }) => Effect.Effect; + readonly fail: (input: { + readonly threadId: ThreadId; + readonly runId: RunId; + readonly cause: unknown; + }) => Effect.Effect; } export class ProviderTurnStartServiceV2 extends Context.Service< @@ -72,6 +78,100 @@ export const layer: Layer.Layer< const runExecution = yield* RunExecutionServiceV2; const runtimePolicy = yield* RuntimePolicyV2; + const fail = Effect.fn("orchestrationV2.providerTurnStart.fail")(function* (input: { + readonly threadId: ThreadId; + readonly runId: RunId; + readonly cause: unknown; + }) { + const projection = yield* projectionStore.getThreadProjection(input.threadId); + const run = projection.runs.find((candidate) => candidate.id === input.runId); + if (run === undefined || run.status !== "starting") return; + const attempt = projection.attempts.find((candidate) => candidate.id === run.activeAttemptId); + const rootNode = projection.nodes.find((candidate) => candidate.id === run.rootNodeId); + const providerThread = projection.providerThreads.find( + (candidate) => candidate.id === run.providerThreadId, + ); + if (attempt === undefined || rootNode === undefined || providerThread === undefined) { + return yield* new ProviderTurnStartError({ + runId: input.runId, + cause: `Run ${input.runId} is missing its failure projection state.`, + }); + } + const occurredAt = yield* DateTime.now; + const providerTurnId = idAllocator.derive.providerTurn({ + driver: providerThread.driver, + nativeTurnId: `failed:${attempt.id}`, + }); + const failure = makeProviderFailure({ + cause: input.cause, + code: "provider_start_failed", + class: "provider_error", + retryable: false, + }); + const failureItem = makeProviderFailureTurnItem({ + idAllocator, + driver: providerThread.driver, + threadId: projection.thread.id, + runId: run.id, + nodeId: rootNode.id, + providerThreadId: providerThread.id, + providerTurnId, + itemOrdinal: Math.max(0, ...projection.turnItems.map((item) => item.ordinal)) + 1, + failure, + occurredAt, + }); + const events: Array = [ + { + id: yield* idAllocator.allocate.event({ threadId: projection.thread.id }), + type: "turn-item.updated", + threadId: projection.thread.id, + runId: run.id, + nodeId: rootNode.id, + driver: providerThread.driver, + providerInstanceId: run.providerInstanceId, + occurredAt, + payload: failureItem, + }, + { + id: yield* idAllocator.allocate.event({ threadId: projection.thread.id }), + type: "run-attempt.updated", + threadId: projection.thread.id, + runId: run.id, + nodeId: rootNode.id, + providerInstanceId: run.providerInstanceId, + occurredAt, + payload: { ...attempt, status: "failed", completedAt: occurredAt }, + }, + { + id: yield* idAllocator.allocate.event({ threadId: projection.thread.id }), + type: "node.updated", + threadId: projection.thread.id, + runId: run.id, + nodeId: rootNode.id, + providerInstanceId: run.providerInstanceId, + occurredAt, + payload: { ...rootNode, status: "failed", completedAt: occurredAt }, + }, + { + id: yield* idAllocator.allocate.event({ threadId: projection.thread.id }), + type: "run.updated", + threadId: projection.thread.id, + runId: run.id, + nodeId: rootNode.id, + providerInstanceId: run.providerInstanceId, + occurredAt, + payload: { ...run, status: "failed", queuePosition: null, completedAt: occurredAt }, + }, + ]; + yield* eventSink.writeIfRunCurrent({ + threadId: projection.thread.id, + runId: run.id, + activeAttemptId: attempt.id, + expectedStatus: "starting", + events, + }); + }); + const start = Effect.fn("orchestrationV2.providerTurnStart.start")(function* (input: { readonly threadId: ThreadId; readonly runId: RunId; @@ -500,6 +600,14 @@ export const layer: Layer.Layer< }); return ProviderTurnStartServiceV2.of({ + fail: (input) => + fail(input).pipe( + Effect.mapError((cause) => + isProviderTurnStartError(cause) + ? cause + : new ProviderTurnStartError({ runId: input.runId, cause }), + ), + ), start: (input) => start(input).pipe( Effect.mapError((cause) => diff --git a/apps/server/src/orchestration-v2/ThreadLaunchService.test.ts b/apps/server/src/orchestration-v2/ThreadLaunchService.test.ts index 7f8b86b80c2..d43707c00e7 100644 --- a/apps/server/src/orchestration-v2/ThreadLaunchService.test.ts +++ b/apps/server/src/orchestration-v2/ThreadLaunchService.test.ts @@ -36,6 +36,7 @@ import type { ProviderAdapterV2Shape } from "./ProviderAdapter.ts"; import * as ProviderAdapterRegistry from "./ProviderAdapterRegistry.ts"; import * as ThreadLaunch from "./ThreadLaunchService.ts"; import * as ThreadManagement from "./ThreadManagementService.ts"; +import * as PreparedWorktreeVerifier from "./PreparedWorktreeVerifier.ts"; import * as ThreadTitleRegeneration from "./ThreadTitleRegenerationService.ts"; import { makeOrchestratorV2ReplayLayerWithRegistry } from "./testkit/ProviderReplayHarness.ts"; @@ -74,6 +75,7 @@ interface HarnessOptions { readonly generateBranchName?: TextGeneration.TextGeneration["Service"]["generateBranchName"]; readonly serverSettings?: Parameters[0]; readonly providers?: ReadonlyArray; + readonly verifyPrepared?: PreparedWorktreeVerifier.PreparedWorktreeVerifier["Service"]["verify"]; } function makeHarness(options: HarnessOptions = {}) { @@ -106,6 +108,17 @@ function makeHarness(options: HarnessOptions = {}) { const generateThreadTitle = vi.fn( options.generateTitle ?? (() => Effect.succeed({ title: "Generated title" })), ); + const verifyPrepared = vi.fn( + options.verifyPrepared ?? + ((checkout) => + Effect.succeed({ + repositoryRoot: checkout.repositoryRoot, + gitCommonDir: checkout.gitCommonDir, + worktreePath: checkout.worktreePath, + branch: checkout.branch, + startingCommit: checkout.startingCommit, + })), + ); const externalServices = Layer.mergeAll( Layer.succeed(ProjectService.ProjectService, { create: () => Effect.die("unused"), @@ -133,6 +146,7 @@ function makeHarness(options: HarnessOptions = {}) { }), ServerSettings.layerTest(options.serverSettings), makeProviderRegistryLayer(options.providers), + Layer.succeed(PreparedWorktreeVerifier.PreparedWorktreeVerifier, { verify: verifyPrepared }), ); const launch = ThreadLaunch.layer.pipe( Layer.provide(Layer.mergeAll(externalServices, threadManagement, receipts, IdAllocator.layer)), @@ -165,6 +179,7 @@ function makeHarness(options: HarnessOptions = {}) { generateBranchName, generateThreadTitle, runSetup, + verifyPrepared, }; } @@ -897,6 +912,84 @@ it.effect("renames a temporary branch on an existing worktree to a generated nam }), ); +it.effect("verifies a prepared worktree before release and skips setup and branch rename", () => + Effect.gen(function* () { + const verificationEntered = yield* Deferred.make(); + const allowVerification = yield* Deferred.make(); + const harness = makeHarness({ + verifyPrepared: (checkout) => + Deferred.succeed(verificationEntered, undefined).pipe( + Effect.andThen(Deferred.await(allowVerification)), + Effect.as({ + repositoryRoot: checkout.repositoryRoot, + gitCommonDir: checkout.gitCommonDir, + worktreePath: checkout.worktreePath, + branch: checkout.branch, + startingCommit: checkout.startingCommit, + }), + ), + }); + yield* Effect.gen(function* () { + const launches = yield* ThreadLaunch.ThreadLaunchService; + const threads = yield* ThreadManagement.ThreadManagementService; + const workspace = { + type: "prepared_worktree" as const, + repositoryRoot: "/repo", + gitCommonDir: "/repo/.git", + worktreePath: "/repo-worktrees/prepared", + branch: "t3code/prepared", + startingCommit: "abc123", + }; + const launched = yield* launches.launch( + launchInput({ + command: "command:launch:prepared-worktree", + thread: "thread:launch:prepared-worktree", + message: "Build the feature", + workspace, + }), + ); + yield* Deferred.await(verificationEntered); + assert.equal( + (yield* threads.getThreadProjection(launched.threadId)).runs[0]?.status, + "preparing", + ); + assert.equal(harness.runSetup.mock.calls.length, 0); + assert.equal(harness.generateBranchName.mock.calls.length, 0); + assert.equal(harness.renameBranch.mock.calls.length, 0); + assert.deepEqual(harness.verifyPrepared.mock.calls[0]?.[0], workspace); + yield* Deferred.succeed(allowVerification, undefined); + yield* waitUntil(() => + threads + .getThreadProjection(launched.threadId) + .pipe(Effect.map((projection) => projection.runs[0]?.status === "starting")), + ); + }).pipe(Effect.provide(harness.layer)); + }), +); + +it.effect("reuses the receipt thread when an idempotent launch did not supply a thread id", () => + Effect.gen(function* () { + const harness = makeHarness(); + yield* Effect.gen(function* () { + const launches = yield* ThreadLaunch.ThreadLaunchService; + const input = launchInput({ + command: "command:launch:receipt-thread", + thread: "unused-explicit-thread", + message: "Build the feature", + workspace: { type: "root" }, + }); + const { threadId: _unused, ...withoutThreadId } = input; + + const first = yield* launches.launch(withoutThreadId); + const replay = yield* launches.launch(withoutThreadId); + + assert.equal(replay.threadId, first.threadId); + assert.equal(replay.projection.runs[0]?.id, first.projection.runs[0]?.id); + assert.isTrue(replay.resumed); + }).pipe(Effect.provide(harness.layer)); + }), +); + for (const failurePoint of ["worktree", "setup"] as const) { it.effect( `${failurePoint} failure keeps the thread and message visible and emits failure items`, diff --git a/apps/server/src/orchestration-v2/ThreadLaunchService.ts b/apps/server/src/orchestration-v2/ThreadLaunchService.ts index 97f9daba388..7a5a63386be 100644 --- a/apps/server/src/orchestration-v2/ThreadLaunchService.ts +++ b/apps/server/src/orchestration-v2/ThreadLaunchService.ts @@ -7,6 +7,7 @@ import { type OrchestrationV2CreationSource, type OrchestrationV2ThreadProjection, type ProviderInteractionMode, + type PreparedWorktreeCheckout, ProjectId, type RunId, type RuntimeMode, @@ -34,6 +35,7 @@ import * as IdAllocator from "./IdAllocator.ts"; import { makeProviderFailure } from "./ProviderFailure.ts"; import { randomUuidV4 } from "./RandomUuid.ts"; import * as ThreadManagement from "./ThreadManagementService.ts"; +import * as PreparedWorktreeVerifier from "./PreparedWorktreeVerifier.ts"; export type ThreadLaunchWorkspaceStrategy = | { readonly type: "root"; readonly branch?: string | undefined } @@ -42,6 +44,7 @@ export type ThreadLaunchWorkspaceStrategy = readonly worktreePath: string; readonly branch?: string | undefined; } + | ({ readonly type: "prepared_worktree" } & PreparedWorktreeCheckout) | { readonly type: "worktree"; readonly baseRef: string; @@ -91,6 +94,7 @@ export class ThreadLaunchError extends Schema.TaggedErrorClass>(new Set()); yield* Effect.addFinalizer(() => Scope.close(preparationScope, Exit.void)); @@ -224,7 +229,8 @@ export const make = Effect.gen(function* () { branch = requestedBranch ?? null; } let worktreePath = - input.workspaceStrategy.type === "existing_worktree" + input.workspaceStrategy.type === "existing_worktree" || + input.workspaceStrategy.type === "prepared_worktree" ? input.workspaceStrategy.worktreePath : null; if (input.workspaceStrategy.type === "worktree") { @@ -286,7 +292,8 @@ export const make = Effect.gen(function* () { worktreePath !== null && branch !== null && initialMessage !== undefined && - isTemporaryWorktreeBranch(branch) + isTemporaryWorktreeBranch(branch) && + input.workspaceStrategy.type !== "prepared_worktree" ) { const oldBranch = branch; const worktreeCwd = worktreePath; @@ -321,22 +328,28 @@ export const make = Effect.gen(function* () { commandId: CommandId.make(`${input.commandId}:progress:setup`), threadId, runId, - phase: "setup", + phase: input.workspaceStrategy.type === "prepared_worktree" ? "verification" : "setup", }) .pipe(Effect.mapError(mapError(input, "update-thread", threadId))); } - yield* setupScripts - .runForThread({ - threadId, - projectId: input.projectId, - projectCwd: project.workspaceRoot, - worktreePath: cwd, - project: { - workspaceRoot: project.workspaceRoot, - scripts: project.scripts, - }, - }) - .pipe(Effect.mapError(mapError(input, "run-setup-script", threadId))); + if (input.workspaceStrategy.type === "prepared_worktree") { + yield* preparedWorktrees + .verify(input.workspaceStrategy, project.workspaceRoot) + .pipe(Effect.mapError(mapError(input, "verify-prepared-worktree", threadId))); + } else { + yield* setupScripts + .runForThread({ + threadId, + projectId: input.projectId, + projectCwd: project.workspaceRoot, + worktreePath: cwd, + project: { + workspaceRoot: project.workspaceRoot, + scripts: project.scripts, + }, + }) + .pipe(Effect.mapError(mapError(input, "run-setup-script", threadId))); + } if (runId !== null) { yield* threads @@ -440,9 +453,11 @@ export const make = Effect.gen(function* () { return yield* Effect.gen(function* () { const candidateThreadId = input.threadId ?? - (yield* ids.allocate - .thread({ projectId: input.projectId }) - .pipe(Effect.mapError(mapError(input, "create-thread")))); + (Option.isSome(launchReceipt) + ? launchReceipt.value.threadId + : yield* ids.allocate + .thread({ projectId: input.projectId }) + .pipe(Effect.mapError(mapError(input, "create-thread")))); if (input.reuseExistingThread === true && Option.isNone(launchReceipt)) { yield* validateReusableThread(input, candidateThreadId); @@ -450,7 +465,8 @@ export const make = Effect.gen(function* () { const initialBranch = input.workspaceStrategy.branch ?? null; const initialWorktreePath = - input.workspaceStrategy.type === "existing_worktree" + input.workspaceStrategy.type === "existing_worktree" || + input.workspaceStrategy.type === "prepared_worktree" ? input.workspaceStrategy.worktreePath : null; const claimDispatch = diff --git a/apps/server/src/orchestration-v2/applicationLayer.ts b/apps/server/src/orchestration-v2/applicationLayer.ts index 2a174a316f8..072aaa95703 100644 --- a/apps/server/src/orchestration-v2/applicationLayer.ts +++ b/apps/server/src/orchestration-v2/applicationLayer.ts @@ -3,6 +3,7 @@ import * as Layer from "effect/Layer"; import { ProjectionProjectRepositoryLive } from "../persistence/Layers/ProjectionProjects.ts"; import { layer as projectServiceLayer } from "../project/ProjectService.ts"; import { layer as threadLaunchServiceLayer } from "./ThreadLaunchService.ts"; +import { layer as preparedWorktreeVerifierLayer } from "./PreparedWorktreeVerifier.ts"; import { layer as threadLifecycleServiceLayer } from "./ThreadLifecycleService.ts"; import { live as resourceCleanupLive } from "./ResourceCleanupService.ts"; import { observerLive as runFinalizationObserverLive } from "./RunFinalizationService.ts"; @@ -12,7 +13,7 @@ const projectServiceProvided = projectServiceLayer.pipe( ); const applicationServices = Layer.mergeAll( - threadLaunchServiceLayer, + threadLaunchServiceLayer.pipe(Layer.provide(preparedWorktreeVerifierLayer)), threadLifecycleServiceLayer, ).pipe(Layer.provideMerge(projectServiceProvided)); diff --git a/apps/server/src/orchestration-v2/programAttemptHttp.ts b/apps/server/src/orchestration-v2/programAttemptHttp.ts new file mode 100644 index 00000000000..0c0ec76144b --- /dev/null +++ b/apps/server/src/orchestration-v2/programAttemptHttp.ts @@ -0,0 +1,94 @@ +import { + AuthOrchestrationOperateScope, + AuthOrchestrationReadScope, + EnvironmentHttpApi, + EnvironmentHttpBadRequestError, + EnvironmentHttpConflictError, +} from "@t3tools/contracts"; +import * as Effect from "effect/Effect"; +import * as HttpApiBuilder from "effect/unstable/httpapi/HttpApiBuilder"; + +import { + annotateEnvironmentRequest, + failEnvironmentInternal, + failEnvironmentNotFound, + requireEnvironmentScope, +} from "../auth/http.ts"; +import * as ProgramAttemptService from "./ProgramAttemptService.ts"; + +const mapProgramAttemptError = Effect.fn("programAttemptHttp.mapError")(function* ( + error: ProgramAttemptService.ProgramAttemptError, +) { + switch (error.reason) { + case "not_found": + return yield* failEnvironmentNotFound("program_attempt_not_found"); + case "request_conflict": + return yield* new EnvironmentHttpConflictError({ message: error.detail }); + case "launch_incomplete": + case "run_missing": + case "not_terminal": + return yield* new EnvironmentHttpBadRequestError({ message: error.detail }); + case "persistence_failed": + case "launch_failed": + case "projection_failed": + case "cancel_failed": + case "invalid_record": + return yield* failEnvironmentInternal("internal_error", error); + } +}); + +export const programAttemptHttpApiLayer = HttpApiBuilder.group( + EnvironmentHttpApi, + "programAttempts", + Effect.fnUntraced(function* (handlers) { + const attempts = yield* ProgramAttemptService.ProgramAttemptService; + + return handlers + .handle( + "launch", + Effect.fn("environment.programAttempts.launch")(function* (args) { + yield* annotateEnvironmentRequest(args.endpoint.name); + yield* requireEnvironmentScope(AuthOrchestrationOperateScope); + return yield* attempts.launch(args.payload).pipe(Effect.catch(mapProgramAttemptError)); + }), + ) + .handle( + "observe", + Effect.fn("environment.programAttempts.observe")(function* (args) { + yield* annotateEnvironmentRequest(args.endpoint.name); + yield* requireEnvironmentScope(AuthOrchestrationReadScope); + return yield* attempts + .observe(args.payload.attemptId) + .pipe(Effect.catch(mapProgramAttemptError)); + }), + ) + .handle( + "observeThread", + Effect.fn("environment.programAttempts.observeThread")(function* (args) { + yield* annotateEnvironmentRequest(args.endpoint.name); + yield* requireEnvironmentScope(AuthOrchestrationReadScope); + return yield* attempts + .observeThread(args.params.threadId) + .pipe(Effect.catch(mapProgramAttemptError)); + }), + ) + .handle( + "cancel", + Effect.fn("environment.programAttempts.cancel")(function* (args) { + yield* annotateEnvironmentRequest(args.endpoint.name); + yield* requireEnvironmentScope(AuthOrchestrationOperateScope); + return yield* attempts.cancel(args.payload).pipe(Effect.catch(mapProgramAttemptError)); + }), + ) + .handle( + "acknowledge", + Effect.fn("environment.programAttempts.acknowledge")(function* (args) { + yield* annotateEnvironmentRequest(args.endpoint.name); + yield* requireEnvironmentScope(AuthOrchestrationOperateScope); + return yield* attempts + .acknowledge(args.payload) + .pipe(Effect.catch(mapProgramAttemptError)); + }), + ); + }), +); diff --git a/apps/server/src/orchestration-v2/runtimeLayer.ts b/apps/server/src/orchestration-v2/runtimeLayer.ts index 6f4df38e1d9..cdfd8381630 100644 --- a/apps/server/src/orchestration-v2/runtimeLayer.ts +++ b/apps/server/src/orchestration-v2/runtimeLayer.ts @@ -41,6 +41,8 @@ import { layerFromProjectRepository as runtimePolicyLayerFromProjectRepository } import { layer as runtimeRequestServiceLayer } from "./RuntimeRequestService.ts"; import { layerWithLegacyImporter as threadManagementServiceLayer } from "./ThreadManagementService.ts"; import { layer as threadLaunchServiceLayer } from "./ThreadLaunchService.ts"; +import { layer as preparedWorktreeVerifierLayer } from "./PreparedWorktreeVerifier.ts"; +import { layer as programAttemptServiceLayer } from "./ProgramAttemptService.ts"; import { layer as threadLifecycleServiceLayer } from "./ThreadLifecycleService.ts"; import { layer as threadForkServiceLayer } from "./ThreadForkService.ts"; import { layer as turnItemPositionStoreLayer } from "./TurnItemPositionStore.ts"; @@ -196,9 +198,13 @@ const threadLaunchProvided = threadLaunchServiceLayer.pipe( threadManagementProvided, commandReceiptStoreProvided, idAllocatorLayer, + preparedWorktreeVerifierLayer, ), ), ); +const programAttemptProvided = programAttemptServiceLayer.pipe( + Layer.provide(Layer.mergeAll(threadLaunchProvided, threadManagementProvided)), +); const threadLifecycleProvided = threadLifecycleServiceLayer.pipe( Layer.provide(threadManagementProvided), ); @@ -258,6 +264,7 @@ export const OrchestrationV2ProductionLayerLive = Layer.mergeAll( OrchestrationV2LayerLive, ProjectServiceLayerLive, threadLaunchProvided, + programAttemptProvided, threadLifecycleProvided, scheduledTaskProvided, providerContinuationWorkerProvided, diff --git a/apps/server/src/persistence/Migrations.ts b/apps/server/src/persistence/Migrations.ts index cd27b51a368..cd4d706ca68 100644 --- a/apps/server/src/persistence/Migrations.ts +++ b/apps/server/src/persistence/Migrations.ts @@ -62,6 +62,7 @@ import Migration0046 from "./Migrations/046_ApplicationEventSource.ts"; import Migration0047 from "./Migrations/047_OrchestrationV2EffectCancellation.ts"; import Migration0048 from "./Migrations/048_ScheduledTasks.ts"; import Migration0049 from "./Migrations/049_LegacyV1ImportState.ts"; +import Migration0050 from "./Migrations/050_ProgramAttempts.ts"; /** * Migration loader with all migrations defined inline. @@ -123,6 +124,7 @@ export const migrationEntries = [ [47, "OrchestrationV2EffectCancellation", Migration0047], [48, "ScheduledTasks", Migration0048], [49, "LegacyV1ImportState", Migration0049], + [50, "ProgramAttempts", Migration0050], ] as const; export const migrationManifest = migrationEntries.map(([id, name]) => [id, name] as const); diff --git a/apps/server/src/persistence/Migrations/041_042_OrchestrationV2.test.ts b/apps/server/src/persistence/Migrations/041_042_OrchestrationV2.test.ts index 75cf364de79..b656695abfc 100644 --- a/apps/server/src/persistence/Migrations/041_042_OrchestrationV2.test.ts +++ b/apps/server/src/persistence/Migrations/041_042_OrchestrationV2.test.ts @@ -13,7 +13,7 @@ layer("038_039_OrchestrationV2", (it) => { Effect.sync(() => { assert.deepStrictEqual( migrationEntries.map(([id]) => id), - Array.from({ length: 49 }, (_, index) => index + 1), + Array.from({ length: 50 }, (_, index) => index + 1), ); }), ); diff --git a/apps/server/src/persistence/Migrations/050_ProgramAttempts.ts b/apps/server/src/persistence/Migrations/050_ProgramAttempts.ts new file mode 100644 index 00000000000..adcabd1188a --- /dev/null +++ b/apps/server/src/persistence/Migrations/050_ProgramAttempts.ts @@ -0,0 +1,28 @@ +import * as Effect from "effect/Effect"; +import * as SqlClient from "effect/unstable/sql/SqlClient"; + +export default Effect.gen(function* () { + const sql = yield* SqlClient.SqlClient; + + yield* sql` + CREATE TABLE program_attempts ( + attempt_id TEXT PRIMARY KEY, + launch_request_id TEXT NOT NULL, + launch_input_json TEXT NOT NULL, + project_id TEXT NOT NULL, + thread_id TEXT, + run_id TEXT, + cancel_input_json TEXT, + acknowledge_input_json TEXT, + terminal_result_json TEXT, + terminal_acknowledged_at TEXT, + created_at TEXT NOT NULL, + updated_at TEXT NOT NULL + ) + `; + + yield* sql` + CREATE UNIQUE INDEX program_attempts_launch_request_idx + ON program_attempts(launch_request_id) + `; +}); diff --git a/apps/server/src/provider/Layers/CodexSessionRuntime.test.ts b/apps/server/src/provider/Layers/CodexSessionRuntime.test.ts index a80ef2cf56a..f7750d78427 100644 --- a/apps/server/src/provider/Layers/CodexSessionRuntime.test.ts +++ b/apps/server/src/provider/Layers/CodexSessionRuntime.test.ts @@ -222,6 +222,20 @@ describe("buildTurnStartParams", () => { }), ); + it("seals read-only turns without an approval route", () => { + const params = Effect.runSync( + buildTurnStartParams({ + threadId: "provider-thread-1", + runtimeMode: "read-only", + prompt: "Inspect only", + }), + ); + + NodeAssert.equal(params.approvalPolicy, "never"); + NodeAssert.equal(params.approvalsReviewer, "user"); + NodeAssert.deepStrictEqual(params.sandboxPolicy, { type: "readOnly" }); + }); + it("omits collaboration mode when interaction mode is absent", () => { const params = Effect.runSync( buildTurnStartParams({ diff --git a/apps/server/src/provider/Layers/CodexSessionRuntime.ts b/apps/server/src/provider/Layers/CodexSessionRuntime.ts index 29bb992611c..26b85e359b5 100644 --- a/apps/server/src/provider/Layers/CodexSessionRuntime.ts +++ b/apps/server/src/provider/Layers/CodexSessionRuntime.ts @@ -270,6 +270,12 @@ function runtimeModeToThreadConfig(input: RuntimeMode): { readonly approvalsReviewer: EffectCodexSchema.V2ThreadStartParams__ApprovalsReviewer; } { switch (input) { + case "read-only": + return { + approvalPolicy: "never", + sandbox: "read-only", + approvalsReviewer: "user", + }; case "approval-required": return { approvalPolicy: "untrusted", @@ -319,6 +325,7 @@ function runtimeModeToTurnSandboxPolicy( input: RuntimeMode, ): EffectCodexSchema.V2TurnStartParams__SandboxPolicy { switch (input) { + case "read-only": case "approval-required": return { type: "readOnly", diff --git a/apps/server/src/server.ts b/apps/server/src/server.ts index 79772fbc669..a8c468cb67d 100644 --- a/apps/server/src/server.ts +++ b/apps/server/src/server.ts @@ -110,6 +110,7 @@ import { persistServerRuntimeState, } from "./serverRuntimeState.ts"; import { orchestrationHttpApiLayer } from "./orchestration-v2/http.ts"; +import { programAttemptHttpApiLayer } from "./orchestration-v2/programAttemptHttp.ts"; import { projectHttpApiLayer } from "./project/http.ts"; import * as NetService from "@t3tools/shared/Net"; import * as RelayClient from "@t3tools/shared/relayClient"; @@ -429,6 +430,7 @@ export const makeRoutesLayer = Layer.mergeAll( Layer.provide(authHttpApiLayer), Layer.provide(connectHttpApiLayer), Layer.provide(orchestrationHttpApiLayer), + Layer.provide(programAttemptHttpApiLayer), Layer.provide(pullRequestHttpApiLayer), Layer.provide(projectHttpApiLayer), Layer.provide(serverEnvironmentHttpApiLayer), diff --git a/apps/server/src/serverRuntimeStartup.ts b/apps/server/src/serverRuntimeStartup.ts index 99af107696c..573d4f034c5 100644 --- a/apps/server/src/serverRuntimeStartup.ts +++ b/apps/server/src/serverRuntimeStartup.ts @@ -29,6 +29,7 @@ import * as ExternalLauncher from "./process/externalLauncher.ts"; import * as EffectWorker from "./orchestration-v2/EffectWorker.ts"; import * as LegacyV1ThreadImporter from "./orchestration-v2/LegacyV1ThreadImporter.ts"; import * as ProjectionMaintenance from "./orchestration-v2/ProjectionMaintenance.ts"; +import * as ProgramAttempt from "./orchestration-v2/ProgramAttemptService.ts"; import * as ProviderRuntimeRecovery from "./orchestration-v2/ProviderRuntimeRecoveryService.ts"; import * as ProviderSessionManager from "./orchestration-v2/ProviderSessionManager.ts"; import * as ThreadLaunch from "./orchestration-v2/ThreadLaunchService.ts"; @@ -370,6 +371,7 @@ export const make = (options?: StartupOptions) => const projectionMaintenance = yield* ProjectionMaintenance.ProjectionMaintenanceV2; const legacyV1ThreadImporter = yield* LegacyV1ThreadImporter.LegacyV1ThreadImporter; const providerRuntimeRecovery = yield* ProviderRuntimeRecovery.ProviderRuntimeRecoveryService; + const programAttempts = yield* ProgramAttempt.ProgramAttemptService; const providerSessions = yield* ProviderSessionManager.ProviderSessionManagerV2; const agentAwarenessRelay = yield* AgentAwarenessRelay.AgentAwarenessRelay; const lifecycleEvents = yield* ServerLifecycleEvents.ServerLifecycleEvents; @@ -396,6 +398,7 @@ export const make = (options?: StartupOptions) => if (workerFiber !== null) { yield* Fiber.interrupt(workerFiber).pipe(Effect.ignore); } + yield* programAttempts.retainProcessInterruptions; yield* providerSessions.shutdown; const reconciliation = yield* providerRuntimeRecovery.reconcile("shutdown"); yield* Effect.logInfo("V2 orchestration shutdown reconciliation completed", reconciliation); @@ -487,7 +490,12 @@ export const make = (options?: StartupOptions) => "orchestration-v2.projections.rebuild", projectionMaintenance.rebuild, ), - recover: runStartupPhase("orchestration-v2.recovery", providerRuntimeRecovery.recover), + recover: runStartupPhase( + "orchestration-v2.recovery", + programAttempts.retainProcessInterruptions.pipe( + Effect.andThen(providerRuntimeRecovery.recover), + ), + ), startEffectWorker: runStartupPhase( "orchestration-v2.effect-worker.start", startEffectWorkerWithRelay({ diff --git a/apps/web/src/components/chat/ChatComposer.tsx b/apps/web/src/components/chat/ChatComposer.tsx index 8552c097594..cdccaff41f0 100644 --- a/apps/web/src/components/chat/ChatComposer.tsx +++ b/apps/web/src/components/chat/ChatComposer.tsx @@ -247,6 +247,11 @@ const runtimeModeConfig: Record< RuntimeMode, { label: string; description: string; icon: LucideIcon } > = { + "read-only": { + label: "Read only", + description: "Allow inspection but deny commands and file changes that need write access.", + icon: LockIcon, + }, "approval-required": { label: "Supervised", description: "Ask before commands and file changes.", diff --git a/apps/web/src/components/chat/ThreadDetailsPanel.test.tsx b/apps/web/src/components/chat/ThreadDetailsPanel.test.tsx index fc1ff4e3571..74133281910 100644 --- a/apps/web/src/components/chat/ThreadDetailsPanel.test.tsx +++ b/apps/web/src/components/chat/ThreadDetailsPanel.test.tsx @@ -20,6 +20,7 @@ vi.mock("../ProjectScriptsControl", () => ({ return null; }, })); +vi.mock("./ThreadProgramAttemptPanel", () => ({ ThreadProgramAttemptPanel: () => null })); vi.mock("./ThreadAutomationsPanel", () => ({ ThreadAutomationsPanel: () => null, })); diff --git a/apps/web/src/components/chat/ThreadDetailsPanel.tsx b/apps/web/src/components/chat/ThreadDetailsPanel.tsx index 60192d9cba4..879b2173048 100644 --- a/apps/web/src/components/chat/ThreadDetailsPanel.tsx +++ b/apps/web/src/components/chat/ThreadDetailsPanel.tsx @@ -24,6 +24,7 @@ import { cn } from "../../lib/utils"; import { OpenInPicker } from "./OpenInPicker"; import { ThreadAutomationsPanel } from "./ThreadAutomationsPanel"; import { ThreadRelationshipsPanel } from "./ThreadRelationshipsControl"; +import { ThreadProgramAttemptPanel } from "./ThreadProgramAttemptPanel"; interface VersionMismatchIssue { readonly clientVersion: string; @@ -253,6 +254,13 @@ export function ThreadDetailsPanel(props: ThreadDetailsPanelProps) { ) : null} + {!props.draftId ? ( + + ) : null} + {!props.draftId ? ( ) : null} diff --git a/apps/web/src/components/chat/ThreadProgramAttemptPanel.test.tsx b/apps/web/src/components/chat/ThreadProgramAttemptPanel.test.tsx new file mode 100644 index 00000000000..697da0e6b27 --- /dev/null +++ b/apps/web/src/components/chat/ThreadProgramAttemptPanel.test.tsx @@ -0,0 +1,77 @@ +import type { EnvironmentId, ProgramAttemptSnapshot } from "@t3tools/contracts"; +import { renderToStaticMarkup } from "react-dom/server"; +import { describe, expect, it } from "vite-plus/test"; + +import { ProgramAttemptSummary, programAttemptAttention } from "./ThreadProgramAttemptPanel"; + +function snapshot(overrides: Partial = {}): ProgramAttemptSnapshot { + return { + attemptId: "attempt:s6", + programId: "agents-dlr", + taskId: "agents-dlr.7", + attemptKind: "task", + candidateId: null, + reviewId: null, + reviewKind: null, + title: "S6 certification", + checkout: { + repositoryRoot: "/repo", + gitCommonDir: "/repo/.git", + worktreePath: "/repo/worktrees/prepared", + branch: "lavender/dirtyloops-parallel-runner", + startingCommit: "1234567890abcdef", + }, + projectId: "project:s6", + threadId: "thread:s6", + runId: "run:s6", + state: "active", + runStatus: "running", + terminalResult: null, + terminalAcknowledged: false, + ...overrides, + } as ProgramAttemptSnapshot; +} + +describe("ThreadProgramAttemptPanel", () => { + it("renders exact Task identity and read-only CLI guidance", () => { + const markup = renderToStaticMarkup( + , + ); + expect(markup).toContain("S6 certification"); + expect(markup).toContain("agents-dlr.7"); + expect(markup).toContain("/repo/worktrees/prepared"); + expect(markup).toContain("dirtyloops inspect"); + expect(markup).toContain("dirtyloops stop agents-dlr.7"); + expect(markup).not.toContain("Retry"); + expect(markup).not.toContain("Admission"); + }); + + it("identifies a focused candidate review and its live state", () => { + const markup = renderToStaticMarkup( + , + ); + expect(markup).toContain("Dirtyloops review"); + expect(markup).toContain("focused candidate review"); + expect(markup).toContain("Focused review · Completed"); + expect(markup).toContain("candidate:0123456789"); + }); + + it("does not invent a retry decision", () => { + expect(programAttemptAttention(snapshot(), "running")).toBe("None"); + expect(programAttemptAttention(snapshot(), "interrupted")).toContain("Dirtyloops will decide"); + }); +}); diff --git a/apps/web/src/components/chat/ThreadProgramAttemptPanel.tsx b/apps/web/src/components/chat/ThreadProgramAttemptPanel.tsx new file mode 100644 index 00000000000..1417a1c9894 --- /dev/null +++ b/apps/web/src/components/chat/ThreadProgramAttemptPanel.tsx @@ -0,0 +1,190 @@ +import type { + EnvironmentId, + OrchestrationV2RunStatus, + ProgramAttemptSnapshot, + ThreadId, +} from "@t3tools/contracts"; +import { scopeThreadRef } from "@t3tools/client-runtime/environment"; +import { ExternalLinkIcon } from "lucide-react"; +import type { ReactNode } from "react"; + +import { cn } from "../../lib/utils"; +import { useThreadProjection } from "../../state/entities"; +import { useEnvironmentQuery } from "../../state/query"; +import { serverEnvironment } from "../../state/server"; + +const STATUS_LABELS: Record = { + preparing: "Preparing", + queued: "Queued", + starting: "Starting", + running: "Running", + waiting: "Waiting", + completed: "Completed", + interrupted: "Interrupted", + failed: "Failed", + cancelled: "Cancelled", + rolled_back: "Rolled back", +}; + +const STATUS_DOTS: Record = { + preparing: "bg-info", + queued: "bg-muted-foreground/45", + starting: "bg-info", + running: "bg-success", + waiting: "bg-warning", + completed: "bg-success", + interrupted: "bg-warning", + failed: "bg-destructive", + cancelled: "bg-muted-foreground/45", + rolled_back: "bg-warning", +}; + +export function programAttemptAttention( + attempt: ProgramAttemptSnapshot, + status: OrchestrationV2RunStatus, +) { + const failure = attempt.terminalResult?.failure; + if (failure?.message) return failure.message; + if (status === "interrupted") + return "T3 restarted. Dirtyloops will decide whether this Task retries."; + if (status === "failed") + return "The T3 run failed. Inspect the Dirtyloops record before retrying."; + if (status === "cancelled") return "The T3 run was cancelled."; + if (status === "rolled_back") return "The T3 run was rolled back."; + return "None"; +} + +function DetailRow(props: { readonly label: string; readonly children: ReactNode }) { + return ( +
+
{props.label}
+
{props.children}
+
+ ); +} + +export function ProgramAttemptSummary(props: { + readonly attempt: ProgramAttemptSnapshot; + readonly environmentId: EnvironmentId; + readonly status: OrchestrationV2RunStatus; + readonly loadError?: string | null; +}) { + const { attempt, status } = props; + const threadHref = `/${encodeURIComponent(props.environmentId)}/${encodeURIComponent(attempt.threadId)}`; + const stopTarget = attempt.taskId ?? attempt.attemptId; + return ( +
+

+ Dirtyloops {attempt.attemptKind === "review" ? "review" : "task"} +

+
+ + + {attempt.title} + + + {attempt.programId ? ( + + {attempt.programId} + + ) : null} + {attempt.taskId ? ( + + {attempt.taskId} + + ) : null} + {attempt.reviewKind ? ( + + + {attempt.reviewKind === "broad" ? "Broad" : "Focused"} review ·{" "} + {STATUS_LABELS[status]} + + + ) : null} + {attempt.candidateId ? ( + + {attempt.candidateId} + + ) : null} + + + + {STATUS_LABELS[status]} + + + + {programAttemptAttention(attempt, status)} + + + + {attempt.checkout.worktreePath} + + + + {attempt.checkout.branch} + + + + {attempt.checkout.startingCommit.slice(0, 12)} + + + + + {attempt.runId} + + + +
+ {props.loadError ? ( +

+ Live details may be stale: {props.loadError} +

+ ) : null} +
+

+ Program controls run in the prepared checkout. +

+ + {`dirtyloops inspect\ndirtyloops run \ndirtyloops stop ${stopTarget}`} + +
+
+ ); +} + +export function ThreadProgramAttemptPanel(props: { + readonly environmentId: EnvironmentId; + readonly threadId: ThreadId; +}) { + const query = useEnvironmentQuery( + serverEnvironment.programAttempt({ + environmentId: props.environmentId, + input: { threadId: props.threadId }, + }), + ); + const projection = useThreadProjection(scopeThreadRef(props.environmentId, props.threadId)); + if (query.data === null) return null; + const attempt = query.data; + const liveStatus = projection?.projection.runs.find((run) => run.id === attempt.runId)?.status; + return ( + + ); +} diff --git a/packages/client-runtime/src/state/server.ts b/packages/client-runtime/src/state/server.ts index 3a594aaf9c0..6bb626fb2ef 100644 --- a/packages/client-runtime/src/state/server.ts +++ b/packages/client-runtime/src/state/server.ts @@ -6,6 +6,7 @@ import { type ServerLifecycleStreamReadyEvent, type ServerSelfUpdateProgressEvent, type ServerSelfUpdateResult, + type ThreadId, WS_METHODS, } from "@t3tools/contracts"; import * as Cause from "effect/Cause"; @@ -19,6 +20,7 @@ import * as Result from "effect/Result"; import * as Schema from "effect/Schema"; import * as Stream from "effect/Stream"; import * as SubscriptionRef from "effect/SubscriptionRef"; +import { HttpClient } from "effect/unstable/http"; import { AsyncResult, Atom } from "effect/unstable/reactivity"; import { @@ -26,13 +28,16 @@ import { createEnvironmentRpcCommand, createEnvironmentRpcQueryAtomFamily, createEnvironmentRpcSubscriptionAtomFamily, + createEnvironmentQueryAtomFamily, createRuntimeCommand, scheduleAtomCommandEffect, } from "./runtime.ts"; import { EnvironmentRegistry } from "../connection/registry.ts"; import { EnvironmentSupervisor } from "../connection/supervisor.ts"; +import { environmentEndpointUrl } from "../environment/endpoint.ts"; import { safeErrorLogAttributes } from "../errors/safeLog.ts"; import { EnvironmentCacheStore } from "../platform/persistence.ts"; +import { ManagedRelayDpopSigner } from "../relay/managedRelay.ts"; import { isRpcClientError, request, @@ -40,6 +45,8 @@ import { subscribe, type EnvironmentRpcInput, } from "../rpc/client.ts"; +import { executeEnvironmentHttpRequest, makeEnvironmentHttpApiClient } from "../rpc/http.ts"; +import { buildEnvironmentAuthHeaders, withEnvironmentCredentials } from "./environmentHttpAuth.ts"; import { followStreamInEnvironment } from "./runtime.ts"; export type ServerUpdateStage = "downloading" | "installing" | "resuming"; @@ -111,6 +118,51 @@ export class ServerUpdateTerminalError extends Schema.TaggedErrorClass()( + "ProgramAttemptConnectionNotReadyError", + { message: Schema.String }, +) {} + +const loadProgramAttemptForThread = Effect.fn( + "clientRuntime.state.server.loadProgramAttemptForThread", +)(function* (threadId: ThreadId) { + const supervisor = yield* EnvironmentSupervisor; + const prepared = yield* SubscriptionRef.get(supervisor.prepared); + if (Option.isNone(prepared)) { + return yield* new ProgramAttemptConnectionNotReadyError({ + message: "The environment HTTP connection is not ready.", + }); + } + const httpClient = yield* Effect.serviceOption(HttpClient.HttpClient); + if (Option.isNone(httpClient)) { + return yield* new ProgramAttemptConnectionNotReadyError({ + message: "The environment HTTP client is unavailable.", + }); + } + const signer = yield* Effect.serviceOption(ManagedRelayDpopSigner); + const requestUrl = environmentEndpointUrl( + prepared.value.httpBaseUrl, + `/api/program-attempts/threads/${threadId}`, + ); + return yield* Effect.gen(function* () { + const client = yield* makeEnvironmentHttpApiClient(prepared.value.httpBaseUrl); + const headers = yield* buildEnvironmentAuthHeaders( + prepared.value.httpAuthorization, + "GET", + requestUrl, + signer, + ); + return yield* executeEnvironmentHttpRequest( + requestUrl, + 6_000, + withEnvironmentCredentials( + prepared.value.httpAuthorization, + client.programAttempts.observeThread({ params: { threadId }, headers }), + ), + ); + }).pipe(Effect.provideService(HttpClient.HttpClient, httpClient.value)); +}); + // Covers the 120-second trial deadline and a final restart of the previous // version when the trial rolls back. const SERVER_UPDATE_RESUME_TIMEOUT = Duration.minutes(4); @@ -685,6 +737,12 @@ export function createServerEnvironmentAtoms( updateStateAtom, settingsValueAtom, providersValueAtom, + programAttempt: createEnvironmentQueryAtomFamily(runtime, { + label: "environment-data:server:program-attempt", + staleTimeMs: 5_000, + execute: (input: { readonly threadId: ThreadId }) => + loadProgramAttemptForThread(input.threadId), + }), traceDiagnostics: createEnvironmentRpcQueryAtomFamily(runtime, { label: "environment-data:server:trace-diagnostics", tag: WS_METHODS.serverGetTraceDiagnostics, diff --git a/packages/contracts/src/baseSchemas.ts b/packages/contracts/src/baseSchemas.ts index f06b39b8eab..82b8429390c 100644 --- a/packages/contracts/src/baseSchemas.ts +++ b/packages/contracts/src/baseSchemas.ts @@ -70,6 +70,10 @@ export const RunId = makeEntityId("RunId"); export type RunId = typeof RunId.Type; export const RunAttemptId = makeEntityId("RunAttemptId"); export type RunAttemptId = typeof RunAttemptId.Type; +export const ProgramAttemptId = makeEntityId("ProgramAttemptId"); +export type ProgramAttemptId = typeof ProgramAttemptId.Type; +export const ProgramAttemptRequestId = makeEntityId("ProgramAttemptRequestId"); +export type ProgramAttemptRequestId = typeof ProgramAttemptRequestId.Type; export const NodeId = makeEntityId("NodeId"); export type NodeId = typeof NodeId.Type; export const AuthSessionId = makeEntityId("AuthSessionId"); diff --git a/packages/contracts/src/environmentHttp.ts b/packages/contracts/src/environmentHttp.ts index ce319d25e60..ea5fe055aa6 100644 --- a/packages/contracts/src/environmentHttp.ts +++ b/packages/contracts/src/environmentHttp.ts @@ -32,6 +32,14 @@ import { OrchestrationV2ThreadDetailSnapshot, OrchestrationV2ThreadHistoryPage, } from "./orchestrationV2.ts"; +import { + ProgramAttemptCancelInput, + ProgramAttemptEffectInput, + ProgramAttemptIdentityInput, + ProgramAttemptLaunchInput, + ProgramAttemptSnapshot, + ProgramAttemptThreadInput, +} from "./programAttempt.ts"; import { Project, ProjectMutation, ProjectSnapshot } from "./project.ts"; import { PullRequestDiffInput, @@ -188,7 +196,10 @@ export class EnvironmentInternalError extends Schema.TaggedErrorClass()( @@ -506,6 +517,14 @@ const EnvironmentOrchestrationThreadHistoryErrors = [ EnvironmentInternalError, ] as const; +const EnvironmentProgramAttemptErrors = [ + EnvironmentHttpBadRequestError, + EnvironmentHttpConflictError, + EnvironmentResourceNotFoundError, + EnvironmentScopeRequiredError, + EnvironmentInternalError, +] as const; + export class EnvironmentOrchestrationHttpApi extends HttpApiGroup.make("orchestration") .add( HttpApiEndpoint.get("shellSnapshot", "/api/orchestration/shell", { @@ -540,6 +559,48 @@ export class EnvironmentOrchestrationHttpApi extends HttpApiGroup.make("orchestr }).middleware(EnvironmentAuthenticatedAuth), ) {} +export class EnvironmentProgramAttemptsHttpApi extends HttpApiGroup.make("programAttempts") + .add( + HttpApiEndpoint.post("launch", "/api/program-attempts/launch", { + headers: OptionalBearerHeaders, + payload: ProgramAttemptLaunchInput, + success: ProgramAttemptSnapshot, + error: EnvironmentProgramAttemptErrors, + }).middleware(EnvironmentAuthenticatedAuth), + ) + .add( + HttpApiEndpoint.post("observe", "/api/program-attempts/observe", { + headers: OptionalBearerHeaders, + payload: ProgramAttemptIdentityInput, + success: ProgramAttemptSnapshot, + error: EnvironmentProgramAttemptErrors, + }).middleware(EnvironmentAuthenticatedAuth), + ) + .add( + HttpApiEndpoint.get("observeThread", "/api/program-attempts/threads/:threadId", { + headers: OptionalBearerHeaders, + params: ProgramAttemptThreadInput, + success: Schema.NullOr(ProgramAttemptSnapshot), + error: EnvironmentProgramAttemptErrors, + }).middleware(EnvironmentAuthenticatedAuth), + ) + .add( + HttpApiEndpoint.post("cancel", "/api/program-attempts/cancel", { + headers: OptionalBearerHeaders, + payload: ProgramAttemptCancelInput, + success: ProgramAttemptSnapshot, + error: EnvironmentProgramAttemptErrors, + }).middleware(EnvironmentAuthenticatedAuth), + ) + .add( + HttpApiEndpoint.post("acknowledge", "/api/program-attempts/acknowledge", { + headers: OptionalBearerHeaders, + payload: ProgramAttemptEffectInput, + success: ProgramAttemptSnapshot, + error: EnvironmentProgramAttemptErrors, + }).middleware(EnvironmentAuthenticatedAuth), + ) {} + export class EnvironmentProjectsHttpApi extends HttpApiGroup.make("projects") .add( HttpApiEndpoint.get("snapshot", "/api/projects", { @@ -638,6 +699,7 @@ export class EnvironmentHttpApi extends HttpApi.make("environment") .add(EnvironmentMetadataHttpApi) .add(EnvironmentAuthHttpApi) .add(EnvironmentOrchestrationHttpApi) + .add(EnvironmentProgramAttemptsHttpApi) .add(EnvironmentPullRequestsHttpApi) .add(EnvironmentProjectsHttpApi) .add(EnvironmentConnectHttpApi) {} diff --git a/packages/contracts/src/index.ts b/packages/contracts/src/index.ts index cfe24f5ffd2..04787175aca 100644 --- a/packages/contracts/src/index.ts +++ b/packages/contracts/src/index.ts @@ -40,6 +40,7 @@ export * from "./previewAutomation.ts"; export * from "./resourceTelemetry.ts"; export * from "./usage.ts"; export * from "./scheduledTask.ts"; +export * from "./programAttempt.ts"; export * from "./worktreeMcp.ts"; export * from "./resourceTelemetry.ts"; export * from "./rpc.ts"; diff --git a/packages/contracts/src/orchestrationV2.ts b/packages/contracts/src/orchestrationV2.ts index d58739aa997..10cfc698dc4 100644 --- a/packages/contracts/src/orchestrationV2.ts +++ b/packages/contracts/src/orchestrationV2.ts @@ -1953,6 +1953,15 @@ export const OrchestrationV2StoredEventJson = Schema.Struct({ }); export type OrchestrationV2StoredEventJson = typeof OrchestrationV2StoredEventJson.Type; +export const PreparedWorktreeCheckout = Schema.Struct({ + repositoryRoot: TrimmedNonEmptyString, + gitCommonDir: TrimmedNonEmptyString, + worktreePath: TrimmedNonEmptyString, + branch: TrimmedNonEmptyString, + startingCommit: TrimmedNonEmptyString, +}); +export type PreparedWorktreeCheckout = typeof PreparedWorktreeCheckout.Type; + export const OrchestrationV2Command = Schema.Union([ Schema.Struct({ type: Schema.Literal("thread.create"), @@ -2126,7 +2135,7 @@ export const OrchestrationV2Command = Schema.Union([ commandId: CommandId, threadId: ThreadId, runId: RunId, - phase: Schema.Literals(["worktree", "setup"]), + phase: Schema.Literals(["worktree", "verification", "setup"]), }), Schema.Struct({ type: Schema.Literal("prepared-run.fail"), diff --git a/packages/contracts/src/programAttempt.ts b/packages/contracts/src/programAttempt.ts new file mode 100644 index 00000000000..9238fc64e34 --- /dev/null +++ b/packages/contracts/src/programAttempt.ts @@ -0,0 +1,92 @@ +import * as Schema from "effect/Schema"; + +import { + IsoDateTime, + ProgramAttemptId, + ProgramAttemptRequestId, + ProjectId, + RunId, + ThreadId, + TrimmedNonEmptyString, +} from "./baseSchemas.ts"; +import { ModelSelection } from "./modelSelection.ts"; +import { + OrchestrationV2ProviderFailure, + OrchestrationV2RunStatus, + PreparedWorktreeCheckout, +} from "./orchestrationV2.ts"; +import { ProviderInteractionMode, RuntimeMode } from "./providerPolicy.ts"; + +export const ProgramAttemptProviderPolicy = Schema.Struct({ + modelSelection: ModelSelection, + runtimeMode: RuntimeMode, + interactionMode: ProviderInteractionMode, +}); +export type ProgramAttemptProviderPolicy = typeof ProgramAttemptProviderPolicy.Type; + +export const ProgramAttemptLaunchInput = Schema.Struct({ + attemptId: ProgramAttemptId, + requestId: ProgramAttemptRequestId, + programId: Schema.optional(TrimmedNonEmptyString), + taskId: Schema.optional(TrimmedNonEmptyString), + attemptKind: Schema.optional(Schema.Literals(["task", "review"])), + candidateId: Schema.optional(TrimmedNonEmptyString), + reviewId: Schema.optional(TrimmedNonEmptyString), + reviewKind: Schema.optional(Schema.Literals(["broad", "focused"])), + projectId: ProjectId, + title: TrimmedNonEmptyString, + prompt: TrimmedNonEmptyString, + checkout: PreparedWorktreeCheckout, + providerPolicy: ProgramAttemptProviderPolicy, +}); +export type ProgramAttemptLaunchInput = typeof ProgramAttemptLaunchInput.Type; + +export const ProgramAttemptIdentityInput = Schema.Struct({ + attemptId: ProgramAttemptId, +}); +export type ProgramAttemptIdentityInput = typeof ProgramAttemptIdentityInput.Type; + +export const ProgramAttemptThreadInput = Schema.Struct({ + threadId: ThreadId, +}); +export type ProgramAttemptThreadInput = typeof ProgramAttemptThreadInput.Type; + +export const ProgramAttemptEffectInput = Schema.Struct({ + attemptId: ProgramAttemptId, + requestId: ProgramAttemptRequestId, +}); +export type ProgramAttemptEffectInput = typeof ProgramAttemptEffectInput.Type; + +export const ProgramAttemptCancelInput = Schema.Struct({ + ...ProgramAttemptEffectInput.fields, + reason: Schema.optional(TrimmedNonEmptyString), +}); +export type ProgramAttemptCancelInput = typeof ProgramAttemptCancelInput.Type; + +export const ProgramAttemptTerminalResult = Schema.Struct({ + status: Schema.Literals(["completed", "interrupted", "failed", "cancelled", "rolled_back"]), + output: Schema.NullOr(Schema.String), + failure: Schema.NullOr(OrchestrationV2ProviderFailure), + completedAt: Schema.NullOr(IsoDateTime), +}); +export type ProgramAttemptTerminalResult = typeof ProgramAttemptTerminalResult.Type; + +export const ProgramAttemptSnapshot = Schema.Struct({ + attemptId: ProgramAttemptId, + programId: Schema.NullOr(TrimmedNonEmptyString), + taskId: Schema.NullOr(TrimmedNonEmptyString), + attemptKind: Schema.NullOr(Schema.Literals(["task", "review"])), + candidateId: Schema.NullOr(TrimmedNonEmptyString), + reviewId: Schema.NullOr(TrimmedNonEmptyString), + reviewKind: Schema.NullOr(Schema.Literals(["broad", "focused"])), + title: TrimmedNonEmptyString, + checkout: PreparedWorktreeCheckout, + projectId: ProjectId, + threadId: ThreadId, + runId: RunId, + state: Schema.Literals(["preparing", "active", "terminal"]), + runStatus: OrchestrationV2RunStatus, + terminalResult: Schema.NullOr(ProgramAttemptTerminalResult), + terminalAcknowledged: Schema.Boolean, +}); +export type ProgramAttemptSnapshot = typeof ProgramAttemptSnapshot.Type; diff --git a/packages/contracts/src/providerPolicy.ts b/packages/contracts/src/providerPolicy.ts index 40517e99b38..2850a67689a 100644 --- a/packages/contracts/src/providerPolicy.ts +++ b/packages/contracts/src/providerPolicy.ts @@ -16,6 +16,7 @@ export const ProviderSandboxMode = Schema.Literals([ export type ProviderSandboxMode = typeof ProviderSandboxMode.Type; export const RuntimeMode = Schema.Literals([ + "read-only", "approval-required", "auto-accept-edits", "auto",