diff --git a/apps/server/src/auth/RpcAuthorization.ts b/apps/server/src/auth/RpcAuthorization.ts index 2e2bdea73fc72b9b496b1a19a9997280752ddd8f..ca55e4e95e0139458af07c12d457715cdbb97b0e 100644 --- a/apps/server/src/auth/RpcAuthorization.ts +++ b/apps/server/src/auth/RpcAuthorization.ts @@ -111,6 +111,8 @@ export const RPC_REQUIRED_SCOPES = { [WS_METHODS.attachmentsDelete]: AuthOrchestrationOperateScope, [WS_METHODS.providerUploadFeedback]: AuthOrchestrationOperateScope, [WS_METHODS.subscribeVcsStatus]: AuthOrchestrationReadScope, + [WS_METHODS.subscribeWorktreeSetup]: AuthOrchestrationReadScope, + [WS_METHODS.worktreeSetupCancel]: AuthOrchestrationOperateScope, [WS_METHODS.subscribeResourceTelemetry]: AuthOrchestrationReadScope, [WS_METHODS.vcsRefreshStatus]: AuthOrchestrationReadScope, [WS_METHODS.vcsPull]: AuthOrchestrationOperateScope, diff --git a/apps/server/src/git/GitWorkflowService.ts b/apps/server/src/git/GitWorkflowService.ts index 5e3e5b0420f2b4fc0ecc19e71417c7ba6d76ac6e..9f3231beb49066c9f7835977ce6423c570e071c0 100644 --- a/apps/server/src/git/GitWorkflowService.ts +++ b/apps/server/src/git/GitWorkflowService.ts @@ -69,6 +69,7 @@ export class GitWorkflowService extends Context.Service< ) => Effect.Effect; readonly createWorktree: ( input: VcsCreateWorktreeInput, + options?: GitVcsDriver.CreateWorktreeOptions, ) => Effect.Effect; readonly fetchRemote: (input: { readonly cwd: string; @@ -338,9 +339,9 @@ export const make = Effect.gen(function* () { isGitRepository ? git.listRefs(input) : Effect.succeed(nonRepositoryListRefs()), ), ), - createWorktree: (input) => + createWorktree: (input, options) => ensureGitCommand("GitWorkflowService.createWorktree", input.cwd).pipe( - Effect.andThen(git.createWorktree(input)), + Effect.andThen(git.createWorktree(input, options)), ), fetchRemote: (input) => ensureGitCommand("GitWorkflowService.fetchRemote", input.cwd).pipe( diff --git a/apps/server/src/project/ProjectSetupScriptRunner.test.ts b/apps/server/src/project/ProjectSetupScriptRunner.test.ts index fc582ef355168b97eff0ff3f688ba4942f2b94a6..c0d15f105af7c9f0807f95d57a4b613774a72374 100644 --- a/apps/server/src/project/ProjectSetupScriptRunner.test.ts +++ b/apps/server/src/project/ProjectSetupScriptRunner.test.ts @@ -1,5 +1,6 @@ import { describe, expect, it, vi } from "@effect/vitest"; -import { type OrchestrationProject, ProjectId } from "@t3tools/contracts"; +import { type OrchestrationProject, ProjectId, type TerminalEvent } from "@t3tools/contracts"; +import { HostProcessEnvironment, HostProcessPlatform } from "@t3tools/shared/hostProcess"; import * as Effect from "effect/Effect"; import * as Layer from "effect/Layer"; import * as Option from "effect/Option"; @@ -54,11 +55,11 @@ const makeProjectionSnapshotQueryLayer = (project: OrchestrationProject) => searchThreads: () => Effect.succeed({ matches: [] }), }); -const makeTerminalManagerLayer = ( - overrides: Pick, -) => +type TerminalOverrides = Pick & + Partial>; + +const makeTerminalManagerLayer = (overrides: TerminalOverrides) => Layer.succeed(TerminalManager.TerminalManager, { - ...overrides, attachStream: () => Effect.die(new Error("unused")), resize: () => Effect.void, clear: () => Effect.void, @@ -66,11 +67,12 @@ const makeTerminalManagerLayer = ( close: () => Effect.void, subscribe: () => Effect.succeed(() => undefined), subscribeMetadata: () => Effect.succeed(() => undefined), + ...overrides, }); const testLayer = ( project: OrchestrationProject, - terminal: Pick, + terminal: TerminalOverrides, settings = ServerSettings.layerTest(), ) => ProjectSetupScriptRunner.layer.pipe( @@ -198,6 +200,7 @@ describe("ProjectSetupScriptRunner", () => { status: "started", scriptId: "setup", scriptName: "Setup", + scriptCommand: "bun install", terminalId: "setup-setup", cwd: "/repo/worktrees/a", }); @@ -220,6 +223,209 @@ describe("ProjectSetupScriptRunner", () => { }, ); + it.effect( + "wraps the command with a completion sentinel and resolves the exit code from terminal output", + () => { + const open = vi.fn(() => + Effect.succeed({ + threadId: "thread-1", + terminalId: "setup-setup", + cwd: "/repo/worktrees/a", + worktreePath: "/repo/worktrees/a", + status: "running" as const, + pid: 123, + history: "", + exitCode: null, + exitSignal: null, + label: "setup-setup", + updatedAt: "2026-01-01T00:00:00.000Z", + }), + ); + const writes: string[] = []; + const write = vi.fn((input: { data: string }) => + Effect.sync(() => void writes.push(input.data)), + ); + let listener: ((event: TerminalEvent) => Effect.Effect) | null = null; + const subscribe = vi.fn((next: (event: TerminalEvent) => Effect.Effect) => { + listener = next; + return Effect.succeed(() => { + listener = null; + }); + }); + const project = makeProject([ + { + id: "setup", + name: "Setup", + command: "bun install", + icon: "configure", + runOnWorktreeCreate: true, + }, + ]); + const emit = (data: string) => + Effect.suspend(() => + listener + ? listener({ threadId: "thread-1", terminalId: "setup-setup", type: "output", data }) + : Effect.void, + ); + + return Effect.gen(function* () { + const runner = yield* ProjectSetupScriptRunner.ProjectSetupScriptRunner; + const seen: string[] = []; + const result = yield* runner.runForThread({ + threadId: "thread-1", + projectCwd: "/repo/project", + worktreePath: "/repo/worktrees/a", + observeCompletion: { + onOutputLine: (line) => Effect.sync(() => void seen.push(line)), + }, + }); + expect(result.status).toBe("started"); + if (result.status !== "started") return; + expect(result.completion).toBeDefined(); + + // The subscription is attached before the command is written. + expect(subscribe).toHaveBeenCalledTimes(1); + expect(writes).toHaveLength(1); + // The block closes on its own line so a trailing comment in the + // command cannot swallow the sentinel, and the sentinel carries a + // per-run token so script output cannot spoof it. + const written = writes[0] ?? ""; + const sentinel = /__T3_SETUP_DONE___[0-9a-f]{32}:/.exec(written)?.[0]; + expect(sentinel).toBeDefined(); + expect(written).toBe(`( bun install\r); printf '\\n${sentinel}%s\\n' "$?"\r`); + + // Output arrives in chunks; partial lines are buffered until a newline, + // control sequences are stripped, and the echoed wrapper is hidden. + yield* emit(`( bun install\r\n> ); printf '\\n${sentinel}%s\\n' "$?"\r\n`); + yield* emit("\u001b[32mResolving"); + yield* emit(" deps\u001b[0m\r\nDone in 2s\r\n"); + // A spoofed sentinel from the script itself must not settle completion. + yield* emit("__T3_SETUP_DONE__:0\r\n"); + yield* emit(`__T3_SETUP_DONE___${"0".repeat(32)}:0\r\n`); + yield* emit(`${sentinel}3\r\n`); + + const completion = yield* result.completion!; + expect(completion.exitCode).toBe(3); + expect(seen).toEqual([ + "Resolving deps", + "Done in 2s", + "__T3_SETUP_DONE__:0", + `__T3_SETUP_DONE___${"0".repeat(32)}:0`, + ]); + // The subscription is torn down once the sentinel arrives. + expect(listener).toBeNull(); + }).pipe( + Effect.provide(testLayer(project, { open, write, subscribe })), + Effect.provideService(HostProcessPlatform, "linux"), + Effect.provideService(HostProcessEnvironment, { SHELL: "/bin/zsh" }), + ); + }, + ); + + it.effect("unsubscribes from terminal output when the command cannot be written", () => { + const open = vi.fn(() => + Effect.succeed({ + threadId: "thread-1", + terminalId: "setup-setup", + cwd: "/repo/worktrees/a", + worktreePath: "/repo/worktrees/a", + status: "running" as const, + pid: 123, + history: "", + exitCode: null, + exitSignal: null, + label: "setup-setup", + updatedAt: "2026-01-01T00:00:00.000Z", + }), + ); + const write = vi.fn(() => + Effect.fail( + new TerminalManager.TerminalCwdStatError({ cwd: "/repo/worktrees/a", cause: {} }), + ), + ); + const unsubscribe = vi.fn(); + const subscribe = vi.fn(() => Effect.succeed(unsubscribe)); + const project = makeProject([ + { + id: "setup", + name: "Setup", + command: "bun install", + icon: "configure", + runOnWorktreeCreate: true, + }, + ]); + + return Effect.gen(function* () { + const runner = yield* ProjectSetupScriptRunner.ProjectSetupScriptRunner; + const result = yield* runner + .runForThread({ + threadId: "thread-1", + projectCwd: "/repo/project", + worktreePath: "/repo/worktrees/a", + observeCompletion: {}, + }) + .pipe(Effect.result); + expect(result._tag).toBe("Failure"); + expect(unsubscribe).toHaveBeenCalledTimes(1); + }).pipe(Effect.provide(testLayer(project, { open, write, subscribe }))); + }); + + it.effect.each([ + { + shell: "/usr/bin/fish", + expected: + /^begin\rbun install\rend; printf '\\n__T3_SETUP_DONE___[0-9a-f]{32}:%s\\n' \$status\r$/, + }, + { + shell: "/bin/bash", + expected: /^\( bun install\r\); printf '\\n__T3_SETUP_DONE___[0-9a-f]{32}:%s\\n' "\$\?"\r$/, + }, + ])("wraps the command for the $shell syntax", ({ shell, expected }) => { + const open = vi.fn(() => + Effect.succeed({ + threadId: "thread-1", + terminalId: "setup-setup", + cwd: "/repo/worktrees/a", + worktreePath: "/repo/worktrees/a", + status: "running" as const, + pid: 123, + history: "", + exitCode: null, + exitSignal: null, + label: "setup-setup", + updatedAt: "2026-01-01T00:00:00.000Z", + }), + ); + const writes: string[] = []; + const write = vi.fn((input: { data: string }) => + Effect.sync(() => void writes.push(input.data)), + ); + const project = makeProject([ + { + id: "setup", + name: "Setup", + command: "bun install", + icon: "configure", + runOnWorktreeCreate: true, + }, + ]); + return Effect.gen(function* () { + const runner = yield* ProjectSetupScriptRunner.ProjectSetupScriptRunner; + yield* runner.runForThread({ + threadId: "thread-1", + projectCwd: "/repo/project", + worktreePath: "/repo/worktrees/a", + observeCompletion: {}, + }); + expect(writes).toHaveLength(1); + expect(writes[0]).toMatch(expected); + }).pipe( + Effect.provide(testLayer(project, { open, write })), + Effect.provideService(HostProcessPlatform, "linux"), + Effect.provideService(HostProcessEnvironment, { SHELL: shell }), + ); + }); + it.effect("keeps terminal failures as the exact cause of a structured operation error", () => { const rootCause = new Error("stat failed"); const terminalError = new TerminalManager.TerminalCwdStatError({ diff --git a/apps/server/src/project/ProjectSetupScriptRunner.ts b/apps/server/src/project/ProjectSetupScriptRunner.ts index 3bbb0daa7994087127533bc1b0fc9112ffb29cdc..c80dd535f51440e42daa6a6c78c38d02ae0266e1 100644 --- a/apps/server/src/project/ProjectSetupScriptRunner.ts +++ b/apps/server/src/project/ProjectSetupScriptRunner.ts @@ -1,10 +1,15 @@ import { ProjectId } from "@t3tools/contracts"; +import { HostProcessEnvironment, HostProcessPlatform } from "@t3tools/shared/hostProcess"; import { projectScriptRuntimeEnv, resolveProjectScripts, setupProjectScript, } from "@t3tools/shared/projectScripts"; +import * as NodeCrypto from "node:crypto"; + +import * as Clock from "effect/Clock"; import * as Context from "effect/Context"; +import * as Deferred from "effect/Deferred"; import * as Effect from "effect/Effect"; import * as Layer from "effect/Layer"; import * as Option from "effect/Option"; @@ -22,8 +27,24 @@ export interface ProjectSetupScriptRunnerResultStarted { readonly status: "started"; readonly scriptId: string; readonly scriptName: string; + readonly scriptCommand: string; readonly terminalId: string; readonly cwd: string; + /** + * Resolves when the script's shell prints the completion sentinel. The + * exit code is null when the terminal exited or was closed before the + * sentinel arrived. Only present when `observeCompletion` was requested. + */ + readonly completion?: Effect.Effect; +} + +export interface ProjectSetupScriptCompletion { + readonly exitCode: number | null; + readonly durationMs: number; +} + +export interface ProjectSetupScriptOutputLine { + readonly line: string; } export type ProjectSetupScriptRunnerResult = @@ -36,6 +57,14 @@ export interface ProjectSetupScriptRunnerInput { readonly projectCwd?: string; readonly worktreePath: string; readonly preferredTerminalId?: string; + /** + * Wrap the command so the shell reports its exit code back through the + * terminal stream, and forward cleaned output lines while it runs. The + * bootstrap flow uses this to drive the worktree setup card. + */ + readonly observeCompletion?: { + readonly onOutputLine?: (line: string) => Effect.Effect; + }; } export class ProjectSetupScriptOperationError extends Schema.TaggedError()( @@ -83,11 +112,170 @@ export class ProjectSetupScriptRunner extends Context.Service< } >()("t3/project/ProjectSetupScriptRunner") {} +/** @public Service construction is part of the canonical Effect module API. */ +/** + * Marker the wrapped setup command echoes so the exit code can be read from + * the PTY stream. Each run gets its own random token so script output cannot + * spoof completion, and the sentinel pattern is built per run from it. + */ +const COMPLETION_SENTINEL_PREFIX = "__T3_SETUP_DONE__"; +const OUTPUT_LINE_MAX_LENGTH = 400; +/** A partial line longer than this is a byte stream, not a line. Keep only the tail. */ +const PARTIAL_LINE_MAX_LENGTH = 4_096; + +function completionSentinel(token: string): string { + return `${COMPLETION_SENTINEL_PREFIX}_${token}:`; +} + +function completionSentinelPattern(token: string): RegExp { + return new RegExp(`${COMPLETION_SENTINEL_PREFIX}_${token}:(-?\\d+)`); +} + +/** Removes ANSI escape sequences and cursor controls so lines can be shown as plain text. */ +function stripTerminalControl(text: string): string { + return ( + text + .replace( + // eslint-disable-next-line no-control-regex + /\x1b\[[0-9;?]*[ -/]*[@-~]|\x1b\][^\x07\x1b]*(?:\x07|\x1b\\)|\x1b[()][A-Za-z0-9]|\x1b[=>]/g, + "", + ) + // eslint-disable-next-line no-control-regex + .replace(/[\x00-\x08\x0b\x0c\x0e-\x1f\x7f]/g, "") + ); +} + +type CompletionShell = "posix" | "fish" | "powershell"; + +/** + * Predicts the shell TerminalManager will spawn for the setup terminal. The + * manager takes `$SHELL` on POSIX and PowerShell on Windows, falling back to + * other shells only when that one fails to spawn. + */ +function resolveCompletionShell( + platform: NodeJS.Platform, + env: NodeJS.ProcessEnv, +): CompletionShell { + if (platform === "win32") return "powershell"; + const shell = env.SHELL ?? ""; + const name = shell.split("/").at(-1) ?? shell; + if (name === "fish") return "fish"; + if (name === "pwsh" || name === "powershell") return "powershell"; + return "posix"; +} + +/** + * Builds the shell input for the setup script. The command runs inside a + * block and the block closes on its own line, so a trailing `# comment` or a + * heredoc terminator in the command cannot swallow the sentinel. The shell + * reads the whole block before running any of it, so a script that reads + * stdin cannot consume the sentinel line either. Lines are separated by `\r` + * because that is the Enter key for every shell's line editor. + */ +function wrapCommandForCompletion( + command: string, + shell: CompletionShell, + sentinel: string, +): string { + const body = command.replace(/\r?\n/g, "\r"); + switch (shell) { + case "powershell": + return `$global:LASTEXITCODE = $null; & {\r${body}\r}; if ($null -ne $LASTEXITCODE) { $__t3c = $LASTEXITCODE } elseif ($?) { $__t3c = 0 } else { $__t3c = 1 }; Write-Host "${sentinel}$__t3c"`; + case "fish": + return `begin\r${body}\rend; printf '\\n${sentinel}%s\\n' $status`; + case "posix": + return `( ${body}\r); printf '\\n${sentinel}%s\\n' "$?"`; + } +} + /** @public Service construction is part of the canonical Effect module API. */ export const make = Effect.gen(function* () { const projectionSnapshotQuery = yield* ProjectionSnapshotQuery.ProjectionSnapshotQuery; const terminalManager = yield* TerminalManager.TerminalManager; const serverSettings = yield* ServerSettings.ServerSettingsService; + const completionShell = resolveCompletionShell( + yield* HostProcessPlatform, + yield* HostProcessEnvironment, + ); + + /** + * Watches the setup terminal for the completion sentinel. Terminal output is + * a byte stream, so partial lines are buffered until a newline. The + * subscription is torn down once the sentinel, an exit, or a close arrives. + */ + const observeTerminalCompletion = (input: { + readonly threadId: string; + readonly terminalId: string; + /** Per-run sentinel, so only this run's wrapper can settle completion. */ + readonly sentinel: string; + readonly sentinelPattern: RegExp; + /** The shell echoes typed input; lines ending with these are the wrapper, not output. */ + readonly echoedWrapperLines: ReadonlyArray; + readonly onOutputLine: ((line: string) => Effect.Effect) | undefined; + }) => + Effect.gen(function* () { + const startedAtMs = yield* Clock.currentTimeMillis; + const done = yield* Deferred.make(); + let lineBuffer = ""; + let settled = false; + + const settle = (exitCode: number | null) => + Effect.suspend(() => { + if (settled) return Effect.void; + settled = true; + return Clock.currentTimeMillis.pipe( + Effect.flatMap((nowMs) => + Deferred.succeed(done, { exitCode, durationMs: nowMs - startedAtMs }), + ), + Effect.asVoid, + ); + }); + + const handleLine = (rawLine: string) => + Effect.suspend(() => { + const sentinel = input.sentinelPattern.exec(rawLine); + if (sentinel) { + const parsed = Number(sentinel[1]); + return settle(Number.isFinite(parsed) ? parsed : null); + } + const cleaned = stripTerminalControl(rawLine).trimEnd(); + if ( + cleaned.length === 0 || + cleaned.includes(input.sentinel) || + input.echoedWrapperLines.some((echoed) => cleaned.endsWith(echoed)) || + input.onOutputLine === undefined + ) { + return Effect.void; + } + return input.onOutputLine(cleaned.slice(0, OUTPUT_LINE_MAX_LENGTH)); + }); + + const unsubscribe = yield* terminalManager.subscribe((event) => { + if (event.threadId !== input.threadId || event.terminalId !== input.terminalId) { + return Effect.void; + } + if (event.type === "output") { + lineBuffer += event.data; + const lines = lineBuffer.split(/\r?\n/); + lineBuffer = lines.pop() ?? ""; + // A script that never prints a newline must not grow this forever. + // The sentinel is always on its own line, so keeping the tail is safe. + if (lineBuffer.length > PARTIAL_LINE_MAX_LENGTH) { + lineBuffer = lineBuffer.slice(-PARTIAL_LINE_MAX_LENGTH); + } + return Effect.forEach(lines, handleLine, { discard: true }); + } + if (event.type === "exited" || event.type === "closed") { + return settle(null); + } + return Effect.void; + }); + + const completion = Deferred.await(done).pipe( + Effect.ensuring(Effect.sync(() => unsubscribe())), + ); + return { completion, unsubscribe }; + }); const runForThread: ProjectSetupScriptRunner["Service"]["runForThread"] = Effect.fn( "ProjectSetupScriptRunner.runForThread", @@ -154,6 +342,16 @@ export const make = Effect.gen(function* () { project: { cwd: project.workspaceRoot }, worktreePath: input.worktreePath, }); + const observe = input.observeCompletion; + const completionToken = observe ? NodeCrypto.randomUUID().replaceAll("-", "") : null; + const commandLine = + observe && completionToken + ? wrapCommandForCompletion( + script.command, + completionShell, + completionSentinel(completionToken), + ) + : script.command; yield* terminalManager .open({ @@ -173,11 +371,24 @@ export const make = Effect.gen(function* () { }), ), ); + // Subscribe before writing so the sentinel cannot race past the listener. + const observed = + observe && completionToken + ? yield* observeTerminalCompletion({ + threadId: input.threadId, + terminalId, + sentinel: completionSentinel(completionToken), + sentinelPattern: completionSentinelPattern(completionToken), + echoedWrapperLines: commandLine.split("\r").filter((line) => line.length > 0), + onOutputLine: observe.onOutputLine, + }) + : undefined; + yield* terminalManager .write({ threadId: input.threadId, terminalId, - data: `${script.command}\r`, + data: `${commandLine}\r`, }) .pipe( Effect.mapError( @@ -188,14 +399,18 @@ export const make = Effect.gen(function* () { cause, }), ), + // Nothing will ever settle the completion if the command never ran. + Effect.tapError(() => Effect.sync(() => observed?.unsubscribe())), ); return { status: "started", scriptId: script.id, scriptName: script.name, + scriptCommand: script.command, terminalId, cwd, + ...(observed ? { completion: observed.completion } : {}), } as const; }); diff --git a/apps/server/src/project/WorktreeSetupTracker.test.ts b/apps/server/src/project/WorktreeSetupTracker.test.ts new file mode 100644 index 0000000000000000000000000000000000000000..16e7b6aa64c2ca4c70fe41206c6648f97c9ba119 --- /dev/null +++ b/apps/server/src/project/WorktreeSetupTracker.test.ts @@ -0,0 +1,226 @@ +import { describe, expect, it } from "@effect/vitest"; +import { ThreadId, WorktreeSetupSnapshot } from "@t3tools/contracts"; +import * as Deferred from "effect/Deferred"; +import * as Duration from "effect/Duration"; +import * as Effect from "effect/Effect"; +import * as Exit from "effect/Exit"; +import * as Fiber from "effect/Fiber"; +import * as Schema from "effect/Schema"; +import * as Stream from "effect/Stream"; +import * as TestClock from "effect/testing/TestClock"; + +import * as WorktreeSetupTracker from "./WorktreeSetupTracker.ts"; + +const threadId = ThreadId.make("thread-1"); + +describe("WorktreeSetupTracker", () => { + it.effect("records stage transitions, checkout progress, and the final phase", () => + Effect.gen(function* () { + const tracker = yield* WorktreeSetupTracker.make; + yield* tracker.begin({ + threadId, + branch: "feature", + baseRef: "main", + stages: ["checkout", "fetch", "agent"], + fiber: null, + }); + + const initial = yield* tracker.get(threadId); + // Stages are reordered into the canonical setup order. + expect(initial?.stages.map((stage) => stage.id)).toEqual(["fetch", "checkout", "agent"]); + expect(initial?.phase).toBe("running"); + + yield* tracker.stageStatus(threadId, "fetch", "running"); + yield* tracker.stageStatus(threadId, "fetch", "done", "origin/main at abc1234"); + yield* tracker.stageStatus(threadId, "checkout", "running"); + yield* tracker.stage(threadId, "checkout", { percent: 42, detail: "42 / 100 files" }); + yield* tracker.finish(threadId, "failed", "boom"); + + const final = yield* tracker.get(threadId); + expect(final?.phase).toBe("failed"); + expect(final?.error).toBe("boom"); + const [fetch, checkout, agent] = final?.stages ?? []; + expect(fetch).toMatchObject({ status: "done", detail: "origin/main at abc1234" }); + expect(fetch?.startedAt).not.toBeNull(); + expect(fetch?.endedAt).not.toBeNull(); + // A stage still running when the setup fails is marked failed. + expect(checkout).toMatchObject({ status: "failed", percent: 42 }); + expect(agent?.status).toBe("pending"); + expect(final?.sequence).toBeGreaterThan(initial?.sequence ?? 0); + }), + ); + + it.effect("stream emits the current snapshot first and then only newer ones", () => + Effect.gen(function* () { + const tracker = yield* WorktreeSetupTracker.make; + yield* tracker.begin({ + threadId, + branch: null, + baseRef: null, + stages: ["agent"], + fiber: null, + }); + + const collected = yield* tracker.stream(threadId).pipe( + Stream.takeUntil((snapshot) => snapshot?.sequence === 2), + Stream.runCollect, + Effect.forkChild, + ); + yield* Effect.yieldNow; + yield* tracker.stageStatus(threadId, "agent", "running"); + yield* tracker.appendTail(threadId, "agent", "line 1"); + + const snapshots = yield* Fiber.join(collected); + const sequences = snapshots.map((snapshot) => snapshot?.sequence ?? -1); + expect(sequences.at(-1)).toBe(2); + // Delivery is latest-value per subscriber, so intermediates may be + // skipped but never delivered out of order. + expect(sequences).toEqual([...sequences].toSorted((a, b) => a - b)); + expect(snapshots.at(-1)?.stages[0]?.tail).toEqual(["line 1"]); + }), + ); + + it.effect("stream never steps back behind the snapshot it started from", () => + Effect.gen(function* () { + const tracker = yield* WorktreeSetupTracker.make; + yield* tracker.begin({ + threadId, + branch: null, + baseRef: null, + stages: ["agent"], + fiber: null, + }); + yield* tracker.stageStatus(threadId, "agent", "running"); + yield* tracker.stageStatus(threadId, "agent", "done"); + + // A late subscriber starts at sequence 2 and must never see 0 or 1. + const collected = yield* tracker.stream(threadId).pipe( + Stream.takeUntil((snapshot) => snapshot?.phase === "done"), + Stream.runCollect, + Effect.forkChild, + ); + yield* Effect.yieldNow; + yield* tracker.finish(threadId, "done"); + + const snapshots = yield* Fiber.join(collected); + expect(snapshots.length).toBeGreaterThan(0); + expect(snapshots.every((snapshot) => (snapshot?.sequence ?? -1) >= 2)).toBe(true); + expect(snapshots.at(-1)?.phase).toBe("done"); + }), + ); + + it.effect("a new setup on the same thread keeps sequences increasing", () => + Effect.gen(function* () { + const tracker = yield* WorktreeSetupTracker.make; + yield* tracker.begin({ + threadId, + branch: "first", + baseRef: null, + stages: ["agent"], + fiber: null, + }); + yield* tracker.finish(threadId, "failed", "boom"); + const failedSequence = (yield* tracker.get(threadId))?.sequence ?? -1; + + // A stream opened on the failed setup must still receive the next one. + const collected = yield* tracker.stream(threadId).pipe( + Stream.takeUntil((snapshot) => snapshot?.branch === "second"), + Stream.runCollect, + Effect.forkChild, + ); + yield* Effect.yieldNow; + yield* tracker.begin({ + threadId, + branch: "second", + baseRef: null, + stages: ["agent"], + fiber: null, + }); + + const snapshots = yield* Fiber.join(collected); + const last = snapshots.at(-1); + expect(last?.phase).toBe("running"); + expect(last?.sequence).toBeGreaterThan(failedSequence); + }), + ); + + it.effect("finished setups are dropped after the retention window", () => + Effect.gen(function* () { + const tracker = yield* WorktreeSetupTracker.make; + yield* tracker.begin({ + threadId, + branch: null, + baseRef: null, + stages: ["agent"], + fiber: null, + }); + yield* tracker.finish(threadId, "done"); + expect((yield* tracker.get(threadId))?.phase).toBe("done"); + + yield* TestClock.adjust(Duration.seconds(31)); + expect(yield* tracker.get(threadId)).toBeNull(); + }), + ); + + it.effect("cancel interrupts the bootstrap fiber and reports whether one was running", () => + Effect.gen(function* () { + const tracker = yield* WorktreeSetupTracker.make; + const started = yield* Deferred.make(); + const fiber = yield* Deferred.succeed(started, undefined).pipe( + Effect.andThen(Effect.never), + Effect.forkChild, + ); + yield* Deferred.await(started); + yield* tracker.begin({ threadId, branch: null, baseRef: null, stages: ["agent"], fiber }); + + expect(yield* tracker.cancel(threadId)).toBe(true); + // cancel returns only after the bootstrap fiber has unwound. + const exit = yield* Fiber.await(fiber); + expect(Exit.hasInterrupts(exit)).toBe(true); + + yield* tracker.finish(threadId, "cancelled"); + expect(yield* tracker.cancel(threadId)).toBe(false); + expect(yield* tracker.cancel(ThreadId.make("unknown"))).toBe(false); + }), + ); + + it.effect("markUncancellable makes a later cancel a no-op while the setup keeps running", () => + Effect.gen(function* () { + const tracker = yield* WorktreeSetupTracker.make; + const fiber = yield* Effect.forkChild(Effect.never); + yield* tracker.begin({ threadId, branch: null, baseRef: null, stages: ["agent"], fiber }); + + yield* tracker.markUncancellable(threadId); + expect(yield* tracker.cancel(threadId)).toBe(false); + expect((yield* tracker.get(threadId))?.phase).toBe("running"); + yield* Fiber.interrupt(fiber); + }), + ); + + it.effect("clamps free text to the contract limits before publishing", () => + Effect.gen(function* () { + const tracker = yield* WorktreeSetupTracker.make; + yield* tracker.begin({ + threadId, + branch: null, + baseRef: null, + stages: ["checkout", "setup-script"], + fiber: null, + }); + const long = "x".repeat(2_000); + + yield* tracker.stageStatus(threadId, "checkout", "done", long); + yield* tracker.stage(threadId, "setup-script", { detail: long }); + yield* tracker.appendTail(threadId, "setup-script", long); + yield* tracker.finish(threadId, "failed", long); + + const snapshot = yield* tracker.get(threadId); + expect(snapshot?.stages[0]?.detail?.length).toBe(200); + expect(snapshot?.stages[1]?.detail?.length).toBe(200); + expect(snapshot?.stages[1]?.tail[0]?.length).toBe(400); + expect(snapshot?.error?.length).toBe(1000); + // The wire schema must accept what the tracker publishes. + expect(Schema.is(WorktreeSetupSnapshot)(snapshot)).toBe(true); + }), + ); +}); diff --git a/apps/server/src/project/WorktreeSetupTracker.ts b/apps/server/src/project/WorktreeSetupTracker.ts new file mode 100644 index 0000000000000000000000000000000000000000..41c84f2d649d680151b22c9f218671fa914308a9 --- /dev/null +++ b/apps/server/src/project/WorktreeSetupTracker.ts @@ -0,0 +1,364 @@ +import type { + ThreadId, + WorktreeSetupSnapshot, + WorktreeSetupStage, + WorktreeSetupStageId, + WorktreeSetupStageStatus, +} from "@t3tools/contracts"; +import { + WORKTREE_SETUP_DETAIL_MAX_LENGTH, + WORKTREE_SETUP_ERROR_MAX_LENGTH, + WORKTREE_SETUP_STAGE_ORDER, + WORKTREE_SETUP_TAIL_LINE_MAX_LENGTH, +} from "@t3tools/contracts"; +import * as Context from "effect/Context"; +import * as DateTime from "effect/DateTime"; +import * as Effect from "effect/Effect"; +import * as Fiber from "effect/Fiber"; +import * as Layer from "effect/Layer"; +import * as PubSub from "effect/PubSub"; +import * as Queue from "effect/Queue"; +import * as Ref from "effect/Ref"; +import * as Stream from "effect/Stream"; + +/** + * Tracks the live stages of a bootstrap worktree setup per thread so clients + * can render a progress card while the first turn is still being prepared. + * + * State is memory only. It exists from the first `begin` until the turn starts + * or the setup fails, plus a short grace window so a client that subscribes + * late still sees the final state. Nothing here is persisted or event-sourced: + * the durable record of a setup is the thread's worktree path and the setup + * script activities, both of which already exist. + */ +export class WorktreeSetupTracker extends Context.Service< + WorktreeSetupTracker, + { + /** Creates a fresh running snapshot for the thread, replacing any prior one. */ + readonly begin: (input: { + readonly threadId: ThreadId; + readonly branch: string | null; + readonly baseRef: string | null; + readonly stages: ReadonlyArray; + /** Interrupting this fiber cancels the bootstrap. */ + readonly fiber: Fiber.Fiber | null; + }) => Effect.Effect; + readonly update: ( + threadId: ThreadId, + mutate: (snapshot: WorktreeSetupSnapshot) => WorktreeSetupSnapshot, + ) => Effect.Effect; + readonly stage: ( + threadId: ThreadId, + stageId: WorktreeSetupStageId, + patch: Partial>, + ) => Effect.Effect; + readonly stageStatus: ( + threadId: ThreadId, + stageId: WorktreeSetupStageId, + status: WorktreeSetupStageStatus, + detail?: string | null, + ) => Effect.Effect; + readonly appendTail: ( + threadId: ThreadId, + stageId: WorktreeSetupStageId, + line: string, + ) => Effect.Effect; + readonly finish: ( + threadId: ThreadId, + phase: "done" | "failed" | "cancelled", + error?: string | null, + ) => Effect.Effect; + /** + * Drops the cancel handle. Called right before the turn is dispatched so a + * late cancel cannot roll back a thread whose agent has already started. + */ + readonly markUncancellable: (threadId: ThreadId) => Effect.Effect; + /** + * Interrupts the running bootstrap and waits for it to unwind, so the + * caller's dispatch has already failed and rolled back when this returns. + * Returns false when nothing is running or the setup is past cancellation. + */ + readonly cancel: (threadId: ThreadId) => Effect.Effect; + readonly get: (threadId: ThreadId) => Effect.Effect; + /** Emits the current snapshot (or null) first, then every change until unsubscribed. */ + readonly stream: (threadId: ThreadId) => Stream.Stream; + } +>()("t3/project/WorktreeSetupTracker") {} + +const TAIL_LINE_LIMIT = 4; + +/** Keeps free text inside the contract limit, ending in an ellipsis when cut. */ +function clampText(text: string, maxLength: number): string { + return text.length <= maxLength ? text : `${text.slice(0, maxLength - 1)}\u2026`; +} + +const clampDetail = (detail: string | null): string | null => + detail === null ? null : clampText(detail, WORKTREE_SETUP_DETAIL_MAX_LENGTH); +/** Finished snapshots stay visible this long so a late subscriber sees the outcome. */ +const FINISHED_RETENTION = "30 seconds"; + +const nowIso = Effect.map(DateTime.now, DateTime.formatIso); + +interface TrackedSetup { + readonly snapshot: WorktreeSetupSnapshot; + readonly fiber: Fiber.Fiber | null; +} + +function emptyStage(id: WorktreeSetupStageId): WorktreeSetupStage { + return { + id, + status: "pending", + startedAt: null, + endedAt: null, + percent: null, + detail: null, + tail: [], + }; +} + +export const make = Effect.gen(function* () { + const setups = yield* Ref.make(new Map()); + const changes = yield* PubSub.unbounded<{ + readonly threadId: ThreadId; + readonly snapshot: WorktreeSetupSnapshot | null; + }>(); + const retentionFibers = new Map>(); + // Sequences keep increasing across setups of the same thread so a stream + // opened during a previous setup still accepts the next one's first snapshot. + const lastSequenceByThread = new Map(); + + const publish = (threadId: ThreadId, snapshot: WorktreeSetupSnapshot | null) => + PubSub.publish(changes, { threadId, snapshot }).pipe(Effect.asVoid); + + const modify = ( + threadId: ThreadId, + mutate: (tracked: TrackedSetup) => TrackedSetup, + ): Effect.Effect => + Ref.modify(setups, (current) => { + const existing = current.get(threadId); + if (!existing) return [null, current] as const; + const nextTracked = mutate(existing); + const nextSnapshot = { + ...nextTracked.snapshot, + sequence: existing.snapshot.sequence + 1, + }; + lastSequenceByThread.set(threadId, nextSnapshot.sequence); + const next = new Map(current); + next.set(threadId, { ...nextTracked, snapshot: nextSnapshot }); + return [nextSnapshot, next] as const; + }).pipe(Effect.tap((snapshot) => (snapshot ? publish(threadId, snapshot) : Effect.void))); + + const clearRetention = (threadId: ThreadId) => { + const fiber = retentionFibers.get(threadId); + retentionFibers.delete(threadId); + return fiber ? Fiber.interrupt(fiber).pipe(Effect.ignore) : Effect.void; + }; + + const remove = (threadId: ThreadId) => + Ref.update(setups, (current) => { + if (!current.has(threadId)) return current; + const next = new Map(current); + next.delete(threadId); + return next; + }).pipe( + Effect.andThen(publish(threadId, null)), + // A subscriber that outlives retention sees `null` here and accepts any + // sequence after it, so the counter can start over for this thread. + Effect.tap(() => Effect.sync(() => lastSequenceByThread.delete(threadId))), + ); + + const begin: WorktreeSetupTracker["Service"]["begin"] = (input) => + Effect.gen(function* () { + yield* clearRetention(input.threadId); + const startedAt = yield* nowIso; + const ordered = WORKTREE_SETUP_STAGE_ORDER.filter((id) => input.stages.includes(id)); + const snapshot: WorktreeSetupSnapshot = { + threadId: input.threadId, + phase: "running", + startedAt, + endedAt: null, + branch: input.branch, + baseRef: input.baseRef, + worktreePath: null, + setupScript: null, + stages: ordered.map(emptyStage), + error: null, + sequence: (lastSequenceByThread.get(input.threadId) ?? -1) + 1, + }; + lastSequenceByThread.set(input.threadId, snapshot.sequence); + yield* Ref.update(setups, (current) => { + const next = new Map(current); + next.set(input.threadId, { snapshot, fiber: input.fiber }); + return next; + }); + yield* publish(input.threadId, snapshot); + }); + + const update: WorktreeSetupTracker["Service"]["update"] = (threadId, mutate) => + modify(threadId, (tracked) => ({ ...tracked, snapshot: mutate(tracked.snapshot) })).pipe( + Effect.asVoid, + ); + + const stage: WorktreeSetupTracker["Service"]["stage"] = (threadId, stageId, patch) => + update(threadId, (snapshot) => ({ + ...snapshot, + stages: snapshot.stages.map((entry) => + entry.id === stageId + ? { + ...entry, + ...patch, + ...(patch.detail === undefined ? {} : { detail: clampDetail(patch.detail ?? null) }), + } + : entry, + ), + })); + + const stageStatus: WorktreeSetupTracker["Service"]["stageStatus"] = ( + threadId, + stageId, + status, + detail, + ) => + nowIso.pipe( + Effect.flatMap((at) => + update(threadId, (snapshot) => ({ + ...snapshot, + stages: snapshot.stages.map((entry) => { + if (entry.id !== stageId) return entry; + const startedAt = entry.startedAt ?? (status === "pending" ? null : at); + const endedAt = + status === "running" || status === "pending" ? null : (entry.endedAt ?? at); + return { + ...entry, + status, + startedAt, + endedAt, + ...(detail === undefined ? {} : { detail: clampDetail(detail ?? null) }), + }; + }), + })), + ), + ); + + const appendTail: WorktreeSetupTracker["Service"]["appendTail"] = (threadId, stageId, line) => + update(threadId, (snapshot) => ({ + ...snapshot, + stages: snapshot.stages.map((entry) => + entry.id === stageId + ? { + ...entry, + tail: [...entry.tail, clampText(line, WORKTREE_SETUP_TAIL_LINE_MAX_LENGTH)].slice( + -TAIL_LINE_LIMIT, + ), + } + : entry, + ), + })); + + const finish: WorktreeSetupTracker["Service"]["finish"] = (threadId, phase, error) => + Effect.gen(function* () { + const endedAt = yield* nowIso; + const snapshot = yield* modify(threadId, (tracked) => ({ + fiber: null, + snapshot: { + ...tracked.snapshot, + phase, + endedAt, + error: + error === undefined || error === null + ? null + : clampText(error, WORKTREE_SETUP_ERROR_MAX_LENGTH), + stages: tracked.snapshot.stages.map((entry) => + entry.status === "running" + ? { + ...entry, + status: phase === "done" ? "done" : phase === "cancelled" ? "skipped" : "failed", + endedAt, + } + : entry, + ), + }, + })); + if (!snapshot) return; + yield* clearRetention(threadId); + const fiber = yield* remove(threadId).pipe( + Effect.delay(FINISHED_RETENTION), + Effect.ensuring( + Effect.sync(() => { + // Only drop our own entry: a newer setup may have replaced it. + if (retentionFibers.get(threadId) === fiber) retentionFibers.delete(threadId); + }), + ), + Effect.forkDetach, + ); + retentionFibers.set(threadId, fiber); + }); + + const markUncancellable: WorktreeSetupTracker["Service"]["markUncancellable"] = (threadId) => + Ref.update(setups, (current) => { + const existing = current.get(threadId); + if (!existing || existing.fiber === null) return current; + const next = new Map(current); + next.set(threadId, { ...existing, fiber: null }); + return next; + }); + + const cancel: WorktreeSetupTracker["Service"]["cancel"] = (threadId) => + Effect.gen(function* () { + const current = yield* Ref.get(setups); + const tracked = current.get(threadId); + if (!tracked || tracked.snapshot.phase !== "running" || !tracked.fiber) { + return false; + } + yield* Fiber.interrupt(tracked.fiber); + return true; + }); + + const get: WorktreeSetupTracker["Service"]["get"] = (threadId) => + Ref.get(setups).pipe(Effect.map((current) => current.get(threadId)?.snapshot ?? null)); + + /** + * Each subscriber gets a one-slot sliding mailbox: a slow WebSocket only + * ever holds the newest snapshot, so a chatty setup script cannot grow the + * server heap. Snapshots are whole states, so skipping intermediates is safe. + */ + const stream: WorktreeSetupTracker["Service"]["stream"] = (threadId) => + Stream.callback( + (mailbox) => + Effect.gen(function* () { + const subscription = yield* PubSub.subscribe(changes); + const initial = yield* get(threadId); + // Changes published between subscribing and reading `initial` are + // already folded into it. Drop them so the client never steps back. + let lastSequence = initial?.sequence ?? -1; + Queue.offerUnsafe(mailbox, initial); + yield* Stream.fromSubscription(subscription).pipe( + Stream.runForEach((change) => + Effect.sync(() => { + if (change.threadId !== threadId) return; + if (change.snapshot !== null && change.snapshot.sequence <= lastSequence) return; + lastSequence = change.snapshot?.sequence ?? -1; + Queue.offerUnsafe(mailbox, change.snapshot); + }), + ), + Effect.forkScoped, + ); + }), + { bufferSize: 1, strategy: "sliding" }, + ); + + return WorktreeSetupTracker.of({ + begin, + update, + stage, + stageStatus, + appendTail, + finish, + markUncancellable, + cancel, + get, + stream, + }); +}); + +export const layer = Layer.effect(WorktreeSetupTracker, make); diff --git a/apps/server/src/server.test.ts b/apps/server/src/server.test.ts index af934c59d480e1fd8d714ba3785a326ccf36f10c..3eb40a1b99062b1619dab03d8a7e88cf5783fcbd 100644 --- a/apps/server/src/server.test.ts +++ b/apps/server/src/server.test.ts @@ -148,6 +148,7 @@ import * as ServerRuntimeStartup from "./serverRuntimeStartup.ts"; import * as ServiceLauncherClient from "./cloud/serviceLauncherClient.ts"; import * as ServerSettings from "./serverSettings.ts"; import * as TerminalManager from "./terminal/Manager.ts"; +import * as WorktreeSetupTracker from "./project/WorktreeSetupTracker.ts"; import * as PreviewManager from "./preview/Manager.ts"; import * as PortScanner from "./preview/PortScanner.ts"; import * as BrowserTraceCollector from "./observability/BrowserTraceCollector.ts"; @@ -924,9 +925,12 @@ const buildAppUnderTest = (options?: { }), ), Layer.provide( - Layer.mock(TerminalManager.TerminalManager)({ - ...options?.layers?.terminalManager, - }), + Layer.mergeAll( + Layer.mock(TerminalManager.TerminalManager)({ + ...options?.layers?.terminalManager, + }), + WorktreeSetupTracker.layer, + ), ), Layer.provide( Layer.mergeAll( @@ -10558,6 +10562,7 @@ it.layer(NodeServices.layer)("server router seam", (it) => { status: "started" as const, scriptId: "setup", scriptName: "Setup", + scriptCommand: "npm install", terminalId: "setup-setup", cwd: "/tmp/bootstrap-worktree", }), @@ -10673,12 +10678,23 @@ it.layer(NodeServices.layer)("server router seam", (it) => { "resolve-remote-commit", "create-worktree", ]); - assert.deepEqual(runForThread.mock.calls[0]?.[0], { - threadId: ThreadId.make("thread-bootstrap"), - projectId: defaultProjectId, - projectCwd: "/tmp/project", - worktreePath: "/tmp/bootstrap-worktree", - }); + const runForThreadInput = runForThread.mock.calls[0]?.[0]; + assert.deepEqual( + runForThreadInput && { + threadId: runForThreadInput.threadId, + projectId: runForThreadInput.projectId, + projectCwd: runForThreadInput.projectCwd, + worktreePath: runForThreadInput.worktreePath, + }, + { + threadId: ThreadId.make("thread-bootstrap"), + projectId: defaultProjectId, + projectCwd: "/tmp/project", + worktreePath: "/tmp/bootstrap-worktree", + }, + ); + // Worktree bootstraps observe script completion so the setup card can show the exit code. + assert.isDefined(runForThreadInput?.observeCompletion); assert.deepEqual(refreshStatus.mock.calls[0]?.[0], "/tmp/bootstrap-worktree"); const setupActivities = dispatchedCommands.filter( @@ -11111,6 +11127,7 @@ it.layer(NodeServices.layer)("server router seam", (it) => { status: "started" as const, scriptId: "setup", scriptName: "Setup", + scriptCommand: "npm install", terminalId: "setup-setup", cwd: "/tmp/bootstrap-worktree", }), diff --git a/apps/server/src/server.ts b/apps/server/src/server.ts index 6148bed9bdd5f6f2ca193586421791a1c7618900..189d62dd8362222a077f17f52b197dfa3648defe 100644 --- a/apps/server/src/server.ts +++ b/apps/server/src/server.ts @@ -110,6 +110,7 @@ import * as PullRequestReadCache from "./pullRequest/PullRequestReadCache.ts"; import * as SourceControlRateLimit from "./sourceControl/SourceControlRateLimit.ts"; import * as SourceControlRepositoryService from "./sourceControl/SourceControlRepositoryService.ts"; import * as ProjectSetupScriptRunner from "./project/ProjectSetupScriptRunner.ts"; +import * as WorktreeSetupTracker from "./project/WorktreeSetupTracker.ts"; import { ObservabilityLive } from "./observability/Layers/Observability.ts"; import * as ServerEnvironment from "./environment/ServerEnvironment.ts"; import * as RemoteOpenTargets from "./environment/RemoteOpenTargets.ts"; @@ -332,6 +333,7 @@ const PullRequestServiceLive = PullRequestService.layer.pipe( const GitManagerLayerLive = GitManager.layer.pipe( Layer.provideMerge(ProjectSetupScriptRunner.layer.pipe(Layer.provide(ServerSettingsLayerLive))), + Layer.provideMerge(WorktreeSetupTracker.layer), Layer.provideMerge(GitVcsDriver.layer), Layer.provideMerge(SourceControlProviderRegistryLayerLive), Layer.provideMerge(TextGeneration.layer), diff --git a/apps/server/src/vcs/GitVcsDriver.ts b/apps/server/src/vcs/GitVcsDriver.ts index 9b25e915973c92b7b37d8b7385a240137a83cb78..4d63a447d63ff4d0b35588b8139cf55d77729b0e 100644 --- a/apps/server/src/vcs/GitVcsDriver.ts +++ b/apps/server/src/vcs/GitVcsDriver.ts @@ -102,6 +102,36 @@ export interface ExecuteGitProgress { }) => Effect.Effect; } +/** + * Progress callbacks for `createWorktree`. Git prints `Updating files: 78% (2104/2700)` + * to stderr during checkout, and `Submodule path 'x': checked out` during + * submodule init. The tracker uses these to drive the worktree setup card. + */ +export interface CreateWorktreeProgress { + /** + * Fires once `git worktree add` has created and registered the directory, + * before the (possibly long) submodule step. Git refuses an existing path, + * so a path reported here belongs to this call and is safe to remove on + * cancel. + */ + readonly onWorktreeClaimed?: (path: string) => Effect.Effect; + readonly onCheckoutProgress?: (input: { + percent: number; + completed: number; + total: number; + }) => Effect.Effect; + readonly onSubmodulesStarted?: () => Effect.Effect; + readonly onSubmoduleLine?: (line: string) => Effect.Effect; + readonly onSubmodulesFinished?: (input: { + ok: boolean; + detail: string | null; + }) => Effect.Effect; +} + +export interface CreateWorktreeOptions { + readonly progress?: CreateWorktreeProgress; +} + export interface GitCommitProgress { readonly onOutputLine?: (input: { stream: "stdout" | "stderr"; @@ -280,6 +310,7 @@ export class GitVcsDriver extends Context.Service< readonly pullCurrentBranch: (cwd: string) => Effect.Effect; readonly createWorktree: ( input: VcsCreateWorktreeInput, + options?: CreateWorktreeOptions, ) => Effect.Effect; readonly fetchPullRequestBranch: ( input: GitFetchPullRequestBranchInput, diff --git a/apps/server/src/vcs/GitVcsDriverCore.test.ts b/apps/server/src/vcs/GitVcsDriverCore.test.ts index 4f1f0204a4ed0934ffbf92e82d9b1a14c6661a73..bc8a12700997e418cc2f8125235ff643cbad4dde 100644 --- a/apps/server/src/vcs/GitVcsDriverCore.test.ts +++ b/apps/server/src/vcs/GitVcsDriverCore.test.ts @@ -20,7 +20,11 @@ import { ChildProcess, ChildProcessSpawner } from "effect/unstable/process"; import { GitCommandError, type ReviewDiffFileContentsInput } from "@t3tools/contracts"; import { ServerConfig } from "../config.ts"; -import { makeGitVcsDriverCore, splitNullSeparatedGitStdoutPaths } from "./GitVcsDriverCore.ts"; +import { + makeGitVcsDriverCore, + parseGitCheckoutProgressLine, + splitNullSeparatedGitStdoutPaths, +} from "./GitVcsDriverCore.ts"; import * as GitVcsDriver from "./GitVcsDriver.ts"; const ServerConfigLayer = ServerConfig.layerTest(process.cwd(), { @@ -1553,6 +1557,20 @@ it.layer(TestLayer)("GitVcsDriver core integration", (it) => { }); describe("worktree operations", () => { + it("parses checkout progress lines from git's stderr", () => { + assert.deepStrictEqual(parseGitCheckoutProgressLine("Updating files: 78% (2104/2700)"), { + percent: 78, + completed: 2104, + total: 2700, + }); + // Progress lines arrive carriage-return separated and end with a done marker. + assert.deepStrictEqual( + parseGitCheckoutProgressLine("Updating files: 100% (2700/2700), done."), + { percent: 100, completed: 2700, total: 2700 }, + ); + assert.strictEqual(parseGitCheckoutProgressLine("Preparing worktree (new branch 'x')"), null); + }); + // NTFS rejects a newline in a file name, so there is nothing to preserve there. it.effect.skipIf(HostProcessPlatform.defaultValue() === "win32")( "preserves newline characters in worktree paths when listing refs", @@ -1668,6 +1686,53 @@ it.layer(TestLayer)("GitVcsDriver core integration", (it) => { }), ); + it.effect("reports checkout progress while creating a worktree", () => + Effect.gen(function* () { + const cwd = yield* makeTmpDir(); + const { initialBranch } = yield* initRepoWithCommit(cwd); + for (let index = 0; index < 5; index += 1) { + yield* writeTextFile(cwd, `file-${index}.txt`, `${index}\n`); + } + yield* git(cwd, ["add", "."]); + yield* git(cwd, ["commit", "-m", "add files"]); + const pathService = yield* Path.Path; + const worktreePath = pathService.join( + yield* makeTmpDir("git-worktrees-"), + "progress-worktree", + ); + const driver = yield* GitVcsDriver.GitVcsDriver; + const seen = yield* Ref.make>( + [], + ); + const claimed = yield* Ref.make<{ path: string; existed: boolean } | null>(null); + + yield* driver.createWorktree( + { cwd, path: worktreePath, refName: initialBranch, newRefName: "feature/progress" }, + { + progress: { + onWorktreeClaimed: (path) => + Ref.set(claimed, { path, existed: NodeFS.existsSync(path) }), + onCheckoutProgress: (update) => Ref.update(seen, (all) => [...all, update]), + }, + }, + ); + // Claimed only once git has registered the directory. + assert.deepEqual(yield* Ref.get(claimed), { path: worktreePath, existed: true }); + + // Git separates live progress updates with `\r`, so the driver must + // surface every intermediate percentage, not just the final line. + const updates = yield* Ref.get(seen); + assert.isAbove(updates.length, 1); + assert.equal(updates.at(-1)?.percent, 100); + assert.equal(updates.at(-1)?.total, 6); + const completed = updates.map((update) => update.completed); + assert.deepEqual( + completed, + completed.toSorted((a, b) => a - b), + ); + }), + ); + it.effect("creates and removes a worktree for a new refName", () => Effect.gen(function* () { const cwd = yield* makeTmpDir(); diff --git a/apps/server/src/vcs/GitVcsDriverCore.ts b/apps/server/src/vcs/GitVcsDriverCore.ts index d371e63617f6a1a042292638f1eac15f2c52f44e..86a3e2ebd8123a54008010a625cdebad3ad3c04e 100644 --- a/apps/server/src/vcs/GitVcsDriverCore.ts +++ b/apps/server/src/vcs/GitVcsDriverCore.ts @@ -641,6 +641,25 @@ const createTrace2Monitor = Effect.fn("createTrace2Monitor")(function* ( }; }); +const GIT_CHECKOUT_PROGRESS_LINE = /Updating files:\s+(\d+)%\s+\((\d+)\/(\d+)\)/; + +/** Parses `Updating files: 78% (2104/2700)` from git's stderr progress output. */ +export function parseGitCheckoutProgressLine( + line: string, +): { percent: number; completed: number; total: number } | null { + const match = GIT_CHECKOUT_PROGRESS_LINE.exec(line); + if (!match) return null; + const percent = Number(match[1]); + const completed = Number(match[2]); + const total = Number(match[3]); + if (!Number.isFinite(percent) || !Number.isFinite(completed) || !Number.isFinite(total)) { + return null; + } + return { percent: Math.max(0, Math.min(100, percent)), completed, total }; +} + +const OUTPUT_LINE_SEPARATOR = /\r\n|\r|\n/; + const collectOutput = Effect.fnUntraced(function* ( input: Pick, stream: Stream.Stream, @@ -654,19 +673,21 @@ const collectOutput = Effect.fnUntraced(function* ( let lineBuffer = ""; let truncated = false; + // Git redraws progress with a bare `\r` between updates and only ends the + // line once the step is done, so `\r` has to count as a line break here. const emitCompleteLines = Effect.fnUntraced(function* (flush: boolean) { - let newlineIndex = lineBuffer.indexOf("\n"); - while (newlineIndex >= 0) { - const line = lineBuffer.slice(0, newlineIndex).replace(/\r$/, ""); - lineBuffer = lineBuffer.slice(newlineIndex + 1); + let separator = OUTPUT_LINE_SEPARATOR.exec(lineBuffer); + while (separator) { + const line = lineBuffer.slice(0, separator.index); + lineBuffer = lineBuffer.slice(separator.index + separator[0].length); if (line.length > 0 && onLine) { yield* onLine(line); } - newlineIndex = lineBuffer.indexOf("\n"); + separator = OUTPUT_LINE_SEPARATOR.exec(lineBuffer); } if (flush) { - const trailing = lineBuffer.replace(/\r$/, ""); + const trailing = lineBuffer; lineBuffer = ""; if (trailing.length > 0 && onLine) { yield* onLine(trailing); @@ -3010,7 +3031,7 @@ export const makeGitVcsDriverCore = Effect.fn("makeGitVcsDriverCore")(function* const createWorktree: GitVcsDriver.GitVcsDriver["Service"]["createWorktree"] = Effect.fn( "createWorktree", - )(function* (input) { + )(function* (input, options) { const targetBranch = input.newRefName ?? input.refName; const sanitizedBranch = targetBranch.replace(/\//g, "-"); const repoName = path.basename(input.cwd); @@ -3018,12 +3039,31 @@ export const makeGitVcsDriverCore = Effect.fn("makeGitVcsDriverCore")(function* const args = input.newRefName ? ["worktree", "add", "-b", input.newRefName, worktreePath, input.refName] : ["worktree", "add", worktreePath, input.refName]; + const progress = options?.progress; + const onCheckoutProgress = progress?.onCheckoutProgress; yield* executeGit("GitVcsDriver.createWorktree", input.cwd, args, { fallbackErrorDetail: "git worktree add failed", timeoutMs: WORKTREE_ADD_TIMEOUT_MS, + ...(onCheckoutProgress + ? { + // Git only prints checkout progress when stderr is a tty or the + // delay elapsed. GIT_PROGRESS_DELAY=0 forces it through the pipe. + env: { GIT_PROGRESS_DELAY: "0", LC_ALL: "C" }, + progress: { + onStderrLine: (line) => { + const parsed = parseGitCheckoutProgressLine(line); + return parsed ? onCheckoutProgress(parsed) : Effect.void; + }, + }, + } + : {}), }); + if (progress?.onWorktreeClaimed) { + yield* progress.onWorktreeClaimed(worktreePath); + } + // `git worktree add` leaves submodules empty, so a repo that keeps agent // skills, tooling or source in one gets a worktree that is quietly missing // them. Best-effort: the objects are usually already in the parent's @@ -3033,18 +3073,38 @@ export const makeGitVcsDriverCore = Effect.fn("makeGitVcsDriverCore")(function* .exists(path.join(worktreePath, ".gitmodules")) .pipe(Effect.orElseSucceed(() => false)); if (hasSubmodules) { - yield* runGit("GitVcsDriver.createWorktree.updateSubmodules", worktreePath, [ - "submodule", - "update", - "--init", - "--recursive", - ]).pipe( - Effect.catch((cause) => - Effect.logWarning("worktree submodule checkout failed; submodule paths are empty", { - worktreePath, - cause, - }), - ), + if (progress?.onSubmodulesStarted) { + yield* progress.onSubmodulesStarted(); + } + const onSubmoduleLine = progress?.onSubmoduleLine; + yield* runGit( + "GitVcsDriver.createWorktree.updateSubmodules", + worktreePath, + ["submodule", "update", "--init", "--recursive"], + onSubmoduleLine + ? { + env: { LC_ALL: "C" }, + progress: { onStdoutLine: onSubmoduleLine, onStderrLine: onSubmoduleLine }, + } + : {}, + ).pipe( + Effect.matchEffect({ + onFailure: (cause) => + Effect.logWarning("worktree submodule checkout failed; submodule paths are empty", { + worktreePath, + cause, + }).pipe( + Effect.andThen( + progress?.onSubmodulesFinished + ? progress.onSubmodulesFinished({ ok: false, detail: cause.message }) + : Effect.void, + ), + ), + onSuccess: () => + progress?.onSubmodulesFinished + ? progress.onSubmodulesFinished({ ok: true, detail: null }) + : Effect.void, + }), ); } @@ -3496,7 +3556,8 @@ export const makeGitVcsDriverCore = Effect.fn("makeGitVcsDriverCore")(function* getReviewDiffFileContents, readConfigValue, listRefs, - createWorktree: (input) => withListRefsInvalidation(input.cwd, createWorktree(input)), + createWorktree: (input, options) => + withListRefsInvalidation(input.cwd, createWorktree(input, options)), fetchPullRequestBranch: (input) => withListRefsInvalidation(input.cwd, fetchPullRequestBranch(input)), fetchPullRequestHeadCommit, diff --git a/apps/server/src/ws.ts b/apps/server/src/ws.ts index 1f960c488d8ab2b71a7b20d11628b1ef39c510ed..6ddd8c821c9f60da28ba1acb07f4233a928238cb 100644 --- a/apps/server/src/ws.ts +++ b/apps/server/src/ws.ts @@ -7,11 +7,13 @@ import * as Crypto from "effect/Crypto"; import * as DateTime from "effect/DateTime"; import * as Duration from "effect/Duration"; import * as Effect from "effect/Effect"; +import * as Fiber from "effect/Fiber"; import * as Layer from "effect/Layer"; import * as Option from "effect/Option"; import * as Path from "effect/Path"; import * as Queue from "effect/Queue"; import * as Ref from "effect/Ref"; +import * as Schedule from "effect/Schedule"; import * as Schema from "effect/Schema"; import * as Stream from "effect/Stream"; import { @@ -129,6 +131,7 @@ import * as GitWorkflowService from "./git/GitWorkflowService.ts"; import { linkCreatedPullRequest } from "./git/linkCreatedPullRequest.ts"; import * as ReviewService from "./review/ReviewService.ts"; import * as ProjectSetupScriptRunner from "./project/ProjectSetupScriptRunner.ts"; +import * as WorktreeSetupTracker from "./project/WorktreeSetupTracker.ts"; import * as AgentSessionScanner from "./project/AgentSessionScanner.ts"; import { importRecentAgentThreads } from "./project/AgentSessionImporter.ts"; import * as ServerEnvironment from "./environment/ServerEnvironment.ts"; @@ -597,6 +600,7 @@ const makeWsRpcLayer = ( return true; }); const projectSetupScriptRunner = yield* ProjectSetupScriptRunner.ProjectSetupScriptRunner; + const worktreeSetupTracker = yield* WorktreeSetupTracker.WorktreeSetupTracker; const agentSessionScanner = yield* AgentSessionScanner.AgentSessionScanner; const serverEnvironment = yield* ServerEnvironment.ServerEnvironment; const backgroundPolicy = yield* BackgroundPolicy.BackgroundPolicy; @@ -991,6 +995,9 @@ const makeWsRpcLayer = ( let targetProjectId = bootstrap?.createThread?.projectId; let targetProjectCwd = bootstrap?.prepareWorktree?.projectCwd; let targetWorktreePath = bootstrap?.createThread?.worktreePath ?? null; + // The setup script's terminal, once started. Cancel closes only this + // one so terminals the user opened meanwhile survive. + let setupTerminalId: string | null = null; const cleanupCreatedThread = () => createdThread @@ -1083,19 +1090,37 @@ const makeWsRpcLayer = ( ); }); + const tracked = bootstrap?.prepareWorktree !== undefined; + const threadId = command.threadId; + const track = (effect: Effect.Effect) => (tracked ? effect : Effect.void); + + // Runs the setup script and, for tracked bootstraps, waits for it to + // exit so the card can show the exit code and the agent stage never + // starts on a half-installed tree. Untracked callers keep the old + // fire-and-forget behavior. const runSetupProgram = () => Effect.gen(function* () { if (!bootstrap?.runSetupScript || !targetWorktreePath) { + yield* track(worktreeSetupTracker.stageStatus(threadId, "setup-script", "skipped")); return; } const worktreePath = targetWorktreePath; const requestedAt = yield* nowIso; - yield* projectSetupScriptRunner + yield* track(worktreeSetupTracker.stageStatus(threadId, "setup-script", "running")); + const setupResult = yield* projectSetupScriptRunner .runForThread({ - threadId: command.threadId, + threadId, ...(targetProjectId ? { projectId: targetProjectId } : {}), ...(targetProjectCwd ? { projectCwd: targetProjectCwd } : {}), worktreePath, + ...(tracked + ? { + observeCompletion: { + onOutputLine: (line) => + worktreeSetupTracker.appendTail(threadId, "setup-script", line), + }, + } + : {}), }) .pipe( Effect.matchEffect({ @@ -1104,21 +1129,71 @@ const makeWsRpcLayer = ( error, requestedAt, worktreePath, - }), + }).pipe( + Effect.andThen( + track( + worktreeSetupTracker.stageStatus( + threadId, + "setup-script", + "failed", + "failed to start", + ), + ), + ), + Effect.as(null), + ), onSuccess: (setupResult) => { if (setupResult.status !== "started") { - return Effect.void; + return track( + worktreeSetupTracker.stageStatus( + threadId, + "setup-script", + "skipped", + "no setup script", + ), + ).pipe(Effect.as(null)); } + setupTerminalId = setupResult.terminalId; return recordSetupScriptStarted({ requestedAt, worktreePath, scriptId: setupResult.scriptId, scriptName: setupResult.scriptName, terminalId: setupResult.terminalId, - }); + }).pipe( + Effect.andThen( + track( + worktreeSetupTracker.update(threadId, (snapshot) => ({ + ...snapshot, + setupScript: { + name: setupResult.scriptName, + command: setupResult.scriptCommand, + terminalId: setupResult.terminalId, + }, + })), + ), + ), + Effect.as(setupResult), + ); }, }), ); + if (!tracked || !setupResult?.completion) { + return; + } + // The setup script is best effort, like the untracked path: a + // failed install must not throw away the worktree the user just + // waited for. The card keeps the failed stage and its terminal. + const completion = yield* setupResult.completion; + if (completion.exitCode === 0) { + yield* worktreeSetupTracker.stageStatus(threadId, "setup-script", "done"); + return; + } + const detail = + completion.exitCode === null + ? "terminal closed before the script finished" + : `exit ${completion.exitCode}`; + yield* worktreeSetupTracker.stageStatus(threadId, "setup-script", "failed", detail); }); const bootstrapProgram = Effect.gen(function* () { @@ -1138,6 +1213,7 @@ const makeWsRpcLayer = ( remoteName: "origin", })); if (startFromOrigin) { + yield* track(worktreeSetupTracker.stageStatus(threadId, "fetch", "running")); yield* gitWorkflow.fetchRemote({ cwd: prepareWorktree.projectCwd, remoteName: "origin", @@ -1154,7 +1230,26 @@ const makeWsRpcLayer = ( fallbackRemoteName: "origin", }); worktreeBaseRef = resolvedRemoteBase.commitSha; + yield* track( + worktreeSetupTracker.stageStatus( + threadId, + "fetch", + "done", + `origin/${prepareWorktree.baseBranch} at ${resolvedRemoteBase.commitSha.slice(0, 7)}`, + ), + ); + } else { + yield* track( + worktreeSetupTracker.stageStatus( + threadId, + "fetch", + "warning", + `origin/${prepareWorktree.baseBranch} not found, using local branch`, + ), + ); } + } else { + yield* track(worktreeSetupTracker.stageStatus(threadId, "fetch", "skipped")); } const resolvedWorktreeBaseRef = worktreeBaseRef ?? prepareWorktree.baseBranch; @@ -1163,6 +1258,27 @@ const makeWsRpcLayer = ( refName: resolvedWorktreeBaseRef, }); worktreeBaseRef = resolvedWorktreeBaseRef; + yield* track( + worktreeSetupTracker.update(threadId, (snapshot) => ({ + ...snapshot, + baseRef: resolvedWorktreeBaseRef, + })), + ); + } + + if (prepareWorktree && !shouldPrepareWorktree) { + // Not a git repo, or the base has no commit: the thread runs in + // the project checkout instead. The card says so and moves on. + yield* track( + worktreeSetupTracker.update(threadId, (snapshot) => ({ + ...snapshot, + stages: snapshot.stages.map((stage) => + stage.id === "fetch" || stage.id === "checkout" || stage.id === "submodules" + ? { ...stage, status: "skipped", detail: "using project checkout" } + : stage, + ), + })), + ); } if (bootstrap?.createThread) { @@ -1188,18 +1304,92 @@ const makeWsRpcLayer = ( } if (prepareWorktree && shouldPrepareWorktree && worktreeBaseRef) { - const worktree = yield* gitWorkflow.createWorktree({ - cwd: prepareWorktree.projectCwd, - refName: worktreeBaseRef, - newRefName: prepareWorktree.branch, - baseRefName: prepareWorktree.baseBranch, - path: null, - }); + yield* worktreeSetupTracker.stageStatus(threadId, "checkout", "running"); + let checkoutTotal: number | null = null; + const worktree = yield* gitWorkflow.createWorktree( + { + cwd: prepareWorktree.projectCwd, + refName: worktreeBaseRef, + newRefName: prepareWorktree.branch, + baseRefName: prepareWorktree.baseBranch, + path: null, + }, + { + progress: { + // Git has registered the directory at this point, so a + // cancel during the submodule step can still remove it. + onWorktreeClaimed: (path) => + Effect.sync(() => { + targetWorktreePath = path; + }), + onCheckoutProgress: ({ percent, completed, total }) => { + checkoutTotal = total; + return worktreeSetupTracker.stage(threadId, "checkout", { + percent, + detail: `${completed.toLocaleString("en-US")} / ${total.toLocaleString("en-US")} files`, + }); + }, + onSubmodulesStarted: () => + worktreeSetupTracker + .stageStatus( + threadId, + "checkout", + "done", + checkoutTotal === null + ? null + : `${checkoutTotal.toLocaleString("en-US")} files`, + ) + .pipe( + Effect.andThen( + worktreeSetupTracker.stageStatus(threadId, "submodules", "running"), + ), + ), + onSubmoduleLine: (line) => { + const submodulePath = /Submodule path '([^']+)'/.exec(line)?.[1]; + return submodulePath === undefined + ? Effect.void + : worktreeSetupTracker.stage(threadId, "submodules", { + detail: submodulePath, + }); + }, + onSubmodulesFinished: ({ ok, detail }) => + worktreeSetupTracker.stageStatus( + threadId, + "submodules", + ok ? "done" : "warning", + ok ? undefined : (detail ?? "submodule checkout failed"), + ), + }, + }, + ); + const checkoutEndedAt = yield* nowIso; + yield* worktreeSetupTracker.update(threadId, (snapshot) => ({ + ...snapshot, + worktreePath: worktree.worktree.path, + stages: snapshot.stages.map((stage) => { + if (stage.id === "checkout" && stage.status === "running") { + return { + ...stage, + status: "done", + percent: 100, + endedAt: checkoutEndedAt, + detail: + checkoutTotal === null + ? stage.detail + : `${checkoutTotal.toLocaleString("en-US")} files`, + }; + } + if (stage.id === "submodules" && stage.status === "pending") { + return { ...stage, status: "skipped", detail: "none" }; + } + return stage; + }), + })); targetWorktreePath = worktree.worktree.path; yield* dispatchFromClient({ type: "thread.meta.update", commandId: yield* serverCommandId("bootstrap-thread-meta-update"), - threadId: command.threadId, + threadId, branch: worktree.worktree.refName, worktreePath: targetWorktreePath, }); @@ -1208,36 +1398,114 @@ const makeWsRpcLayer = ( yield* runSetupProgram(); - return yield* dispatchFromClient(finalTurnStartCommand); + yield* track(worktreeSetupTracker.stageStatus(threadId, "agent", "running")); + // Past this point a cancel would roll back a thread whose turn has + // started. Drop the cancel handle and make the handoff atomic. + yield* track(worktreeSetupTracker.markUncancellable(threadId)); + const started = yield* Effect.uninterruptible( + dispatchFromClient(finalTurnStartCommand), + ); + yield* track( + worktreeSetupTracker + .stageStatus(threadId, "agent", "done") + .pipe(Effect.andThen(worktreeSetupTracker.finish(threadId, "done"))), + ); + return started; }); - return yield* bootstrapProgram.pipe( + const runBootstrap = tracked + ? Effect.gen(function* () { + const fiber = yield* Effect.forkChild(bootstrapProgram); + yield* worktreeSetupTracker.begin({ + threadId, + branch: bootstrap?.prepareWorktree?.branch ?? null, + baseRef: bootstrap?.prepareWorktree?.baseBranch ?? null, + stages: ["fetch", "checkout", "submodules", "setup-script", "agent"], + fiber, + }); + return yield* Fiber.join(fiber); + }) + : bootstrapProgram; + + const cleanupAndFail = ( + cause: Cause.Cause, + dispatchError: OrchestrationDispatchCommandError, + ) => + Effect.uninterruptible(cleanupCreatedThread()).pipe( + Effect.matchCauseEffect({ + onFailure: (cleanupCause) => + Effect.logWarning("bootstrap thread cleanup failed", { + threadId, + detail: Cause.pretty(cleanupCause), + }).pipe(Effect.flatMap(() => Effect.fail(dispatchError))), + onSuccess: (threadDeleted) => + Effect.fail( + threadDeleted + ? new OrchestrationDispatchCommandError({ + message: dispatchError.message, + ...(dispatchError.cause !== undefined + ? { cause: dispatchError.cause } + : {}), + bootstrapThreadDisposition: "deleted", + }) + : dispatchError, + ), + }), + ); + + return yield* runBootstrap.pipe( Effect.catchCause((cause) => { const dispatchError = toBootstrapDispatchCommandCauseError(cause); if (Cause.hasInterruptsOnly(cause)) { - return Effect.fail(dispatchError); + // A user cancel interrupts the forked bootstrap fiber. The + // created thread is rolled back like any other failure so the + // draft returns to the composer. The setup terminal is closed + // first so a still-running script cannot hold files open in + // the worktree while git removes it. Closing kills the + // process asynchronously, so the removal retries briefly. + const closeSetupTerminal = setupTerminalId + ? terminalManager.close({ + threadId, + terminalId: setupTerminalId, + deleteHistory: true, + }) + : Effect.void; + const removeCreatedWorktree = + tracked && targetWorktreePath && bootstrap?.prepareWorktree + ? closeSetupTerminal.pipe( + Effect.ignoreCause({ log: true }), + Effect.andThen( + gitWorkflow + .removeWorktree({ + cwd: bootstrap.prepareWorktree.projectCwd, + path: targetWorktreePath, + force: true, + }) + .pipe( + Effect.retry({ times: 4, schedule: Schedule.spaced("500 millis") }), + ), + ), + Effect.ignoreCause({ log: true }), + Effect.uninterruptible, + ) + : Effect.void; + return track(worktreeSetupTracker.finish(threadId, "cancelled")).pipe( + Effect.andThen(removeCreatedWorktree), + Effect.andThen( + tracked + ? cleanupAndFail( + cause, + new OrchestrationDispatchCommandError({ + message: "Worktree setup cancelled.", + }), + ) + : Effect.fail(dispatchError), + ), + ); } - return Effect.uninterruptible(cleanupCreatedThread()).pipe( - Effect.matchCauseEffect({ - onFailure: (cleanupCause) => - Effect.logWarning("bootstrap thread cleanup failed", { - threadId: command.threadId, - detail: Cause.pretty(cleanupCause), - }).pipe(Effect.flatMap(() => Effect.fail(dispatchError))), - onSuccess: (threadDeleted) => - Effect.fail( - threadDeleted - ? new OrchestrationDispatchCommandError({ - message: dispatchError.message, - ...(dispatchError.cause !== undefined - ? { cause: dispatchError.cause } - : {}), - bootstrapThreadDisposition: "deleted", - }) - : dispatchError, - ), - }), - ); + return track( + worktreeSetupTracker.finish(threadId, "failed", dispatchError.message), + ).pipe(Effect.andThen(cleanupAndFail(cause, dispatchError))); }), ); }); @@ -2614,6 +2882,20 @@ const makeWsRpcLayer = ( "rpc.aggregate": "vcs", }, ), + [WS_METHODS.subscribeWorktreeSetup]: (input) => + observeRpcStream( + WS_METHODS.subscribeWorktreeSetup, + worktreeSetupTracker.stream(input.threadId), + { "rpc.aggregate": "vcs" }, + ), + [WS_METHODS.worktreeSetupCancel]: (input) => + observeRpcEffect( + WS_METHODS.worktreeSetupCancel, + worktreeSetupTracker + .cancel(input.threadId) + .pipe(Effect.map((cancelled) => ({ cancelled }))), + { "rpc.aggregate": "vcs" }, + ), [WS_METHODS.vcsRefreshStatus]: (input) => observeRpcEffect( WS_METHODS.vcsRefreshStatus, diff --git a/apps/web/src/components/ChatView.logic.test.ts b/apps/web/src/components/ChatView.logic.test.ts index be279f220e7e0cb36da23e807bb07d8004f0c273..6f2a9e1199f838c7fc99875d7bda5828683c7ce9 100644 --- a/apps/web/src/components/ChatView.logic.test.ts +++ b/apps/web/src/components/ChatView.logic.test.ts @@ -597,6 +597,30 @@ describe("draft hero submission transition", () => { ).toBe(false); }); + it("leaves the hero layout while a worktree setup card is on the timeline", () => { + expect( + resolveDraftHeroState({ + isLocalDraftThread: true, + hasTimelineEntries: false, + isWorking: false, + draftHeroDockRequested: false, + backgroundSubmissionPending: false, + hasWorktreeSetupCard: true, + }), + ).toBe(false); + // A background submission normally pins the hero, but never over the card. + expect( + resolveDraftHeroState({ + isLocalDraftThread: true, + hasTimelineEntries: false, + isWorking: false, + draftHeroDockRequested: false, + backgroundSubmissionPending: true, + hasWorktreeSetupCard: true, + }), + ).toBe(false); + }); + it("keeps the composer in the hero layout until navigation after server promotion", () => { expect( resolveDraftHeroState({ diff --git a/apps/web/src/components/ChatView.logic.ts b/apps/web/src/components/ChatView.logic.ts index ed1422a71109c95a678c0e0be697936b84a38956..eae137201d3725719057fa4512ddd00a77c6547f 100644 --- a/apps/web/src/components/ChatView.logic.ts +++ b/apps/web/src/components/ChatView.logic.ts @@ -257,7 +257,12 @@ export function resolveDraftHeroState(input: { isWorking: boolean; draftHeroDockRequested: boolean; backgroundSubmissionPending: boolean; + /** A worktree setup card is on the timeline, so the timeline must stay visible. */ + hasWorktreeSetupCard?: boolean; }): boolean { + if (input.hasWorktreeSetupCard) { + return false; + } if (input.backgroundSubmissionPending) { return true; } diff --git a/apps/web/src/components/ChatView.tsx b/apps/web/src/components/ChatView.tsx index 9e11132269e6bdae9284315e40776b04da5b5ea3..574ed0f372a1bb16e2b6d5153d07e8af43bb0dee 100644 --- a/apps/web/src/components/ChatView.tsx +++ b/apps/web/src/components/ChatView.tsx @@ -43,6 +43,7 @@ import { resolveEnvironmentMachineKind, RuntimeMode, TerminalOpenInput, + type WorktreeSetupSnapshot, } from "@t3tools/contracts"; import { type EnvironmentConnectionPresentation } from "@t3tools/client-runtime/connection"; import { wasBootstrapThreadDeleted } from "@t3tools/client-runtime/errors"; @@ -1645,6 +1646,27 @@ export default function ChatView(props: ChatViewProps) { return () => revokeBlobPreviewUrl(src); }, [expandedImage]); const [optimisticUserMessages, setOptimisticUserMessages] = useState([]); + // The bootstrap worktree setup this composer last dispatched. Set when a + // worktree send starts and cleared once the turn starts or the next send + // begins, so a failed or cancelled card stays until the user acts. + const [worktreeSetupRef, setWorktreeSetupRef] = useState<{ + environmentId: EnvironmentId; + threadId: ThreadId; + ownerKey: string; + } | null>(null); + const [heldWorktreeSetup, setHeldWorktreeSetup] = useState(null); + // Set by "Work locally": the draft whose restored message should be resent + // once the cancelled dispatch has settled and the draft is in local mode. + // Keyed by draft id so a bootstrap rotating the thread id keeps it, while + // moving to another draft drops it without an effect. + const [workLocallyResendDraftId, setWorkLocallyResendDraftId] = useState(null); + // The draft route reuses this component across drafts, so a resend recorded + // for one draft must not fire when the user comes back to it later. + useEffect(() => { + if (workLocallyResendDraftId !== null && workLocallyResendDraftId !== draftId) { + setWorkLocallyResendDraftId(null); + } + }, [draftId, workLocallyResendDraftId]); const [feedbackSubmissionsByThreadKey, setFeedbackSubmissionsByThreadKey] = useState< Record> >({}); @@ -3337,6 +3359,64 @@ export default function ChatView(props: ChatViewProps) { activeThreadKey, ); const displayedThreadRef = parseScopedThreadKey(displayedTimelineKey); + // Live stages of a bootstrap worktree setup. The subscription follows the + // thread that was set up, not the route: a deleted bootstrap thread rotates + // the draft's thread id, and the failed card must survive that. + const worktreeSetupOwnerKey = draftId ?? routeThreadKey; + const worktreeSetupActive = + worktreeSetupRef !== null && worktreeSetupRef.ownerKey === worktreeSetupOwnerKey; + // The setup runs on the environment that received the dispatch, so both + // the subscription and cancel target that one even if the draft's machine + // picker changes underneath. + const worktreeSetupQuery = useEnvironmentQuery( + worktreeSetupActive + ? vcsEnvironment.worktreeSetup({ + environmentId: worktreeSetupRef.environmentId, + input: { threadId: worktreeSetupRef.threadId }, + }) + : null, + ); + const latestWorktreeSetup = worktreeSetupQuery.data; + useEffect(() => { + // The server drops finished snapshots after a grace period and emits null. + // Hold the last real snapshot so a settled card does not vanish. + if (latestWorktreeSetup) setHeldWorktreeSetup(latestWorktreeSetup); + }, [latestWorktreeSetup]); + const worktreeSetup = + worktreeSetupActive && heldWorktreeSetup?.threadId === worktreeSetupRef.threadId + ? heldWorktreeSetup + : null; + // A finished card is dropped once the agent's turn shows in the timeline: + // the card belongs to the send, and the agent takes over from there. + const worktreeSetupDoneAndTurnVisible = + worktreeSetup?.phase === "done" && activeThread?.latestTurn?.startedAt != null; + useEffect(() => { + if (!worktreeSetupDoneAndTurnVisible) return; + setWorktreeSetupRef(null); + setHeldWorktreeSetup(null); + }, [worktreeSetupDoneAndTurnVisible]); + const cancelWorktreeSetup = useAtomCommand(vcsEnvironment.cancelWorktreeSetup, { + reportFailure: false, + }); + const onCancelWorktreeSetup = useCallback(() => { + if (!worktreeSetup || !worktreeSetupRef || worktreeSetup.phase !== "running") return; + void cancelWorktreeSetup({ + environmentId: worktreeSetupRef.environmentId, + input: { threadId: worktreeSetup.threadId }, + }); + }, [cancelWorktreeSetup, worktreeSetup, worktreeSetupRef]); + // The setup terminal belongs to the thread that was set up. A failed + // bootstrap deletes that thread and closes its terminals, so only offer the + // terminal while the setup thread is still the active one. + const onOpenWorktreeSetupTerminal = useMemo(() => { + if (!worktreeSetup || !activeThreadRef || worktreeSetup.threadId !== activeThreadRef.threadId) { + return null; + } + const setupThreadRef = activeThreadRef; + return (terminalId: string) => { + storeEnsureTerminal(setupThreadRef, terminalId, { open: true, active: true }); + }; + }, [activeThreadRef, storeEnsureTerminal, worktreeSetup]); const [dockedDraftHeroThreadKey, setDockedDraftHeroThreadKey] = useState(null); const draftHeroDockRequested = activeThreadKey !== null && dockedDraftHeroThreadKey === activeThreadKey; @@ -3346,6 +3426,9 @@ export default function ChatView(props: ChatViewProps) { isWorking, draftHeroDockRequested, backgroundSubmissionPending, + // A cancelled or failed setup card stays on the draft's timeline; the + // hero headline would paint over it. + hasWorktreeSetupCard: worktreeSetup !== null, }); const [ attachDraftHeroTransitionGroupRef, @@ -7279,6 +7362,11 @@ export default function ChatView(props: ChatViewProps) { preparingWorktree: Boolean(baseBranchForWorktree), submissionIntent: resolvedSubmissionIntent, }); + setWorktreeSetupRef( + baseBranchForWorktree + ? { environmentId, threadId: threadIdForSend, ownerKey: worktreeSetupOwnerKey } + : null, + ); const messageIdForSend = newMessageId(); const messageCreatedAt = new Date().toISOString(); @@ -8365,6 +8453,70 @@ export default function ChatView(props: ChatViewProps) { ], ); + // "Work locally" on the setup card: cancel the bootstrap and remember the + // draft. The cancelled dispatch deletes the half-made thread and puts the + // message back in the composer; the effect below then flips the draft to + // local mode and resends. The draft is a server thread for the whole + // setup (the bootstrap created it), so this keys off the route, not + // `isLocalDraftThread`. + const onWorktreeSetupWorkLocally = useCallback(() => { + if (!worktreeSetup || !worktreeSetupRef || worktreeSetup.phase !== "running" || !draftId) { + return; + } + const target = { + environmentId: worktreeSetupRef.environmentId, + input: { threadId: worktreeSetup.threadId }, + }; + void (async () => { + const result = await cancelWorktreeSetup(target); + if (result._tag !== "Success" || !result.value.cancelled) return; + setWorkLocallyResendDraftId(draftId); + })(); + }, [cancelWorktreeSetup, draftId, worktreeSetup, worktreeSetupRef]); + const onSendRef = useRef(onSend); + onSendRef.current = onSend; + // Resend once the cancelled dispatch has settled and the composer is free. + // Every state that makes `onSend` bail and wait is part of the readiness + // check, so the flag survives a reconnect, a reverting checkpoint, or a + // feedback upload in between. What remains inside `onSend` are the checks + // that need the user to change something, and those should not auto retry. + const workLocallyResendReady = + workLocallyResendDraftId !== null && + workLocallyResendDraftId === draftId && + isLocalDraftThread && + !isSendBusy && + !isConnecting && + !isRevertingCheckpoint && + !threadDetailLoading && + clientSettingsHydrated && + !needsLoadBalancing && + !activeEnvironmentUnavailable && + !activePendingProgress && + !feedbackUploading; + useEffect(() => { + if ( + !workLocallyResendReady || + sendInFlightRef.current || + feedbackUploadsInFlightRef.current.has(routeThreadKey) + ) { + return; + } + if (sendEnvMode !== "local") { + // The draft is back; switch it to the project checkout and let the next + // render resend. + setDraftThreadContext(composerDraftTarget, { envMode: "local", startFromOrigin: false }); + return; + } + setWorkLocallyResendDraftId(null); + void onSendRef.current(); + }, [ + composerDraftTarget, + routeThreadKey, + sendEnvMode, + setDraftThreadContext, + workLocallyResendReady, + ]); + const onStartFromOriginChange = (nextStartFromOrigin: boolean) => { if (canOverrideServerThreadEnvMode && activeThread) { setPendingServerThreadStartFromOriginByThreadId((current) => @@ -8808,6 +8960,10 @@ export default function ChatView(props: ChatViewProps) { isPreparingWorktree={!paintOnlyDisplayedTimeline && isPreparingWorktree} isCompacting={!paintOnlyDisplayedTimeline && isCompacting} activeTurnStartedAt={paintOnlyDisplayedTimeline ? null : activeWorkStartedAt} + worktreeSetup={paintOnlyDisplayedTimeline ? null : worktreeSetup} + onCancelWorktreeSetup={onCancelWorktreeSetup} + {...(draftId ? { onWorktreeSetupWorkLocally } : {})} + {...(onOpenWorktreeSetupTerminal ? { onOpenWorktreeSetupTerminal } : {})} listRef={legendListRef} timelineEntries={displayedTimeline.entries} latestTurn={paintOnlyDisplayedTimeline ? null : activeLatestTurn} diff --git a/apps/web/src/components/chat/MessagesTimeline.logic.test.ts b/apps/web/src/components/chat/MessagesTimeline.logic.test.ts index 5fbc686227d9c4e2a41c54833d782929ebd96b56..59b4a0c9856bebe609d0b27f9e4e7e3a8b3c6cf5 100644 --- a/apps/web/src/components/chat/MessagesTimeline.logic.test.ts +++ b/apps/web/src/components/chat/MessagesTimeline.logic.test.ts @@ -10,6 +10,7 @@ import { ThreadId, TurnId, type OrchestrationThread, + type WorktreeSetupSnapshot, } from "@t3tools/contracts"; import { applyThreadDetailEvent, @@ -31,6 +32,7 @@ import { shouldPreserveAssistantLineBreaks, type MessagesTimelineRow, type MessagesTimelineRowsProjection, + WORKTREE_SETUP_ROW_ID, workEntryDisplayLabel, } from "./MessagesTimeline.logic"; import { @@ -1090,6 +1092,82 @@ describe("resolveAssistantMessageCopyState", () => { }); describe("deriveMessagesTimelineRows", () => { + it("shows the worktree setup card instead of the working placeholder", () => { + const snapshot: WorktreeSetupSnapshot = { + threadId: ThreadId.make("thread-setup"), + phase: "running", + startedAt: "2026-01-01T00:00:00Z", + endedAt: null, + branch: "feature", + baseRef: "main", + worktreePath: null, + setupScript: null, + stages: [], + error: null, + sequence: 3, + }; + const userEntry = { + id: "user-entry", + kind: "message", + createdAt: "2026-01-01T00:00:00Z", + message: { + id: "user-1" as never, + role: "user", + text: "Build it", + turnId: null, + createdAt: "2026-01-01T00:00:00Z", + updatedAt: "2026-01-01T00:00:00Z", + streaming: false, + }, + } as const; + const assistantEntry = { + id: "assistant-entry", + kind: "message", + createdAt: "2026-01-01T00:00:30Z", + message: { + id: "assistant-1" as never, + role: "assistant", + text: "On it", + turnId: "turn-1" as never, + createdAt: "2026-01-01T00:00:30Z", + updatedAt: "2026-01-01T00:00:30Z", + streaming: true, + }, + } as const; + const withoutMessages = deriveMessagesTimelineRows({ + timelineEntries: [], + isWorking: true, + activeTurnStartedAt: "2026-01-01T00:00:00Z", + turnDiffSummaries: [], + supportsConversationRollback: false, + worktreeSetup: snapshot, + }); + expect(withoutMessages).toEqual([ + { + kind: "worktree-setup", + id: WORKTREE_SETUP_ROW_ID, + createdAt: "2026-01-01T00:00:00Z", + snapshot, + }, + ]); + + // Once the agent has replied the finished card stays under the send. + const withMessages = deriveMessagesTimelineRows({ + timelineEntries: [userEntry, assistantEntry], + isWorking: true, + activeTurnStartedAt: "2026-01-01T00:00:00Z", + turnDiffSummaries: [], + supportsConversationRollback: false, + worktreeSetup: { ...snapshot, phase: "done" }, + }); + expect(withMessages.map((row) => row.kind)).toEqual([ + "message", + "worktree-setup", + "working", + "message", + ]); + }); + it("keeps context compaction visible outside folded work", () => { const rows = deriveMessagesTimelineRows({ timelineEntries: [ diff --git a/apps/web/src/components/chat/MessagesTimeline.logic.ts b/apps/web/src/components/chat/MessagesTimeline.logic.ts index 8dcb539fb024f2534277f0f234a7564552d12c27..3d7b1e12284eefca4ad5468092f066d4bcda664d 100644 --- a/apps/web/src/components/chat/MessagesTimeline.logic.ts +++ b/apps/web/src/components/chat/MessagesTimeline.logic.ts @@ -28,7 +28,12 @@ import { type WorkLogEntry, } from "../../session-logic"; import { type ChatMessage, type ProposedPlan, type TurnDiffSummary } from "../../types"; -import { type MessageId, type OrchestrationLatestTurn, type TurnId } from "@t3tools/contracts"; +import { + type MessageId, + type OrchestrationLatestTurn, + type TurnId, + type WorktreeSetupSnapshot, +} from "@t3tools/contracts"; import { formatWorkspaceRelativePath } from "../../filePathDisplay"; const TIMELINE_MINIMAP_ITEM_SPACING = 8; @@ -389,6 +394,12 @@ export type MessagesTimelineRow = kind: "thinking"; id: string; createdAt: string | null; + } + | { + kind: "worktree-setup"; + id: string; + createdAt: string | null; + snapshot: WorktreeSetupSnapshot; }; export interface StableMessagesTimelineRowsState { @@ -857,6 +868,8 @@ export function deriveMessagesTimelineRows(input: { supportsConversationRollback: boolean; /** Task ids of subagents still working, used by the active tool indicator. */ liveAgentTaskIds?: ReadonlySet | undefined; + /** Live bootstrap progress. Renders a stage card under the first user message. */ + worktreeSetup?: WorktreeSetupSnapshot | null; }): MessagesTimelineRow[] { const turnDiffSummaryByAssistantMessageId = new Map(); for (const summary of input.turnDiffSummaries) { @@ -1247,6 +1260,30 @@ export function deriveMessagesTimelineRows(input: { }); } + // The setup card takes the place of the working and thinking placeholders + // while a worktree is being prepared. It stays after the setup settles so a + // failure and its actions remain visible until the thread state moves on. + if (input.worktreeSetup) { + const setupRow = { + kind: "worktree-setup", + id: WORKTREE_SETUP_ROW_ID, + createdAt: input.worktreeSetup.startedAt, + snapshot: input.worktreeSetup, + } as const; + // Sit directly under the first user message: a finished snapshot can + // outlive the first assistant reply, and it belongs to the send, not the + // end of the thread. + const firstUserRowIndex = nextRows.findIndex( + (row) => row.kind === "message" && row.message.role === "user", + ); + if (firstUserRowIndex >= 0) { + nextRows.splice(firstUserRowIndex + 1, 0, setupRow); + } else { + nextRows.push(setupRow); + } + return attachTrailingToolGroupsToAssistant(nextRows); + } + if (input.isWorking && activeTurnHeaderIndex === input.timelineEntries.length) { appendWorkingRow(); } @@ -1261,6 +1298,8 @@ export function deriveMessagesTimelineRows(input: { return attachTrailingToolGroupsToAssistant(nextRows); } +export const WORKTREE_SETUP_ROW_ID = "worktree-setup-row"; + type MessagesTimelineRowsInput = Parameters[0]; export interface MessagesTimelineRowsProjection { @@ -1363,6 +1402,8 @@ function isRowUnchanged(a: MessagesTimelineRow, b: MessagesTimelineRow): boolean case "working": case "thinking": return a.createdAt === (b as typeof a).createdAt; + case "worktree-setup": + return a.snapshot === (b as typeof a).snapshot; case "assistant-meta": { const bm = b as typeof a; diff --git a/apps/web/src/components/chat/MessagesTimeline.tsx b/apps/web/src/components/chat/MessagesTimeline.tsx index 084287080bd9819cc0804a19fe9b5a323f25b6f9..272ad320ad2b7df339da9fd8169b8e56718d97a6 100644 --- a/apps/web/src/components/chat/MessagesTimeline.tsx +++ b/apps/web/src/components/chat/MessagesTimeline.tsx @@ -19,6 +19,7 @@ import { type ServerProviderSkill, type ToolActivityIcon, type TurnId, + type WorktreeSetupSnapshot, } from "@t3tools/contracts"; import { parseScopedThreadKey } from "@t3tools/client-runtime/environment"; import { replaceComposerContextReferences } from "@t3tools/shared/composerContextReferences"; @@ -188,6 +189,7 @@ import { } from "./MessagesTimeline.logic"; import { TerminalContextInlineChip } from "./TerminalContextInlineChip"; import { Tooltip, TooltipPopup, TooltipTrigger } from "../ui/tooltip"; +import { WorktreeSetupCard } from "./WorktreeSetupCard"; import { ContextChipPopover as UserMessageContextPopover, ContextChipShell, @@ -275,6 +277,9 @@ interface TimelineRowSharedState { agentPanelModel: AgentPanelModel; expandedSpawnEntryIds: ReadonlySet; onOpenAgents: () => void; + onCancelWorktreeSetup: (() => void) | null; + onWorktreeSetupWorkLocally: (() => void) | null; + onOpenWorktreeSetupTerminal: ((terminalId: string) => void) | null; } interface TimelineRowActivityState { @@ -374,6 +379,11 @@ interface MessagesTimelineProps { isPreparingWorktree?: boolean; isCompacting?: boolean; activeTurnStartedAt: string | null; + /** Live bootstrap progress for this thread, or null when none is tracked. */ + worktreeSetup?: WorktreeSetupSnapshot | null; + onCancelWorktreeSetup?: () => void; + onWorktreeSetupWorkLocally?: () => void; + onOpenWorktreeSetupTerminal?: (terminalId: string) => void; listRef: React.RefObject; timelineEntries: ReturnType; latestTurn: TimelineLatestTurn | null; @@ -433,6 +443,10 @@ export const MessagesTimeline = memo(function MessagesTimeline({ citationHistoryLoading = false, onCiteAssistantText, isWorking, + worktreeSetup = null, + onCancelWorktreeSetup, + onWorktreeSetupWorkLocally, + onOpenWorktreeSetupTerminal, isPreparingWorktree = false, isCompacting = false, activeTurnStartedAt, @@ -692,6 +706,7 @@ export const MessagesTimeline = memo(function MessagesTimeline({ turnDiffSummaries, supportsConversationRollback, liveAgentTaskIds, + worktreeSetup, }, previous?.threadKey === listIdentityKey && previous.workspaceRoot === workspaceRoot ? previous.projection @@ -713,6 +728,7 @@ export const MessagesTimeline = memo(function MessagesTimeline({ turnDiffSummaries, supportsConversationRollback, liveAgentTaskIds, + worktreeSetup, ]); const rows = useStableRows(rawRows, listIdentityKey); const minimapItems = useMemo(() => deriveTimelineMinimapItems(rows), [rows]); @@ -905,6 +921,9 @@ export const MessagesTimeline = memo(function MessagesTimeline({ agentPanelModel: agentPanelModel ?? EMPTY_AGENT_PANEL_MODEL, expandedSpawnEntryIds: paintedExpandedSpawnEntryIds, onOpenAgents, + onCancelWorktreeSetup: onCancelWorktreeSetup ?? null, + onWorktreeSetupWorkLocally: onWorktreeSetupWorkLocally ?? null, + onOpenWorktreeSetupTerminal: onOpenWorktreeSetupTerminal ?? null, }), [ readyCitationRequest, @@ -932,6 +951,9 @@ export const MessagesTimeline = memo(function MessagesTimeline({ agentPanelModel, paintedExpandedSpawnEntryIds, onOpenAgents, + onCancelWorktreeSetup, + onWorktreeSetupWorkLocally, + onOpenWorktreeSetupTerminal, ], ); const activityState = useMemo( @@ -1386,7 +1408,8 @@ const TimelineRowContent = memo(function TimelineRowContent({ row }: { row: Time row.kind === "work" || row.kind === "work-live" || row.kind === "work-toggle" || - row.kind === "thinking" + row.kind === "thinking" || + row.kind === "worktree-setup" ? "pb-2" : "pb-4", (row.kind === "message" && row.message.role === "assistant") || @@ -1421,10 +1444,33 @@ const TimelineRowContent = memo(function TimelineRowContent({ row }: { row: Time {row.kind === "proposed-plan" ? : null} {row.kind === "working" ? : null} {row.kind === "thinking" ? : null} + {row.kind === "worktree-setup" ? : null} ); }); +function WorktreeSetupTimelineRow({ + row, +}: { + row: Extract; +}) { + const ctx = use(TimelineRowCtx); + const terminalId = row.snapshot.setupScript?.terminalId ?? null; + const openTerminal = ctx.onOpenWorktreeSetupTerminal; + const onOpenTerminal = useMemo( + () => (openTerminal && terminalId ? () => openTerminal(terminalId) : null), + [openTerminal, terminalId], + ); + return ( + + ); +} + function ContextCompactionTimelineRow({ row, }: { diff --git a/apps/web/src/components/chat/WorktreeSetupCard.tsx b/apps/web/src/components/chat/WorktreeSetupCard.tsx new file mode 100644 index 0000000000000000000000000000000000000000..8b838fb81cc95eb607f5cfc05c0d26b6c71e2b72 --- /dev/null +++ b/apps/web/src/components/chat/WorktreeSetupCard.tsx @@ -0,0 +1,296 @@ +import { + worktreeSetupStageLabel, + type WorktreeSetupSnapshot, + type WorktreeSetupStage, +} from "@t3tools/contracts"; +import { formatDuration } from "@t3tools/shared/orchestrationTiming"; +import { + CheckIcon, + ChevronDownIcon, + ChevronRightIcon, + CircleAlertIcon, + CircleIcon, + GitBranchIcon, + LaptopIcon, + MinusIcon, + TerminalIcon, + XIcon, +} from "lucide-react"; +import { useEffect, useState } from "react"; + +import { Button } from "~/components/ui/button"; +import { Spinner } from "~/components/ui/spinner"; +import { cn } from "~/lib/utils"; + +interface WorktreeSetupCardProps { + snapshot: WorktreeSetupSnapshot; + /** Interrupts the server-side bootstrap. Hidden once the setup has settled. */ + onCancel: (() => void) | null; + /** Restarts the same message in the project checkout instead of a worktree. */ + onWorkLocally: (() => void) | null; + /** Reveals the setup script terminal tab. Null when no script ran. */ + onOpenTerminal: (() => void) | null; +} + +function stageElapsedMs(stage: WorktreeSetupStage, nowMs: number): number | null { + if (!stage.startedAt) return null; + const start = Date.parse(stage.startedAt); + const end = stage.endedAt ? Date.parse(stage.endedAt) : nowMs; + if (!Number.isFinite(start) || !Number.isFinite(end)) return null; + return Math.max(0, end - start); +} + +/** + * Ticks once a second while any stage runs so elapsed labels stay live + * without pushing a React commit through the timeline for every second. + */ +function useNowWhile(active: boolean): number { + const [nowMs, setNowMs] = useState(() => Date.now()); + useEffect(() => { + if (!active) return; + const id = setInterval(() => setNowMs(Date.now()), 1_000); + return () => clearInterval(id); + }, [active]); + return nowMs; +} + +function StageIcon({ status }: { status: WorktreeSetupStage["status"] }) { + const className = "size-4 shrink-0 stroke-[1.8]"; + switch (status) { + case "done": + return ; + case "running": + return ; + case "failed": + return ; + case "warning": + return ; + case "skipped": + return ; + case "pending": + return ; + } +} + +function stageRowClassName(status: WorktreeSetupStage["status"]): string { + switch (status) { + case "running": + return "text-foreground"; + case "failed": + return "text-destructive-foreground"; + case "warning": + return "text-warning-foreground"; + case "pending": + case "skipped": + return "text-secondary-label opacity-50"; + case "done": + return "text-secondary-label"; + } +} + +function StageRow({ + stage, + nowMs, + scriptName, +}: { + stage: WorktreeSetupStage; + nowMs: number; + scriptName: string | null; +}) { + const elapsed = stageElapsedMs(stage, nowMs); + const label = + stage.id === "setup-script" && scriptName ? scriptName : worktreeSetupStageLabel(stage.id); + const showBar = stage.id === "checkout" && stage.status === "running" && stage.percent !== null; + const trailing = + stage.status === "pending" + ? null + : stage.status === "skipped" + ? (stage.detail ?? "skipped") + : stage.detail; + + return ( +
+ + + + {label} + + {showBar ? ( + <> + + + + {stage.percent}% + + ) : null} + {!showBar && trailing ? {trailing} : null} + {elapsed !== null && stage.status !== "skipped" && stage.status !== "pending" ? ( + {formatDuration(elapsed)} + ) : null} + +
+ ); +} + +function OutputTail({ lines, failed }: { lines: ReadonlyArray; failed: boolean }) { + if (lines.length === 0) return null; + return ( + 
+      {lines.join("\n")}
+
+ ); +} + +function headerLabel(snapshot: WorktreeSetupSnapshot): string { + switch (snapshot.phase) { + case "running": + return "Creating worktree"; + case "done": + return snapshot.stages.some((stage) => stage.status === "failed") + ? "Worktree ready, setup script failed" + : "Worktree ready"; + case "failed": + return "Worktree setup failed"; + case "cancelled": + return "Worktree setup cancelled"; + } +} + +export function WorktreeSetupCard({ + snapshot, + onCancel, + onWorkLocally, + onOpenTerminal, +}: WorktreeSetupCardProps) { + const running = snapshot.phase === "running"; + const nowMs = useNowWhile(running); + const [detailsOpen, setDetailsOpen] = useState(false); + const totalElapsed = (() => { + const start = Date.parse(snapshot.startedAt); + const end = snapshot.endedAt ? Date.parse(snapshot.endedAt) : nowMs; + return Number.isFinite(start) && Number.isFinite(end) ? Math.max(0, end - start) : null; + })(); + const setupStage = snapshot.stages.find((stage) => stage.id === "setup-script"); + const failed = snapshot.phase === "failed"; + const finishedWithFailedStage = + snapshot.phase === "done" && snapshot.stages.some((stage) => stage.status === "failed"); + const headerClassName = failed + ? "text-destructive-foreground" + : finishedWithFailedStage + ? "text-warning-foreground" + : snapshot.phase === "cancelled" + ? "text-muted-foreground" + : "text-secondary-label"; + + return ( +
+
+ + + + {headerLabel(snapshot)} + {totalElapsed !== null ? ( + + {formatDuration(totalElapsed)} + + ) : null} +
+ + {snapshot.stages.map((stage) => ( +
+ + {stage.id === "setup-script" && + (stage.status === "running" || stage.status === "failed") ? ( + + ) : null} +
+ ))} + + {failed && snapshot.error ? ( +

{snapshot.error}

+ ) : null} + + {detailsOpen ? ( +
+ {snapshot.branch ? ( + <> +
Branch
+
{snapshot.branch}
+ + ) : null} + {snapshot.baseRef ? ( + <> +
Base
+
{snapshot.baseRef}
+ + ) : null} + {snapshot.worktreePath ? ( + <> +
Path
+
{snapshot.worktreePath}
+ + ) : null} + {snapshot.setupScript ? ( + <> +
Setup
+
{snapshot.setupScript.command}
+ + ) : null} +
+ ) : null} + +
+ + + {onOpenTerminal && setupStage && setupStage.status !== "pending" ? ( + + ) : null} + {onWorkLocally ? ( + + ) : null} + {onCancel && running ? ( + + ) : null} +
+
+ ); +} diff --git a/packages/client-runtime/src/rpc/client.ts b/packages/client-runtime/src/rpc/client.ts index cfabaa00c0b7cce762eea1d2f58d3648d63fcfa1..af140ef2fcde3bf4c9c61873986d1dcf7555d525 100644 --- a/packages/client-runtime/src/rpc/client.ts +++ b/packages/client-runtime/src/rpc/client.ts @@ -56,6 +56,7 @@ export type EnvironmentSubscriptionRpcTag = | typeof WS_METHODS.pullRequestsSubscribeRefreshes | typeof WS_METHODS.previewAutomationConnect | typeof WS_METHODS.subscribeVcsStatus + | typeof WS_METHODS.subscribeWorktreeSetup | typeof WS_METHODS.terminalAttach; export type EnvironmentStreamCommandRpcTag = diff --git a/packages/client-runtime/src/state/vcs.ts b/packages/client-runtime/src/state/vcs.ts index 6c93e6204dc860681fe386b6c71b81a83c55b3d3..bedcb751216ff7e49c02a225d78462e3bd3ff6b9 100644 --- a/packages/client-runtime/src/state/vcs.ts +++ b/packages/client-runtime/src/state/vcs.ts @@ -15,7 +15,11 @@ import * as Stream from "effect/Stream"; import * as SubscriptionRef from "effect/SubscriptionRef"; import { Atom, AtomRegistry } from "effect/unstable/reactivity"; -import { createEnvironmentRpcCommand, createEnvironmentSubscriptionAtomFamily } from "./runtime.ts"; +import { + createEnvironmentRpcCommand, + createEnvironmentRpcSubscriptionAtomFamily, + createEnvironmentSubscriptionAtomFamily, +} from "./runtime.ts"; import type { EnvironmentRegistry } from "../connection/registry.ts"; import { EnvironmentSupervisor } from "../connection/supervisor.ts"; import { safeErrorLogAttributes } from "../errors/safeLog.ts"; @@ -311,6 +315,18 @@ export function createVcsEnvironmentAtoms( concurrency: vcsCommandConcurrency, onSettled: invalidateRefs, }), + // Live stages of a bootstrap worktree setup. Null until the server begins + // tracking, then a snapshot per change, then null again after the setup + // is dropped. Short TTL so a closed thread releases its subscription. + worktreeSetup: createEnvironmentRpcSubscriptionAtomFamily(runtime, { + label: "environment-data:vcs:worktree-setup", + tag: WS_METHODS.subscribeWorktreeSetup, + idleTtlMs: VCS_STATUS_IDLE_TTL_MS, + }), + cancelWorktreeSetup: createEnvironmentRpcCommand(runtime, { + label: "environment-data:vcs:cancel-worktree-setup", + tag: WS_METHODS.worktreeSetupCancel, + }), removeWorktree: createEnvironmentRpcCommand(runtime, { label: "environment-data:vcs:remove-worktree", tag: WS_METHODS.vcsRemoveWorktree, diff --git a/packages/contracts/src/index.ts b/packages/contracts/src/index.ts index 77d72144697a41a2fa8ce825966306415970bd78..007120dcba05f45161c596ed6cb796da3e17f74c 100644 --- a/packages/contracts/src/index.ts +++ b/packages/contracts/src/index.ts @@ -42,3 +42,4 @@ export * from "./previewAutomation.ts"; export * from "./resourceTelemetry.ts"; export * from "./usage.ts"; export * from "./rpc.ts"; +export * from "./worktreeSetup.ts"; diff --git a/packages/contracts/src/rpc.ts b/packages/contracts/src/rpc.ts index 7ec895137eb455e3a4ca089178e81db0f286f69d..d792a31885c1a3ddc07277b02d54b1dc4b4ddf7f 100644 --- a/packages/contracts/src/rpc.ts +++ b/packages/contracts/src/rpc.ts @@ -46,6 +46,12 @@ import { AttachmentDeleteInput, AttachmentUploadSigningKeyError, } from "./assets.ts"; +import { + WorktreeSetupCancelInput, + WorktreeSetupCancelResult, + WorktreeSetupStreamEvent, + WorktreeSetupSubscribeInput, +} from "./worktreeSetup.ts"; import { GitActionProgressEvent, VcsSwitchRefInput, @@ -404,6 +410,8 @@ export const WS_METHODS = { // Streaming subscriptions subscribeVcsStatus: "subscribeVcsStatus", + subscribeWorktreeSetup: "subscribeWorktreeSetup", + worktreeSetupCancel: "worktreeSetup.cancel", subscribeTerminalEvents: "subscribeTerminalEvents", subscribeTerminalMetadata: "subscribeTerminalMetadata", subscribePreviewEvents: "subscribePreviewEvents", @@ -940,6 +948,19 @@ const WsVcsRefreshStatusRpc = Rpc.make(WS_METHODS.vcsRefreshStatus, { error: Schema.Union([GitManagerServiceError, EnvironmentAuthorizationError]), }); +const WsSubscribeWorktreeSetupRpc = Rpc.make(WS_METHODS.subscribeWorktreeSetup, { + payload: WorktreeSetupSubscribeInput, + success: WorktreeSetupStreamEvent, + error: EnvironmentAuthorizationError, + stream: true, +}); + +const WsWorktreeSetupCancelRpc = Rpc.make(WS_METHODS.worktreeSetupCancel, { + payload: WorktreeSetupCancelInput, + success: WorktreeSetupCancelResult, + error: EnvironmentAuthorizationError, +}); + const WsGitRunStackedActionRpc = Rpc.make(WS_METHODS.gitRunStackedAction, { payload: GitRunStackedActionInput, success: GitActionProgressEvent, @@ -1372,6 +1393,8 @@ export const WsRpcGroup = RpcGroup.make( WsAttachmentsDeleteRpc, WsProviderUploadFeedbackRpc, WsSubscribeVcsStatusRpc, + WsSubscribeWorktreeSetupRpc, + WsWorktreeSetupCancelRpc, WsVcsPullRpc, WsVcsRefreshStatusRpc, WsGitRunStackedActionRpc, diff --git a/packages/contracts/src/worktreeSetup.ts b/packages/contracts/src/worktreeSetup.ts new file mode 100644 index 0000000000000000000000000000000000000000..9f3a9f33be237547465b60be91e79e9516feff04 --- /dev/null +++ b/packages/contracts/src/worktreeSetup.ts @@ -0,0 +1,114 @@ +import * as Schema from "effect/Schema"; + +import { IsoDateTime, NonNegativeInt, ThreadId, TrimmedNonEmptyString } from "./baseSchemas.ts"; + +/** + * Live progress for a thread whose first turn is creating a worktree. The + * server keeps this in memory only; a client that reconnects mid-setup gets a + * fresh snapshot, and a finished setup is dropped once its turn starts. + */ +/** Producers clamp free text to these before publishing so encoding never fails. */ +export const WORKTREE_SETUP_DETAIL_MAX_LENGTH = 200; +export const WORKTREE_SETUP_TAIL_LINE_MAX_LENGTH = 400; +export const WORKTREE_SETUP_ERROR_MAX_LENGTH = 1000; + +export const WorktreeSetupStageId = Schema.Literals([ + "fetch", + "checkout", + "submodules", + "setup-script", + "agent", +]); +export type WorktreeSetupStageId = typeof WorktreeSetupStageId.Type; + +export const WorktreeSetupStageStatus = Schema.Literals([ + "pending", + "running", + "done", + "skipped", + "warning", + "failed", +]); +export type WorktreeSetupStageStatus = typeof WorktreeSetupStageStatus.Type; + +export const WorktreeSetupStage = Schema.Struct({ + id: WorktreeSetupStageId, + status: WorktreeSetupStageStatus, + startedAt: Schema.NullOr(IsoDateTime), + endedAt: Schema.NullOr(IsoDateTime), + /** Only the checkout stage reports a real percentage, parsed from git's `Updating files` lines. */ + percent: Schema.NullOr(Schema.Int.check(Schema.isBetween({ minimum: 0, maximum: 100 }))), + /** Short trailing text for the row: a file count, an exit code, a submodule name. */ + detail: Schema.NullOr(Schema.String.check(Schema.isMaxLength(WORKTREE_SETUP_DETAIL_MAX_LENGTH))), + /** Last few output lines from the setup script, ANSI stripped, newest last. */ + tail: Schema.Array(Schema.String.check(Schema.isMaxLength(WORKTREE_SETUP_TAIL_LINE_MAX_LENGTH))), +}); +export type WorktreeSetupStage = typeof WorktreeSetupStage.Type; + +export const WorktreeSetupPhase = Schema.Literals(["running", "done", "failed", "cancelled"]); +export type WorktreeSetupPhase = typeof WorktreeSetupPhase.Type; + +export const WorktreeSetupSnapshot = Schema.Struct({ + threadId: ThreadId, + phase: WorktreeSetupPhase, + startedAt: IsoDateTime, + endedAt: Schema.NullOr(IsoDateTime), + branch: Schema.NullOr(TrimmedNonEmptyString), + baseRef: Schema.NullOr(TrimmedNonEmptyString), + worktreePath: Schema.NullOr(TrimmedNonEmptyString), + /** Display name plus command of the setup script from t3.json, when one runs. */ + setupScript: Schema.NullOr( + Schema.Struct({ + name: TrimmedNonEmptyString, + command: TrimmedNonEmptyString, + terminalId: TrimmedNonEmptyString, + }), + ), + stages: Schema.Array(WorktreeSetupStage), + /** Human readable reason when phase is failed. */ + error: Schema.NullOr(Schema.String.check(Schema.isMaxLength(WORKTREE_SETUP_ERROR_MAX_LENGTH))), + sequence: NonNegativeInt, +}); +export type WorktreeSetupSnapshot = typeof WorktreeSetupSnapshot.Type; + +export const WorktreeSetupSubscribeInput = Schema.Struct({ + threadId: ThreadId, +}); +export type WorktreeSetupSubscribeInput = typeof WorktreeSetupSubscribeInput.Type; + +/** Null means no setup is tracked for that thread. Sent first, then after every change. */ +export const WorktreeSetupStreamEvent = Schema.NullOr(WorktreeSetupSnapshot); +export type WorktreeSetupStreamEvent = typeof WorktreeSetupStreamEvent.Type; + +export const WorktreeSetupCancelInput = Schema.Struct({ + threadId: ThreadId, +}); +export type WorktreeSetupCancelInput = typeof WorktreeSetupCancelInput.Type; + +export const WorktreeSetupCancelResult = Schema.Struct({ + cancelled: Schema.Boolean, +}); +export type WorktreeSetupCancelResult = typeof WorktreeSetupCancelResult.Type; + +export const WORKTREE_SETUP_STAGE_ORDER: ReadonlyArray = [ + "fetch", + "checkout", + "submodules", + "setup-script", + "agent", +]; + +export function worktreeSetupStageLabel(id: WorktreeSetupStageId): string { + switch (id) { + case "fetch": + return "Fetch base branch"; + case "checkout": + return "Check out files"; + case "submodules": + return "Init submodules"; + case "setup-script": + return "Run setup script"; + case "agent": + return "Start agent"; + } +}