diff --git a/server/src/index.ts b/server/src/index.ts index 3c0585bedcd..fac08389db2 100644 --- a/server/src/index.ts +++ b/server/src/index.ts @@ -1185,7 +1185,8 @@ async function startServerWithDatabaseTeardown( }; const executionControlSweepsInFlight = new Set(); const executionControlSweeps = [ - ["computers", () => computerService(db).reconcile()], + ["computer_renewal", () => computerService(db).reconcile("renewal")], + ["computer_cleanup", () => computerService(db).reconcile("cleanup")], ["finalization", () => reconcileAbandonedExecutionControl(db)], ["replacement", () => heartbeat ? reconcileSafeNativeReplacements(db, new Date(), { verifyStoppedSession: run => verifyStoppedNativeSessionForReplacement(db, run) }) : undefined], ["reconciliation_delivery", () => heartbeat ? deliverReconciledExecutions(db, heartbeat.wakeup) : undefined], diff --git a/server/src/modules/computers/adapters/boat.test.ts b/server/src/modules/computers/adapters/boat.test.ts index 0d8a62d3dc7..cdd2ccda045 100644 --- a/server/src/modules/computers/adapters/boat.test.ts +++ b/server/src/modules/computers/adapters/boat.test.ts @@ -790,6 +790,56 @@ process.stdout.write(result.stdout||"");process.stderr.write(result.stderr||""); expect(f.call({ ...input, branch: "another-branch" })).toEqual({ error: "conflict" }); expect(git("-C", result.remoteCwd, "status", "--short")).toContain("preserved"); }); + it.each(["release", "feature/release", "origin/release", "refs/remotes/origin/release"])("refreshes the authoritative remote worktree base %s without rewriting existing work", baseRef => { + const f = fixture(), source = join(f.temp, "source"); mkdirSync(source); + const git = (cwd: string, ...args: string[]) => { + const result = spawnSync("git", args, { cwd, encoding: "utf8" }); + if (result.status !== 0) throw new Error(result.stderr); + return result.stdout.trim(); + }; + const commit = (message: string) => git(source, "-c", "user.name=Test", "-c", "user.email=test@example.com", "commit", "--allow-empty", "-m", message); + git(source, "init", "-b", "main"); commit("default"); + const branch = baseRef === "feature/release" ? baseRef : "release"; + git(source, "checkout", "-b", branch); commit("release one"); + const original = git(source, "rev-parse", "HEAD"); git(source, "tag", "baseline"); git(source, "checkout", "main"); + const repositoryUrl = "https://fixture.invalid/source.git"; + // Real Git clone/fetch over a local transport; the URL rewrite exists only + // in each child invocation, never in repository or global configuration. + const gitAuth = { configArgs: ["-c", `url.file://${source}.insteadOf=${repositoryUrl}`, "-c", "protocol.file.allow=always"], env: {} }; + const request = { action: "workspace", mode: "worktree", repositoryUrl, gitAuth, baseRef, taskId: "one", branch: "paperclip/one" }; + const first = f.call(request); + expect(first.remoteCwd).toBe(join(f.root, "tasks", "one")); + expect(git(first.remoteCwd, "rev-parse", "HEAD")).toBe(original); + const checkout = join(f.root, "checkout"); + expect(spawnSync("git", ["show-ref", "--verify", `refs/heads/${branch}`], { cwd: checkout }).status).not.toBe(0); + writeFileSync(join(first.remoteCwd, "uncommitted"), "keep"); + git(source, "checkout", branch); commit("release two"); + const updated = git(source, "rev-parse", "HEAD"); git(source, "checkout", "main"); + // A divergent local branch must not override the current remote base. + git(checkout, "branch", branch, "HEAD"); + const deniedAuth = { ...gitAuth, configArgs: [...gitAuth.configArgs, "-c", "protocol.file.allow=never"] }; + expect(f.call({ ...request, gitAuth: deniedAuth, taskId: "denied", branch: "paperclip/denied" })).toEqual({ error: "command_failed" }); + expect(existsSync(join(f.root, "tasks", "denied"))).toBe(false); + const second = f.call({ ...request, taskId: "two", branch: "paperclip/two" }); + expect(second.remoteCwd).toBe(join(f.root, "tasks", "two")); + expect(git(second.remoteCwd, "rev-parse", "HEAD")).toBe(updated); + expect(git(checkout, "rev-parse", branch)).not.toBe(updated); + const existing = f.call({ ...request, gitAuth: deniedAuth, taskId: "existing", branch, existingBranch: branch }); + expect(git(existing.remoteCwd, "rev-parse", "HEAD")).toBe(git(checkout, "rev-parse", branch)); + const tagged = f.call({ ...request, gitAuth: deniedAuth, taskId: "tagged", branch: "paperclip/tagged", baseRef: "baseline" }); + expect(git(tagged.remoteCwd, "rev-parse", "HEAD")).toBe(original); + const ancestor = f.call({ ...request, gitAuth: deniedAuth, taskId: "ancestor", branch: "paperclip/ancestor", baseRef: `origin/${branch}~1` }); + expect(git(ancestor.remoteCwd, "rev-parse", "HEAD")).toBe(original); + const defaultHead = f.call({ ...request, gitAuth: deniedAuth, taskId: "default-head", branch: "paperclip/default-head", baseRef: "origin/HEAD" }); + expect(git(defaultHead.remoteCwd, "rev-parse", "HEAD")).toBe(git(checkout, "rev-parse", "HEAD")); + expect(f.call(request).remoteCwd).toBe(first.remoteCwd); + expect(git(first.remoteCwd, "rev-parse", "HEAD")).toBe(original); + expect(readFileSync(join(first.remoteCwd, "uncommitted"), "utf8")).toBe("keep"); + expect(readFileSync(join(checkout, ".git/config"), "utf8")).not.toContain("insteadOf"); + expect(f.call({ ...request, taskId: "missing", branch: "paperclip/missing", baseRef: "missing" })).toEqual({ error: "command_failed" }); + expect(existsSync(join(f.root, "tasks", "missing"))).toBe(false); + expect(spawnSync("git", ["show-ref", "--verify", "refs/heads/paperclip/missing"], { cwd: checkout }).status).not.toBe(0); + }); it.each([false, true])("attaches only the exact existing worktree branch (present: %s)", present => { const f = fixture(), checkout = join(f.root, "checkout"); mkdirSync(checkout, { recursive: true }); diff --git a/server/src/modules/computers/adapters/remote-program.ts b/server/src/modules/computers/adapters/remote-program.ts index 9f58ed285da..520a7043636 100644 --- a/server/src/modules/computers/adapters/remote-program.ts +++ b/server/src/modules/computers/adapters/remote-program.ts @@ -75,19 +75,46 @@ if act=='workspace': if existing_branch and (p.get('mode')!='worktree' or existing_branch!=branch):fail('invalid') if base_ref and (base_ref.startswith('-') or '\n' in base_ref):fail('invalid') checkout=os.path.join(root,'checkout') + # Only network Git children receive operation-scoped credentials. Never + # persist helper configuration in the checkout or the shared agent home. + auth=p.get('gitAuth') or {};git_env=dict(os.environ) + git_env.update({'GIT_TERMINAL_PROMPT':'0','GIT_CONFIG_GLOBAL':'/dev/null','GIT_CONFIG_NOSYSTEM':'1'}) + git_env.update(auth.get('env',{})) if repo: if not (repo.startswith('https://') or repo.startswith('ssh://') or repo.startswith('git@')) or '\n' in repo:fail('invalid') if not os.path.exists(checkout): - # Resolve credentials on the controller for this operation. Only the clone - # child receives the helper environment; no credential file or Git config persists. - auth=p.get('gitAuth') or {};git_env=dict(os.environ) - git_env.update({'GIT_TERMINAL_PROMPT':'0','GIT_CONFIG_GLOBAL':'/dev/null','GIT_CONFIG_NOSYSTEM':'1'}) - git_env.update(auth.get('env',{})) run(['git']+auth.get('configArgs',[])+['clone','--',repo,checkout],env=git_env) elif run(['git','remote','get-url','origin'],checkout)!=repo:fail('conflict') else: os.makedirs(checkout,exist_ok=True) if not os.path.isdir(os.path.join(checkout,'.git')):run(['git','init',checkout]) + def commit_ref(value): + result=subprocess.run(['git','rev-parse','--verify','--end-of-options',value+'^{commit}'],cwd=checkout,capture_output=True,text=True) + return result.stdout.strip() if result.returncode==0 else None + def worktree_base(value): + if not value or value=='HEAD':return 'HEAD' + remotes=run(['git','remote'],checkout).splitlines() + remote=None;remote_branch=None + candidate=value[len('refs/remotes/'):] if value.startswith('refs/remotes/') else value + parts=candidate.split('/',1) + if len(parts)==2 and parts[0] in remotes: + remote,remote_branch=parts + else: + local=value[len('refs/heads/'):] if value.startswith('refs/heads/') else value + local_branch=subprocess.run(['git','show-ref','--verify','--quiet','refs/heads/'+local],cwd=checkout,capture_output=True).returncode==0 + # Tags and commit expressions retain their exact meaning. A local branch + # with origin uses the remote counterpart rather than unpushed divergence. + if not local_branch: + resolved=commit_ref(value) + if resolved:return resolved + if 'origin' in remotes:remote,remote_branch='origin',local + else:return commit_ref(value) or fail('command_failed') + if remote_branch=='HEAD':return commit_ref(value) or fail('command_failed') + if subprocess.run(['git','check-ref-format','refs/heads/'+remote_branch],cwd=checkout,capture_output=True).returncode: + return commit_ref(value) or fail('command_failed') + target_ref='refs/remotes/'+remote+'/'+remote_branch + run(['git']+auth.get('configArgs',[])+['fetch','--no-tags','--',remote,'+refs/heads/'+remote_branch+':'+target_ref],checkout,git_env) + return commit_ref(target_ref) or fail('command_failed') if branch: if branch.startswith('-') or '\n' in branch:fail('invalid') run(['git','check-ref-format','--branch',branch]) @@ -98,7 +125,7 @@ if act=='workspace': if existing_branch:run(['git','show-ref','--verify','refs/heads/'+existing_branch],checkout) if not os.path.exists(target): if existing_branch:run(['git','worktree','add',target,existing_branch],checkout) - else:run(['git','worktree','add','-b',task_branch,target,base_ref or 'HEAD'],checkout) + else:run(['git','worktree','add','-b',task_branch,target,worktree_base(base_ref)],checkout) run(['git','rev-parse','--show-toplevel'],target) if run(['git','symbolic-ref','--quiet','--short','HEAD'],target)!=task_branch:fail('conflict') else: diff --git a/server/src/modules/computers/application/service.test.ts b/server/src/modules/computers/application/service.test.ts index ab7d98e1ced..f2050785e87 100644 --- a/server/src/modules/computers/application/service.test.ts +++ b/server/src/modules/computers/application/service.test.ts @@ -529,6 +529,191 @@ describe("computer ownership", () => { await Promise.all([first, second]); expect(f.backend.stop).toHaveBeenCalledOnce(); }); + async function reconciliationFleet(count: number) { + const f = fixture(); + await f.attach(); + const template = await f.repository.get(f.scope); + const records = Array.from({ length: count }, (_, index): ComputerRecord => ({ + ...structuredClone(template), id: `computer-${index}`, environmentId: `environment-${index}`, + ledger: { ...structuredClone(template.ledger), owners: [{ + id: `viewer-${index}`, generation: 1, kind: "viewer", phase: "active", userId: "alice", port: 0, + deadline: "2026-10-10T13:00:00Z", absoluteDeadline: "2026-10-10T13:00:00Z", process: null, + }] }, + })); + const find = (scope: { companyId: string; environmentId: string }) => { + const record = records.find(record => record.companyId === scope.companyId && record.environmentId === scope.environmentId); + if (!record) throw new ComputerError("not_found", "missing"); + return record; + }; + const repository: ComputerRepository = { + ...f.repository, + get: async scope => structuredClone(find(scope)), + update: async (scope, change) => change(find(scope)), + all: vi.fn(async () => structuredClone(records)), + }; + return { backend: f.backend, repository, service: createComputerService(repository, f.backend) }; + } + it("renews a healthy computer before its TTL despite four unavailable providers", async () => { + vi.useFakeTimers(); + vi.setSystemTime(new Date("2026-10-10T12:00:00Z")); + let pending: Promise | undefined; + try { + const f = await reconciliationFleet(5); + const renewedAt: number[] = []; + const start = Date.now(); + vi.mocked(f.backend.renew).mockImplementation(async record => { + if (record.id === "computer-4") { renewedAt.push(Date.now() - start); return; } + await new Promise(resolve => setTimeout(resolve, 90_000)); + throw new Error("provider unavailable"); + }); + pending = f.service.reconcile().catch(error => error); + await vi.advanceTimersByTimeAsync(300_000); + expect(renewedAt).toEqual([90_000]); + const failure = await pending; + expect(failure).toBeInstanceOf(AggregateError); + expect((failure as AggregateError).errors).toHaveLength(4); + expect(f.backend.stop).not.toHaveBeenCalled(); + vi.mocked(f.backend.renew).mockResolvedValue(undefined).mockClear(); + await f.service.reconcile(); + expect(f.backend.renew).toHaveBeenCalledTimes(5); + } finally { + await vi.runAllTimersAsync(); + await pending; + vi.useRealTimers(); + } + }); + it("renews again while a larger unrelated stop backlog is still running", async () => { + vi.useFakeTimers(); + vi.setSystemTime(new Date("2026-10-10T12:00:00Z")); + let pending: Promise | undefined; + try { + const f = await reconciliationFleet(21); + for (let index = 0; index < 20; index++) await f.repository.update({ companyId: "company", environmentId: `environment-${index}` }, record => { + record.ledger.owners = []; + record.ledger.action = { kind: "stop", id: `action-${index}`, providerStopId: `stop-${index}` }; + }); + let activeStops = 0, maximumStops = 0; + vi.mocked(f.backend.stopStatus).mockImplementation(async (_record, id) => { + maximumStops = Math.max(maximumStops, ++activeStops); + await new Promise(resolve => setTimeout(resolve, 90_000)); + activeStops--; + return { id, status: "pending" }; + }); + const renewedAt: number[] = []; + const start = Date.now(); + vi.mocked(f.backend.renew).mockImplementation(async () => { renewedAt.push(Date.now() - start); }); + pending = f.service.reconcile("cleanup"); + const initialRenewal = f.service.reconcile("renewal"); + await vi.advanceTimersByTimeAsync(0); + expect(f.backend.renew).toHaveBeenCalledOnce(); + expect(f.backend.renew).toHaveBeenCalledWith(expect.objectContaining({ id: "computer-20" })); + await vi.advanceTimersByTimeAsync(240_000); + const nextRenewal = f.service.reconcile("renewal"); + await vi.advanceTimersByTimeAsync(0); + expect(renewedAt).toEqual([0, 240_000]); + await Promise.all([initialRenewal, nextRenewal]); + expect(activeStops).toBe(4); + expect(f.backend.stopStatus).toHaveBeenCalledTimes(12); + await vi.runAllTimersAsync(); await pending; + expect(maximumStops).toBe(4); + expect(f.backend.stopStatus).toHaveBeenCalledTimes(20); + expect((await f.repository.get({ companyId: "company", environmentId: "environment-0" })).ledger.action) + .toMatchObject({ id: "action-0", providerStopId: "stop-0" }); + } finally { + await vi.runAllTimersAsync(); await pending; + vi.useRealTimers(); + } + }); + it("bounds fleet reconciliation and shares an overlapping sweep without duplicate renewals", async () => { + vi.useFakeTimers(); + vi.setSystemTime(new Date("2026-10-10T12:00:00Z")); + let first: Promise | undefined, second: Promise | undefined; + try { + const f = await reconciliationFleet(9); + let active = 0, maximum = 0; + const calls: string[] = []; + vi.mocked(f.backend.renew).mockImplementation(async record => { + calls.push(record.id); maximum = Math.max(maximum, ++active); + await new Promise(resolve => setTimeout(resolve, 1000)); + active--; + }); + first = f.service.reconcile(); second = f.service.reconcile(); + await vi.advanceTimersByTimeAsync(0); + expect(active).toBe(4); + await vi.runAllTimersAsync(); + await Promise.all([first, second]); + expect(maximum).toBe(4); + expect(calls).toHaveLength(9); + expect(new Set(calls).size).toBe(9); + expect(f.repository.all).toHaveBeenCalledTimes(2); + const later = f.service.reconcile(); + await vi.runAllTimersAsync(); await later; + expect(f.repository.all).toHaveBeenCalledTimes(4); + expect(calls).toHaveLength(18); + } finally { + await vi.runAllTimersAsync(); + await Promise.all([first, second]); + vi.useRealTimers(); + } + }); + it("shares exact in-flight work when a record changes reconciliation lanes", async () => { + vi.useFakeTimers(); + vi.setSystemTime(new Date("2026-10-10T12:00:00Z")); + let release!: () => void; + const pending: Promise[] = []; + try { + const f = await reconciliationFleet(1); + vi.mocked(f.backend.renew).mockImplementation(() => new Promise(resolve => { release = resolve; })); + pending.push(f.service.reconcile("renewal")); + await vi.advanceTimersByTimeAsync(0); + expect(f.backend.renew).toHaveBeenCalledOnce(); + await f.repository.update({ companyId: "company", environmentId: "environment-0" }, record => { + record.ledger.owners = []; + record.ledger.action = { kind: "stop", id: "action-0", providerStopId: "stop-0" }; + }); + pending.push(f.service.reconcile("cleanup")); + await vi.advanceTimersByTimeAsync(0); + expect(f.backend.stopStatus).not.toHaveBeenCalled(); + release(); await Promise.all(pending); + await f.service.reconcile("cleanup"); + expect(f.backend.renew).toHaveBeenCalledOnce(); + expect(f.backend.stopStatus).toHaveBeenCalledOnce(); + } finally { + release?.(); await Promise.allSettled(pending); + vi.useRealTimers(); + } + }); + it("bounds independently scheduled renewal and cleanup to eight total workers", async () => { + vi.useFakeTimers(); + vi.setSystemTime(new Date("2026-10-10T12:00:00Z")); + const pending: Promise[] = []; + try { + const f = await reconciliationFleet(12); + for (let index = 6; index < 12; index++) await f.repository.update({ companyId: "company", environmentId: `environment-${index}` }, record => { + record.ledger.owners = []; + record.ledger.action = { kind: "stop", id: `action-${index}`, providerStopId: `stop-${index}` }; + }); + let active = 0, maximum = 0; + const delay = async () => { + maximum = Math.max(maximum, ++active); + await new Promise(resolve => setTimeout(resolve, 1000)); active--; + }; + vi.mocked(f.backend.renew).mockImplementation(delay); + vi.mocked(f.backend.stopStatus).mockImplementation(async (_record, id) => { + await delay(); return { id, status: "pending" }; + }); + pending.push(f.service.reconcile("renewal"), f.service.reconcile("cleanup"), f.service.reconcile()); + await vi.advanceTimersByTimeAsync(0); + expect(active).toBe(8); + await vi.runAllTimersAsync(); await Promise.all(pending); + expect(maximum).toBe(8); + expect(f.backend.renew).toHaveBeenCalledTimes(6); + expect(f.backend.stopStatus).toHaveBeenCalledTimes(6); + } finally { + await vi.runAllTimersAsync(); await Promise.allSettled(pending); + vi.useRealTimers(); + } + }); it.each(["retire", "warm"] as const)("waits for terminal predecessor %s before admitting its follow-up", async (transition) => { const f = fixture(); await f.attach(); diff --git a/server/src/modules/computers/application/service.ts b/server/src/modules/computers/application/service.ts index 4fa5fccdfa2..f9196ca2495 100644 --- a/server/src/modules/computers/application/service.ts +++ b/server/src/modules/computers/application/service.ts @@ -28,6 +28,9 @@ class ComputerSessionBusyError extends ComputerError { } const VIEWER_PREPARATION_BUDGET_MS = 120_000; +// Each independently scheduled lane has four workers. Slow cleanup cannot hold +// the next renewal pass; an arbitrary all-live fleet still has finite capacity. +const RECONCILIATION_CONCURRENCY = 4; const FILE_OPERATION_BUDGET_MS = 120_000; const FILE_OPERATION_SETTLEMENT_MS = 5000; type Scope = { companyId: string; environmentId: string }; @@ -1093,19 +1096,41 @@ finally: current.ledger.action.providerStopId = stop.id; }); } - async function reconcile() { - const failures: unknown[] = []; - for (const record of await repository.all()) - try { - await reconcileRecord(record); - } catch (error) { - failures.push(error); - } - if (failures.length) - throw new AggregateError( - failures, - "Computer reconciliation requires retry", - ); + type ReconciliationLane = "renewal" | "cleanup"; + const reconciliationSweeps = new Map>(); + async function reconcileLane(lane: ReconciliationLane) { + const existing = reconciliationSweeps.get(lane); + if (existing) return existing; + const pending = (async () => { + const observedAt = now(); + const records = (await repository.all()).filter(record => { + const renewalCandidate = record.ledger.status === "attached" && !record.ledger.action && + liveOwners(record.ledger).some(owner => owner.phase !== "retiring" && !expired(owner, observedAt)); + return renewalCandidate === (lane === "renewal"); + }); + // Lane selection is scheduling, not authority: reconcileRecord reloads + // each ledger and shares in-flight work even if it has changed lanes. + const failures: unknown[] = []; + let next = 0; + await Promise.all(Array.from({ length: Math.min(RECONCILIATION_CONCURRENCY, records.length) }, async () => { + while (next < records.length) { + const record = records[next++]; + try { await reconcileRecord(record); } + catch (error) { failures.push(error); } + } + })); + if (failures.length) throw new AggregateError(failures, "Computer reconciliation requires retry"); + })(); + reconciliationSweeps.set(lane, pending); + try { await pending; } + finally { if (reconciliationSweeps.get(lane) === pending) reconciliationSweeps.delete(lane); } + } + async function reconcile(lane?: ReconciliationLane) { + if (lane) return reconcileLane(lane); + const results = await Promise.allSettled([reconcileLane("renewal"), reconcileLane("cleanup")]); + const failures = results.flatMap(result => result.status === "fulfilled" ? [] + : result.reason instanceof AggregateError ? result.reason.errors : [result.reason]); + if (failures.length) throw new AggregateError(failures, "Computer reconciliation requires retry"); } return { async list(companyId: string) {