From 73b206f4bf99e7e98ae144a9d9edbf7cb2700fc1 Mon Sep 17 00:00:00 2001 From: Theo Browne Date: Sun, 13 Sep 2026 21:28:59 -0700 Subject: [PATCH] feat(web): show each worktree setup step and let users cancel it (#11372) Starting a thread in a new worktree showed one static "Setting up worktree" line while git fetched, checked out files, and the setup script ran. Users could not tell which step was slow, see setup script output without hunting for the terminal, or stop a setup that was going wrong. The timeline now shows a card with each step and its elapsed time. Check out files has a percent bar fed by git's own progress output. The setup script step shows the last lines of its terminal inline and links to the full terminal. Cancel stops the bootstrap and removes the half built worktree. Work locally cancels, switches the draft to the project checkout, and resends. The server keeps an in-memory per-thread snapshot of the bootstrap stages and streams it over a new subscribeWorktreeSetup RPC. The bootstrap runs as a child fiber so worktreeSetup.cancel can interrupt it, and the turn handoff is uninterruptible. The setup script's exit code comes from a per-run sentinel echoed after the command in the setup PTY. Created with Claude Fable 5.1 in Claude Code. --- apps/server/src/auth/RpcAuthorization.ts | 2 + apps/server/src/git/GitWorkflowService.ts | 5 +- .../project/ProjectSetupScriptRunner.test.ts | 218 ++++++++++- .../src/project/ProjectSetupScriptRunner.ts | 217 ++++++++++- .../src/project/WorktreeSetupTracker.test.ts | 226 +++++++++++ .../src/project/WorktreeSetupTracker.ts | 364 ++++++++++++++++++ apps/server/src/server.test.ts | 35 +- apps/server/src/server.ts | 2 + apps/server/src/vcs/GitVcsDriver.ts | 31 ++ apps/server/src/vcs/GitVcsDriverCore.test.ts | 67 +++- apps/server/src/vcs/GitVcsDriverCore.ts | 101 ++++- apps/server/src/ws.ts | 356 +++++++++++++++-- .../web/src/components/ChatView.logic.test.ts | 24 ++ apps/web/src/components/ChatView.logic.ts | 5 + apps/web/src/components/ChatView.tsx | 156 ++++++++ .../chat/MessagesTimeline.logic.test.ts | 78 ++++ .../components/chat/MessagesTimeline.logic.ts | 43 ++- .../src/components/chat/MessagesTimeline.tsx | 48 ++- .../src/components/chat/WorktreeSetupCard.tsx | 296 ++++++++++++++ packages/client-runtime/src/rpc/client.ts | 1 + packages/client-runtime/src/state/vcs.ts | 18 +- packages/contracts/src/index.ts | 1 + packages/contracts/src/rpc.ts | 23 ++ packages/contracts/src/worktreeSetup.ts | 114 ++++++ 24 files changed, 2352 insertions(+), 79 deletions(-) create mode 100644 apps/server/src/project/WorktreeSetupTracker.test.ts create mode 100644 apps/server/src/project/WorktreeSetupTracker.ts create mode 100644 apps/web/src/components/chat/WorktreeSetupCard.tsx create mode 100644 packages/contracts/src/worktreeSetup.ts 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"; + } +} -- 2.54.0