From d2998bb1b401ee58eb63d80d2b81980579833d96 Mon Sep 17 00:00:00 2001 From: Claude Date: Mon, 28 Sep 2026 22:05:38 +0000 Subject: [PATCH 1/3] Show the local bridge's health in the UI MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit The backend already reported whether a project's bridge was ready, restarting or given up, but the UI never showed it. - A "Bridge · Restarting" (warning) or "Bridge · Unavailable" (danger) badge sits next to Repository details in the header. Nothing is shown while the bridge is ready. - Repository details then shows a notice with the bridge's message, restart count, time to the next restart and the log path (with Copy). When the bridge has given up it explains that reopening the repository starts a fresh bridge and offers a Reopen repository button. - workspace-setup is polled every 5 s while the bridge needs attention and every 30 s otherwise, and not at all while the tab is hidden. Each poll runs git status on the repository, so this keeps the cost bounded. Co-Authored-By: Claude Opus 5.5 Claude-Session: https://claude.ai/code/session_01DXCvrGWmmRNDLmy7nYS2QP --- tests/bridge-status.test.tsx | 103 +++++++++++++++++++++++++++++++++++ ui/bridge-status.tsx | 75 +++++++++++++++++++++++++ ui/main.tsx | 11 +++- ui/style.css | 9 +++ 4 files changed, 195 insertions(+), 3 deletions(-) create mode 100644 tests/bridge-status.test.tsx create mode 100644 ui/bridge-status.tsx diff --git a/tests/bridge-status.test.tsx b/tests/bridge-status.test.tsx new file mode 100644 index 0000000..9df90f4 --- /dev/null +++ b/tests/bridge-status.test.tsx @@ -0,0 +1,103 @@ +import { createElement } from 'react'; +import { renderToStaticMarkup } from 'react-dom/server'; +import { afterEach, beforeEach, describe, expect, it, vi } from 'vitest'; +import { App, type State } from '../ui/main.js'; +import { BridgeBadge, BridgeNotice, BRIDGE_POLL_MS, BRIDGE_IDLE_POLL_MS, createBridgePoller, secondsUntil, type BridgeHealth } from '../ui/bridge-status.js'; + +const NOW = Date.parse('2026-09-28T12:00:00.000Z'); +const restarting: BridgeHealth = { state: 'restarting', restarts: 2, nextRestartAt: '2026-09-28T12:00:07.200Z', restartDelayMs: 8000, logPath: '/data/local-bridge/prj.log', message: 'Bridge exited with code 1; restarting.' }; +const unavailable: BridgeHealth = { state: 'unavailable', restarts: 5, logPath: '/data/local-bridge/prj.log', message: 'Bridge kept crashing; giving up.' }; + +describe('bridge status rendering', () => { + it('renders nothing when ready or unknown', () => { + expect(renderToStaticMarkup(createElement(BridgeBadge, { status: 'ready' }))).toBe(''); + expect(renderToStaticMarkup(createElement(BridgeBadge, {}))).toBe(''); + expect(renderToStaticMarkup(createElement(BridgeNotice, { status: 'ready', health: { state: 'ready', restarts: 0 } }))).toBe(''); + }); + it('shows a warning badge and notice with restart detail while restarting', () => { + expect(renderToStaticMarkup(createElement(BridgeBadge, { status: 'restarting' }))).toContain('tone-warning'); + const html = renderToStaticMarkup(createElement(BridgeNotice, { status: 'restarting', health: restarting, now: NOW, onCopy: () => {}, onReopen: () => {} })); + expect(html).toContain('Bridge exited with code 1; restarting.'); + expect(html).toContain('2 restarts'); + expect(html).toContain('next restart in 8s'); + expect(html).toContain('/data/local-bridge/prj.log'); + expect(html).toContain('>Copy'); + expect(html).not.toContain('fresh bridge'); + expect(html).not.toContain('Reopen repository'); + }); + it('shows a danger badge, reopen guidance and action when unavailable', () => { + expect(renderToStaticMarkup(createElement(BridgeBadge, { status: 'unavailable' }))).toContain('tone-failed'); + const html = renderToStaticMarkup(createElement(BridgeNotice, { status: 'unavailable', health: unavailable, now: NOW, onReopen: () => {} })); + expect(html).toContain('Bridge kept crashing; giving up.'); + expect(html).toContain('5 restarts'); + expect(html).not.toContain('next restart'); + expect(html).toContain('Reopening the repository starts a fresh bridge.'); + expect(html).toContain('>Reopen repository'); + }); + it('falls back to a default message and clamps the countdown', () => { + expect(renderToStaticMarkup(createElement(BridgeNotice, { status: 'unavailable' }))).toContain('The local bridge is not running.'); + expect(secondsUntil('2026-09-28T11:59:00.000Z', NOW)).toBe(0); + expect(secondsUntil(undefined, NOW)).toBeUndefined(); + expect(secondsUntil('nope', NOW)).toBeUndefined(); + }); + it('integrates into the project header only when the bridge is unhealthy', () => { + const state: State = { projects: [{ id: 'prj_b', name: 'Bridge project', repoPath: '/repo/b', tasks: [] }], roles: [] }; + const base = { repoPath: '/repo/b', head: 'a'.repeat(40), dirty: false, allowedScope: [], validationCommands: [] }; + const healthy = renderToStaticMarkup(createElement(App, { initialState: state, initialWorkspaceSetup: { ...base, bridgeStatus: 'ready' } })); + expect(healthy).not.toContain('bridge-badge'); + expect(healthy).not.toContain('bridge-notice'); + const down = renderToStaticMarkup(createElement(App, { initialState: state, initialWorkspaceSetup: { ...base, bridgeStatus: 'unavailable', bridgeHealth: unavailable } })); + expect(down).toContain('Bridge · Unavailable'); + expect(down).toContain('Reopening the repository starts a fresh bridge.'); + expect(down).toContain('/data/local-bridge/prj.log'); + }); +}); + +describe('bridge health poller', () => { + beforeEach(() => { vi.useFakeTimers(); }); + afterEach(() => { vi.useRealTimers(); vi.unstubAllGlobals(); }); + + it('polls every 30 s while ready or unknown', () => { + const fn = vi.fn(); const p = createBridgePoller(fn); + p.sync('ready'); p.sync(undefined); + vi.advanceTimersByTime(BRIDGE_IDLE_POLL_MS - 1); expect(fn).not.toHaveBeenCalled(); + vi.advanceTimersByTime(1); expect(fn).toHaveBeenCalledTimes(1); + vi.advanceTimersByTime(BRIDGE_IDLE_POLL_MS); expect(fn).toHaveBeenCalledTimes(2); + p.stop(); + }); + it('polls every 5 s while restarting or unavailable', () => { + const fn = vi.fn(); const p = createBridgePoller(fn); + p.sync('restarting'); + vi.advanceTimersByTime(BRIDGE_POLL_MS); expect(fn).toHaveBeenCalledTimes(1); + p.sync('unavailable'); + vi.advanceTimersByTime(BRIDGE_POLL_MS); expect(fn).toHaveBeenCalledTimes(2); + p.stop(); + }); + it('switches rates without stacking timers', () => { + const fn = vi.fn(); const p = createBridgePoller(fn); + p.sync('ready'); p.sync('ready'); expect(vi.getTimerCount()).toBe(1); + p.sync('restarting'); p.sync('restarting'); expect(vi.getTimerCount()).toBe(1); + vi.advanceTimersByTime(BRIDGE_POLL_MS * 2); expect(fn).toHaveBeenCalledTimes(2); + p.sync('ready'); expect(vi.getTimerCount()).toBe(1); + fn.mockClear(); + vi.advanceTimersByTime(BRIDGE_IDLE_POLL_MS - 1); expect(fn).not.toHaveBeenCalled(); + vi.advanceTimersByTime(1); expect(fn).toHaveBeenCalledTimes(1); + p.stop(); + }); + it('skips ticks while the tab is hidden and resumes when visible', () => { + const doc = { visibilityState: 'hidden' }; + vi.stubGlobal('document', doc); + const fn = vi.fn(); const p = createBridgePoller(fn); + p.sync('restarting'); + vi.advanceTimersByTime(BRIDGE_POLL_MS * 3); expect(fn).not.toHaveBeenCalled(); + doc.visibilityState = 'visible'; + vi.advanceTimersByTime(BRIDGE_POLL_MS); expect(fn).toHaveBeenCalledTimes(1); + p.stop(); + }); + it('leaves nothing behind after stop', () => { + const fn = vi.fn(); const p = createBridgePoller(fn); + p.sync('restarting'); p.stop(); p.stop(); + expect(p.isActive()).toBe(false); expect(vi.getTimerCount()).toBe(0); + vi.advanceTimersByTime(BRIDGE_IDLE_POLL_MS * 2); expect(fn).not.toHaveBeenCalled(); + }); +}); diff --git a/ui/bridge-status.tsx b/ui/bridge-status.tsx new file mode 100644 index 0000000..81b3c02 --- /dev/null +++ b/ui/bridge-status.tsx @@ -0,0 +1,75 @@ +/** + * Local bridge health for the open project: a compact badge (only when the bridge is not ready), + * an inline notice with restart detail, and the poller that keeps it fresh while unhealthy. + */ +import React from 'react'; +import { Badge, type Tone } from './badge.js'; + +/** Mirrors `LocalBridgeHealth` / `bridgeStatus` from the backend workspace-setup response. */ +export type BridgeState = 'ready' | 'restarting' | 'unavailable'; +export interface BridgeHealth { + state: BridgeState; + logPath?: string; + lastExit?: { code: number | null; signal: string | null; at: string; uptimeMs: number; reason?: string }; + restarts: number; + nextRestartAt?: string; + restartDelayMs?: number; + message?: string; +} + +export const BRIDGE_POLL_MS = 5000; +export const BRIDGE_IDLE_POLL_MS = 30000; + +const TONE: Record<'restarting' | 'unavailable', Tone> = { restarting: 'warning', unavailable: 'failed' }; +const LABEL: Record<'restarting' | 'unavailable', string> = { restarting: 'Restarting', unavailable: 'Unavailable' }; + +export const bridgeNeedsAttention = (status?: string): status is 'restarting' | 'unavailable' => status === 'restarting' || status === 'unavailable'; + +/** Whole seconds until `iso`, or undefined when absent or unparseable. Never negative. */ +export function secondsUntil(iso: string | undefined, now: number): number | undefined { + if (!iso) return undefined; + const at = Date.parse(iso); + return Number.isFinite(at) ? Math.max(0, Math.ceil((at - now) / 1000)) : undefined; +} + +/** + * Polls `refetch` for as long as it is not stopped: every `attentionMs` while the bridge needs attention, every `idleMs` + * otherwise (the endpoint runs git, so a healthy bridge is polled slowly). Ticks are skipped while the tab is hidden. + * `sync` is idempotent and switches rate without stacking timers; `stop` clears everything. + */ +export function createBridgePoller(refetch: () => void, attentionMs = BRIDGE_POLL_MS, idleMs = BRIDGE_IDLE_POLL_MS) { + let timer: ReturnType | undefined; + let rate: number | undefined; + const stop = () => { if (timer !== undefined) { clearInterval(timer); timer = undefined; rate = undefined; } }; + const sync = (status?: string) => { + const next = bridgeNeedsAttention(status) ? attentionMs : idleMs; + if (timer !== undefined && rate === next) return; + stop(); + rate = next; + timer = setInterval(() => { if (typeof document !== 'undefined' && document.visibilityState === 'hidden') return; refetch(); }, next); + }; + return { sync, stop, isActive: () => timer !== undefined }; +} + +/** Header badge. Renders nothing when the bridge is ready or unknown, like the other healthy states. */ +export function BridgeBadge({ status }: { status?: string }): React.ReactElement | null { + if (!bridgeNeedsAttention(status)) return null; + return Bridge · {LABEL[status]}; +} + +export function BridgeNotice({ status, health, now = Date.now(), onReopen, onCopy, disabled }: { status?: string; health?: BridgeHealth; now?: number; onReopen?: () => void; onCopy?: (text: string) => void; disabled?: boolean }): React.ReactElement | null { + if (!bridgeNeedsAttention(status)) return null; + const wait = status === 'restarting' ? secondsUntil(health?.nextRestartAt, now) : undefined; + const message = health?.message ?? (status === 'restarting' ? 'The local bridge stopped and is being restarted.' : 'The local bridge is not running.'); + const restarts = health ? `${health.restarts} ${health.restarts === 1 ? 'restart' : 'restarts'}` : ''; + const next = wait !== undefined ? `next restart in ${wait}s` : ''; + return ( +
+ Local bridge {LABEL[status].toLowerCase()} + {message} + {(restarts || next) && {[restarts, next].filter(Boolean).join(' · ')}} + {health?.logPath && {health.logPath}{onCopy && }} + {status === 'unavailable' && Reopening the repository starts a fresh bridge.{onReopen && }} +
+ ); +} diff --git a/ui/main.tsx b/ui/main.tsx index 3e9f79b..f340326 100644 --- a/ui/main.tsx +++ b/ui/main.tsx @@ -5,6 +5,7 @@ import { postDecision, useDecisionDigest, type DecisionOutcome } from './decisio import { ChecksPipeline, type CiJobFailure } from './checks-pipeline.js'; import { NetworkToggle, RepoChecksEditor, checksFromSuggestions, validationCommandsPayload, type RepoCheck, type RepoCheckSuggestion } from './repo-checks.js'; import { PrDraftPanel, type PrDraftData } from './pr-draft.js'; +import { BridgeBadge, BridgeNotice, createBridgePoller, type BridgeHealth } from './bridge-status.js'; import { Badge, toneForStatus, type Tone } from './badge.js'; import { createRoot } from 'react-dom/client'; import './style.css'; @@ -48,7 +49,7 @@ export type Task = { id: string; title: string; goal?:string; suggestedAllowedPa export type Project = { id: string; name: string; repoPath?:string; status?: string; plannerSession?:Session; plannerMessages?:PlannerMessage[]; plannerAssignments?:Assignment[]; plannerTurnCount?:number; defaultRoleConfigs?: Record; usage?: UsageMetrics; usageByHarnessModel?: Record; tasks?: Task[] }; type RepoEntry={name:string;path:string;isGitRepo:boolean}; type RepoBrowse={currentPath:string;parentPath?:string;entries:RepoEntry[];roots:{name:string;path:string}[];truncated:boolean}; -type WorkspaceSetup={repoPath:string;head:string;dirty:boolean;allowedScope:string[];validationCommands:{name:string;command:string;args:string[];network?:boolean}[];bridgeStatus?:string}; +type WorkspaceSetup={repoPath:string;head:string;dirty:boolean;allowedScope:string[];validationCommands:{name:string;command:string;args:string[];network?:boolean}[];bridgeStatus?:'ready'|'restarting'|'unavailable';bridgeHealth?:BridgeHealth}; export type TaskStartPreview={taskId?:string;task?:Task;roleConfigs?:Record;scope?:string[];validationCriteria?:string[];validationCommands?:{name:string;command:string;args:string[];network?:boolean}[];budgets?:RunBudgets;baseCommit?:string;currentHead?:string;requiredBaseCommits?:string[];requiresExplicitBase:boolean;reasons?:string[];overlappingTaskIds?:string[];dependencyTaskIds?:string[];dependencies?:Task[];canStart?:boolean;ciChecksNotConfigured?:string[]}; type PrDraft = PrDraftData; type RepoInspect={repoPath:string;head:string;dirty:boolean;suggestedAllowedScope:string[];suggestedValidationCommands:RepoCheckSuggestion[];trackedFiles:string[];ciScripts?:string[]}; @@ -264,7 +265,11 @@ export function App({initialState,initialWorkspaceSetup,initialTaskStartPreview, useEffect(() => { if (task && task.id !== selectedTask) setSelectedTask(task.id); }, [task?.id]); useEffect(() => { if (run && run.id !== selectedRun) setSelectedRun(run.id); }, [run?.id]); useEffect(() => { setBaseCommit(run?.pinnedBaseCommit ?? ''); setBudgets(run?.controller?.budgets??{roleTurns:{planner:3,orchestrator:3,worker:2,reviewer:2},workerAttempts:2}); }, [run?.id]); - useEffect(()=>{if(!project)return;void api(`/api/projects/${project.id}/workspace-setup`).then(setWorkspaceSetup).catch(()=>setWorkspaceSetup(undefined));},[project?.id]); + const loadWorkspaceSetup=useCallback((keepOnError=false)=>{if(!project)return;void api(`/api/projects/${project.id}/workspace-setup`).then(setWorkspaceSetup).catch(()=>{if(!keepOnError)setWorkspaceSetup(undefined);});},[project?.id]); + useEffect(()=>{loadWorkspaceSetup();},[loadWorkspaceSetup]); + // Bridge health stays fresh by polling (never on SSE: the endpoint runs git): 5 s while the bridge needs attention, 30 s otherwise; hidden tabs skip. + const bridgeStatus=workspaceSetup?.bridgeStatus; + useEffect(()=>{if(!project?.id)return;const poller=createBridgePoller(()=>loadWorkspaceSetup(true));poller.sync(bridgeStatus);return poller.stop;},[bridgeStatus,project?.id,loadWorkspaceSetup]); useEffect(()=>{setTaskEdit(false);setTaskTitle(task?.title??'');setTaskGoal(task?.goal??'');setTaskPaths((task?.suggestedAllowedPaths??[]).join('\n'));setTaskCriteria((task?.validationCriteria??[]).join('\n'));setTaskDeps((task?.dependsOn??[]).join(', '));setWorkScope('');setWorkChecks([]);setTaskAdvancedEdited(false);setRoleChoices(Object.fromEntries((state.roles??[]).filter(r=>r.enabled).map(r=>[r.id,project?.defaultRoleConfigs?.[r.id]??r.config??{}])));},[task?.id,project?.id]); useEffect(()=>{setTaskBaseCommit('');},[task?.id]); const taskPreviewRevision=JSON.stringify((project?.tasks??[]).map(item=>({id:item.id,status:item.status,paths:item.suggestedAllowedPaths,dependsOn:item.dependsOn,runs:(item.runs??[]).map(r=>({id:r.id,status:r.status,active:r.controller?.active,phase:r.controller?.phase,promotion:r.promotion?.status,resultCommit:r.promotion?.resultCommit,evidence:Boolean(r.workerEvidence)}))}))); @@ -801,7 +806,7 @@ export function App({initialState,initialWorkspaceSetup,initialTaskStartPreview, return
F
Foreman v2Local Engineering Control Plane
-
{project?<>{project.name}{task&&<>{task.title}}
Repository details
{workspaceSetup?.repoPath??project.repoPath??project.name}{workspaceSetup?`${workspaceSetup.dirty?'Uncommitted changes':'Clean'} · HEAD ${workspaceSetup.head.slice(0,10)}`:'Repository status unavailable'}{workspaceSetup?.allowedScope.length?`${workspaceSetup.allowedScope.length} allowed paths`:'No allowed paths configured'}{Boolean(workspaceSetup?.allowedScope.length)&&{workspaceSetup!.allowedScope.join(', ')}}
:<>No repository open}
+
{project?<>{project.name}{task&&<>{task.title}}
Repository details
{workspaceSetup?.repoPath??project.repoPath??project.name}{workspaceSetup?`${workspaceSetup.dirty?'Uncommitted changes':'Clean'} · HEAD ${workspaceSetup.head.slice(0,10)}`:'Repository status unavailable'}{workspaceSetup?.allowedScope.length?`${workspaceSetup.allowedScope.length} allowed paths`:'No allowed paths configured'}{Boolean(workspaceSetup?.allowedScope.length)&&{workspaceSetup!.allowedScope.join(', ')}}{openRepoDialog();if(project.repoPath)void selectRepo(project.repoPath);}} onCopy={text=>{void navigator.clipboard?.writeText(text).catch(()=>{});}} disabled={pending}/>
:<>No repository open}
{(['uhp','memory'] as const).map(key=>{const service=services?.[key];return {key==='uhp'?'UHP':'Memory'}{service?.status==='ready'?'':` · ${service?serviceStatus(service):'Unknown'}`}})}
{online ? 'Online' : 'Connecting'}
diff --git a/ui/style.css b/ui/style.css index 60feb3d..c7961c4 100644 --- a/ui/style.css +++ b/ui/style.css @@ -118,6 +118,15 @@ select option{background:var(--panel)} .repo-details[open]>summary{color:var(--text)} .repo-details-panel{position:absolute;z-index:30;top:26px;left:0;display:grid;gap:var(--sp-2);width:min(400px,80vw);padding:var(--sp-3);border:1px solid var(--line-strong);border-radius:var(--r-lg);background:var(--panel);box-shadow:0 12px 32px #0009;color:var(--dim);font-size:var(--fs-xs)} .repo-details-panel code{color:var(--text)} +.repo-details>summary .bridge-badge{margin-left:var(--sp-2);vertical-align:middle} +@media(max-width:1180px){.bridge-badge .bridge-word{display:none}} +.bridge-notice{display:grid;gap:var(--sp-1);padding:var(--sp-2) var(--sp-3);border:1px solid var(--t-bd,var(--line));border-left-width:3px;border-radius:var(--r-md);background:var(--t-bg,var(--panel-2,var(--panel)));color:var(--dim);font-size:var(--fs-xs);overflow-wrap:anywhere} +.bridge-notice.bridge-restarting{--t-fg:var(--warning-fg);--t-bg:var(--warning-bg);--t-bd:var(--warning-bd)} +.bridge-notice.bridge-unavailable{--t-fg:var(--failed-fg);--t-bg:var(--failed-bg);--t-bd:var(--failed-bd)} +.bridge-notice>b{color:var(--t-fg);font-size:var(--fs-sm)} +.bridge-meta{color:var(--muted)} +.bridge-log,.bridge-reopen{display:flex;align-items:center;flex-wrap:wrap;gap:var(--sp-2)} +.bridge-log code{flex:1 1 0;min-width:0;padding:2px var(--sp-1);border-radius:var(--r-sm,4px);user-select:all;word-break:break-all} .project-switch{flex:none} .workspace-nav{display:flex;align-items:center;gap:2px;margin-left:auto;padding:2px;border:1px solid var(--line);border-radius:var(--r-md);background:var(--well)} .workspace-nav button,.mobile-inspector-tabs button{height:var(--ctl-sm);padding:0 var(--sp-3);border:0;border-radius:var(--r-sm);background:transparent;color:var(--dim);font-size:var(--fs-sm);font-weight:500;white-space:nowrap} From 654ca877d52c542f7f230b2cb6f4ea22378aaca0 Mon Sep 17 00:00:00 2001 From: Claude Date: Mon, 28 Sep 2026 22:11:07 +0000 Subject: [PATCH 2/3] Recover unfinished work in the background and share the reviewer check Startup and project open awaited Controller.recover(), which can replay a submission against an external UHP server. A slow or hung server kept Foreman from listening and made opening a project hang. - Startup and project open now start recovery without waiting for it. Concurrent calls share one in-flight recovery per controller. - A failure is recorded as a recovery.failed event and written to stderr with credentials masked, never left as an unhandled rejection. - Recovery claims the assignments it is handling, and the 5 s reconcile loop skips only those, so the two never process one assignment at once. Recovery only stops runs that were active when it started, so a run started meanwhile is left alone. A late "still running" observation can no longer overwrite a terminal assignment. - Recovery state (idle, running or failed, with timestamps and the last error) is reported in service status and workspace-setup. - Shutdown does not wait for recovery, and its final flush is bounded. The rule that decides whether a Reviewer response proves the complete read-only evidence package was written out twice, in the live submit path and in reconcile. It is now one method used by both, with each site's error text and events unchanged. Tests pinning both sites (a pass case and eight rejection cases each) pass before and after. Co-Authored-By: Claude Opus 5.5 Claude-Session: https://claude.ai/code/session_01DXCvrGWmmRNDLmy7nYS2QP --- src/controller.ts | 39 +++++++++-- src/server.ts | 11 +-- tests/recovery.test.ts | 114 ++++++++++++++++++++++++++++++++ tests/reviewer-evidence.test.ts | 57 ++++++++++++++++ 4 files changed, 210 insertions(+), 11 deletions(-) create mode 100644 tests/recovery.test.ts create mode 100644 tests/reviewer-evidence.test.ts diff --git a/src/controller.ts b/src/controller.ts index 7f2af30..2ff69b4 100644 --- a/src/controller.ts +++ b/src/controller.ts @@ -62,6 +62,10 @@ export class Controller { private readonly retentionInFlight = new Set(); private readonly workerRetryInFlight = new Set(); private readonly cancellationInFlight = new Map>(); + private recoveryInFlight?: Promise; + private recoveryState: RecoveryStatus = {status:'idle'}; + /** Assignments recover() has yet to settle; the reconcile loop leaves these alone so the two never handle one assignment at once. */ + private readonly recoveringAssignments = new Set(); private readonly prDraftInFlight = new Set(); /** In-memory cache of seeded read-only snapshot workspaces: commit → workspaceId. */ private readonly snapshotWorkspaceCache = new Map(); @@ -80,7 +84,7 @@ export class Controller { private async workspaceConfigForRun(runId:string){const base=this.verifiedWorkspaceConfig;if(!base)return undefined;const run=this.findRun(await this.store.load(),runId);return {...base,allowedScope:run.allowedScope??base.allowedScope,commands:(run.validationCommands as ValidationCommand[]|undefined)??base.commands};} async pinWorkerBase(runId:string,baseCommit:string):Promise{const config=this.verifiedWorkspaceConfig;if(!config)throw Object.assign(new Error('Verified workspace workflow is not configured'),{statusCode:503});const snapshot=await snapshotGitCommit(config.repoPath,baseCommit);await this.store.mutate(s=>{const run=this.findRun(s,runId);if(run.pinnedBaseCommit&&run.pinnedBaseCommit.toLowerCase()!==baseCommit.toLowerCase())throw Object.assign(new Error('Run base is already pinned'),{statusCode:409});run.pinnedBaseCommit=baseCommit.toLowerCase();const scope=run.allowedScope??config.allowedScope;run.baseFilePaths=snapshot.entries.map(entry=>entry.path).filter(path=>scopeContainedBy(path,scope)).slice(0,200);s.events.push(event('worker.base_pinned','run',runId,{pinnedBaseCommit:run.pinnedBaseCommit,provenance:'fixture_or_operator'}));});} async prepareWorkerWorkspace(runId:string,baseCommit:string):Promise{const config=this.verifiedWorkspaceConfig;if(!config?.bridgeBaseUrl)throw Object.assign(new Error('Workspace bridge URL is not configured'),{statusCode:503});const current=this.findRun(await this.store.load(),runId);if(current.workspaceId){if(current.pinnedBaseCommit?.toLowerCase()!==baseCommit.toLowerCase())throw Object.assign(new Error('Run workspace is already pinned to a different base'),{statusCode:409});return current;}await this.pinWorkerBase(runId,baseCommit);const seeded=await seedBridgeWorkspace(config.bridgeBaseUrl,baseCommit,{token:config.bridgeToken});await this.store.mutate(s=>{const run=this.findRun(s,runId);if(run.workspaceId&&run.workspaceId!==seeded.workspaceId)throw Object.assign(new Error('Run already has a different workspace'),{statusCode:409});run.workspaceId=seeded.workspaceId;s.events.push(event('worker.workspace_seeded','run',runId,{workspaceId:run.workspaceId,pinnedBaseCommit:run.pinnedBaseCommit}));});return this.findRun(await this.store.load(),runId);} - async serviceStatus() { const validationSandbox=await validationSandboxStatus(this.verifiedWorkspaceConfig?.sandbox);const discovery=this.discovery?{version:this.discovery.version,capabilities:this.discovery.capabilities,harnesses:this.discovery.harnesses.map(h=>({id:h.id,models:[...(h.models??[]),...(this.discovery!.models??[]).filter(m=>(m as any).harnessId===h.id)]}))}:undefined;return { uhp:{status:this.discovery?'ready':this.discoveryError?'degraded':this.uhpConfigured?'unavailable':'not_configured',configured:this.uhpConfigured,discovery,error:this.discoveryError},memory:{status:this.memoryServiceState,configured:this.memoryConfigured},validationSandbox }; } + async serviceStatus() { const validationSandbox=await validationSandboxStatus(this.verifiedWorkspaceConfig?.sandbox);const discovery=this.discovery?{version:this.discovery.version,capabilities:this.discovery.capabilities,harnesses:this.discovery.harnesses.map(h=>({id:h.id,models:[...(h.models??[]),...(this.discovery!.models??[]).filter(m=>(m as any).harnessId===h.id)]}))}:undefined;return { uhp:{status:this.discovery?'ready':this.discoveryError?'degraded':this.uhpConfigured?'unavailable':'not_configured',configured:this.uhpConfigured,discovery,error:this.discoveryError},memory:{status:this.memoryServiceState,configured:this.memoryConfigured},validationSandbox,recovery:this.recoveryStatus() }; } async refreshDiscovery() { if(!this.uhpConfigured||!this.uhp.discover){this.discoveryError=undefined;return;} try { this.discovery=await this.uhp.discover(true); this.discoveryError=undefined; await this.store.mutate(s=>{const configs:RoleConfig[]=this.discovery!.harnesses.flatMap(h=>[...(h.models??[]),...(this.discovery!.models??[]).filter(m=>(m as any).harnessId===h.id)].filter(m=>m.available!==false).map(m=>({harnessId:h.id,model:m.id})));for(const role of s.roles){role.availableConfigs=structuredClone(configs);role.enabled=configs.length>0;const current=configs.find(c=>c.harnessId===role.config.harnessId&&c.model===role.config.model),configured=this.configuredRoleChoice&&configs.find(c=>c.harnessId===this.configuredRoleChoice!.harnessId&&c.model===this.configuredRoleChoice!.model);if(!current&&(!role.config.harnessId||!role.config.model))role.config=structuredClone(configured??defaultRoleConfig(role.id,configs)??{harnessId:'',model:''});}s.events.push(event('uhp.discovery_updated','service','uhp',{version:this.discovery!.version,configCount:configs.length}));}); } @@ -716,21 +720,39 @@ export class Controller { const prep = await this.store.mutate(s=>{ const {project,task,run}=this.findRunContext(s,runId); const role=must(s.roles.find(x=>x.id===roleId),'Role'); const requested=config ?? resolveRoleConfig(s.roles,project,run,roleId).config ?? role.config; validateRoleConfig(role,requested); if(!prompt.trim()) throw new Error('Assignment prompt is required');assertUhpPromptWithinLimit(prompt.trim(),`${roleId} assignment prompt`);if(run.controller){const roleCount=run.assignments.filter(a=>a.roleId===roleId).length,limit=run.controller.budgets.roleTurns[roleId as RoleKind],lastReviewer=run.assignments.filter(a=>a.roleId==='reviewer').at(-1),authorizedReviewerRetry=roleId==='reviewer'&&run.reviewerRetryAuthorized&&lastReviewer?.status==='failed'&&roleCount===limit;if(roleCount>=limit&&!authorizedReviewerRetry)throw Object.assign(new Error(`${roleId} role-turn budget exhausted (${roleCount}/${limit})`),{statusCode:429});if(roleId==='worker'&&roleCount>=run.controller.budgets.workerAttempts)throw Object.assign(new Error(`Worker attempt budget exhausted (${roleCount}/${run.controller.budgets.workerAttempts})`),{statusCode:429});}if(this.verifiedWorkspaceConfig&&run.approval)throw Object.assign(new Error('No assignments are allowed after the human decision; retry in a new run'),{statusCode:409});if(roleId==='worker'&&this.verifiedWorkspaceConfig){const retryDispatch=dispatchToken===INTERNAL_WORKER_RETRY,proposal=run.workerProposal,source=proposal&&run.assignments.find(a=>a.id===proposal.orchestratorAssignmentId);if(dispatchToken!==INTERNAL_WORKER_DISPATCH&&!retryDispatch||!proposalId||proposal?.id!==proposalId||proposal.status!=='dispatching'||proposal.text!==prompt.trim()||!source||source.roleId!=='orchestrator'||source.status!=='succeeded'||parseWorkerProposal(source.result)!==proposal.text||!run.pinnedBaseCommit||this.verifiedWorkspaceConfig.bridgeBaseUrl&&!run.workspaceId||JSON.stringify(requested)!==JSON.stringify(resolveRoleConfig(s.roles,project,run,'worker').config??role.config))throw Object.assign(new Error('Verified Worker assignment requires a matching controller-dispatched Orchestrator proposal'),{statusCode:409});const workers=run.assignments.filter(a=>a.roleId==='worker'),archivedIds=new Set((run.workerEvidenceHistory??[]).map(e=>e.workerAssignmentId));if(retryDispatch?(!this.workerRetryKind(run,proposal!)||workers.some(a=>!archivedIds.has(a.id)&&!['failed','cancelled'].includes(a.status)&&!(a.id===proposal!.workerAssignmentId&&this.workerRetryKind(run,proposal!)!=='transport_failure'))):workers.length>0)throw Object.assign(new Error('A verified workflow permits only an authorized Worker retry after transport failure, failed validation, or Reviewer feedback'),{statusCode:409});if(retryDispatch&&this.workerRetryKind(run,proposal!)!=='transport_failure'){run.workerEvidenceHistory??=[];run.workerEvidenceHistory.push(structuredClone(run.workerEvidence!));run.validationHistory??=[];run.validationHistory.push(structuredClone(run.validation!));if(run.orchestratorInbox){run.orchestratorInboxHistory??=[];run.orchestratorInboxHistory.push(structuredClone(run.orchestratorInbox));delete run.orchestratorInbox;}delete run.workerEvidence;delete run.validation;s.events.push(event('worker.retry_evidence_archived','run',runId,{proposalId,workerAssignmentId:proposal!.workerAssignmentId,evidenceCount:run.workerEvidenceHistory.length,validationCount:run.validationHistory.length}));}}let reviewerHandoff:{inboxId:string;evidenceDigest:string}|undefined;if(roleId==='reviewer'){const old=run.assignments.some(a=>a.roleId==='reviewer'&&['submitting','submitted','running','cancel_requested'].includes(a.status))||run.assignments.some(a=>a.roleId==='reviewer'&&a.status==='succeeded'&&!run.reviewerRetryAuthorized)||run.assignments.some(a=>a.roleId==='reviewer'&&a.status==='failed'&&!run.reviewerRetryAuthorized);const evidence=this.reviewerEvidencePackage(run);if(!run.workerEvidence?.scopeVerified||run.validation?.status!=='passed'||run.reviewerRecommendation||old||prompt.trim()!==this.reviewerPrompt()||JSON.stringify(requested)!==JSON.stringify(resolveRoleConfig(s.roles,project,run,'reviewer').config??role.config)||requested.options?.workspace_id!==undefined)throw Object.assign(new Error('Reviewer submission must use the controller evidence prompt and selected read-only Reviewer configuration'),{statusCode:409});if(typeof evidence.orchestratorInboxId==='string'&&typeof evidence.orchestratorEvidenceDigest==='string')reviewerHandoff={inboxId:evidence.orchestratorInboxId,evidenceDigest:evidence.orchestratorEvidenceDigest};run.reviewerRetryAuthorized=false;}const a: Assignment={id:assignmentId,roleId,status:'submitting',requestedConfig:structuredClone(requested),...(reviewerHandoff?{reviewerInboxId:reviewerHandoff.inboxId,reviewerEvidenceDigest:reviewerHandoff.evidenceDigest}:{}),prompt:prompt.trim(),submissionId,idempotencyKey:key,createdAt:now()}; run.assignments.push(a); if(roleId==='worker')run.status='running';else if(!run.guidance.some(g=>g.status==='queued'||g.status==='delivered'))run.status=roleId==='reviewer'?'review':'planning';s.events.push(event('assignment.submission_intent','assignment',a.id,{runId,roleId,submissionId,idempotencyKey:key,requestedConfig:requested}));if(reviewerHandoff)s.events.push(event('reviewer.evidence_handoff','run',runId,{assignmentId:a.id,inboxId:reviewerHandoff.inboxId,evidenceDigest:reviewerHandoff.evidenceDigest}));return {a,project,task,run}; }); return this.submitExisting(prep.project.id,prep.task.id,prep.run.id,prep.a.id); } - async recover(): Promise { + recoveryStatus(): RecoveryStatus { return {...this.recoveryState}; } + /** Runs recover() at most once at a time per controller; concurrent callers share the in-flight promise. Rejections are recorded as a failed state and a recovery.failed event, then rethrown to the caller. */ + recover(): Promise { + if(this.recoveryInFlight)return this.recoveryInFlight; + const startedAt=now(),lastError=this.recoveryState.lastError,lastErrorAt=this.recoveryState.lastErrorAt; + this.recoveryState={status:'running',startedAt,...(lastError?{lastError,lastErrorAt}:{})}; + const tracked:Promise=this.runRecovery().then(()=>{this.recoveryState={status:'idle',startedAt,finishedAt:now()};},async error=>{const message=safeRecoveryMessage(error),at=now();this.recoveryState={status:'failed',startedAt,finishedAt:at,lastError:message,lastErrorAt:at};await this.store.mutate(s=>{s.events.push(event('recovery.failed','service','recovery',{error:message}));}).catch(()=>undefined);throw error;}).finally(()=>{if(this.recoveryInFlight===tracked)this.recoveryInFlight=undefined;}); + this.recoveryInFlight=tracked;return tracked; + } + /** Fire-and-forget recovery for startup and project open: never rejects, logs a redacted one-line failure to stderr. Returns the shared in-flight promise so callers and tests can await settlement. */ + startRecovery(label='controller'): Promise { + return this.recover().catch(error=>{process.stderr.write(`Recovery failed for ${label}: ${safeRecoveryMessage(error)}\n`);}); + } + private async runRecovery(): Promise { const s=await this.store.load(); + const activeAtStart=new Set(); + this.recoveringAssignments.clear(); + for(const p of s.projects){if(this.projectScopeId&&p.id!==this.projectScopeId)continue;for(const t of p.tasks)for(const r of t.runs){if(r.controller?.active)activeAtStart.add(r.id);for(const a of r.assignments)if(a.status==='submitting'||a.status==='running'||a.status==='cancel_requested')this.recoveringAssignments.add(a.id);}} + try { for (const p of s.projects) { if(this.projectScopeId&&p.id!==this.projectScopeId)continue; for(const t of p.tasks) for(const r of t.runs) for(const a of r.assignments) { if(a.status==='submitting'&&isDeterministicUhpClientRejection(a.error)) { await this.store.mutate(state=>{const {a:current}=this.findAssignment(state,a.id);if(current.status==='submitting'&&isDeterministicUhpClientRejection(current.error)){current.status='failed';current.error=`${current.error}; no provider execution was accepted; explicit retry is available`;state.events.push(event('assignment.submit_rejected','assignment',current.id,{error:current.error,idempotencyKey:current.idempotencyKey,recovered:true}));}});continue; } if(a.status==='cancel_requested') { if(a.externalId) { await this.dispatchCancel(a).catch(()=>undefined);const latest=this.findAssignment(await this.store.load(),a.id).a;if(latest.status==='cancel_requested'&&latest.responseId)await this.reconcileExisting(a.id).catch(()=>undefined); } else await this.submitExisting(p.id,t.id,r.id,a.id,true).catch(()=>undefined); } else if(a.status==='submitting'||a.status==='running') { if(a.responseId&&this.uhp.retrieve) await this.reconcileExisting(a.id).catch(()=>undefined); else if(a.status==='submitting') await this.submitExisting(p.id,t.id,r.id,a.id,true).catch(()=>undefined); } }} const repaired=await this.store.load();for(const p of repaired.projects){if(this.projectScopeId&&p.id!==this.projectScopeId)continue;for(const t of p.tasks)for(const r of t.runs)await this.recoverPromptLimitRecommendation(r.id);} - if(this.verifiedWorkspaceConfig){const latest=await this.store.load();for(const p of latest.projects){if(this.projectScopeId&&p.id!==this.projectScopeId)continue;for(const t of p.tasks)for(const r of t.runs){if(r.controller?.active){await this.stopAutomaticRun(r.id,r.workerEvidence&&!r.validation?'Controller restarted before validation evidence completed; inspect evidence and explicitly retry validation.':'Controller restarted mid-run; inspect saved assignments and continue only through explicit recovery actions.');}if(r.workerEvidence&&r.validation?.status==='passed'&&!r.reviewerRecommendation){const reviewer=r.assignments.filter(a=>a.roleId==='reviewer'&&a.status==='succeeded').at(-1);if(reviewer)await this.recordReviewerRecommendation(r.id,reviewer.id,r.workerEvidence.provenance==='recorded_replay'?'simulated_fixture':'uhp_response').catch(()=>undefined);}}}} + if(this.verifiedWorkspaceConfig){const latest=await this.store.load();for(const p of latest.projects){if(this.projectScopeId&&p.id!==this.projectScopeId)continue;for(const t of p.tasks)for(const r of t.runs){if(r.controller?.active&&activeAtStart.has(r.id)){await this.stopAutomaticRun(r.id,r.workerEvidence&&!r.validation?'Controller restarted before validation evidence completed; inspect evidence and explicitly retry validation.':'Controller restarted mid-run; inspect saved assignments and continue only through explicit recovery actions.');}if(r.workerEvidence&&r.validation?.status==='passed'&&!r.reviewerRecommendation){const reviewer=r.assignments.filter(a=>a.roleId==='reviewer'&&a.status==='succeeded').at(-1);if(reviewer)await this.recordReviewerRecommendation(r.id,reviewer.id,r.workerEvidence.provenance==='recorded_replay'?'simulated_fixture':'uhp_response').catch(()=>undefined);}}}} await this.resumePendingRetentions(); await this.migrateProjectRoleConfigs(); + } finally { this.recoveringAssignments.clear(); } } - async reconcileRunning():Promise{const s=await this.store.load();for(const p of s.projects){if(this.projectScopeId&&p.id!==this.projectScopeId)continue;for(const t of p.tasks)for(const r of t.runs)for(const a of r.assignments)if((a.status==='running'||a.status==='cancel_requested')&&a.responseId)await this.reconcileExisting(a.id).catch(()=>undefined);}} + async reconcileRunning():Promise{const s=await this.store.load();for(const p of s.projects){if(this.projectScopeId&&p.id!==this.projectScopeId)continue;for(const t of p.tasks)for(const r of t.runs)for(const a of r.assignments)if((a.status==='running'||a.status==='cancel_requested')&&a.responseId&&!this.recoveringAssignments.has(a.id))await this.reconcileExisting(a.id).catch(()=>undefined);}} async refreshAssignment(assignmentId:string):Promise{return this.reconcileExisting(assignmentId).then(async()=>this.findAssignment(await this.store.load(),assignmentId).a);} - private async reconcileExisting(assignmentId:string):Promise{const snapshot=await this.store.load(),{a}=this.findAssignment(snapshot,assignmentId);if(!a.responseId||!this.uhp.retrieve)return;const response=await this.uhp.retrieve(a.responseId);const terminal=normalizeStatus(response.status);if(['succeeded','failed','needs_revision','cancelled'].includes(terminal)){await this.store.mutate(s=>{const {a:r,run}=this.findAssignment(s,assignmentId);const wasCancelRequested=r.status==='cancel_requested';r.status=terminal;if(terminal==='succeeded')delete r.error;else if(terminal==='failed'){const detail=typeof response.error==='string'?response.error:response.error&&typeof response.error==='object'&&typeof (response.error as {message?:unknown}).message==='string'?(response.error as {message:string}).message:undefined;if(detail?.trim())r.error=detail.replace(/[\u0000-\u001f\u007f]/g,' ').trim().slice(0,2000);}if(wasCancelRequested&&terminal!=='cancelled')s.events.push(event('assignment.cancel_raced_terminal','assignment',r.id,{status:terminal,source:'reconciled-response'}));r.result=response.output_text??response.output??response.error;r.externalId=response.id;r.responseId=response.id;const meta=response.metadata??{};const session=meta.session_id??response.session_id??response.sessionId;if(typeof session==='string')r.sessionId=session;const requested=response.requested_model??meta.requested_model;if(typeof requested==='string')r.requestedModel=requested;r.selectedHarnessId=typeof meta.harness_id==='string'?meta.harness_id:undefined;r.reportedHarnessId=typeof meta.harness_id==='string'?meta.harness_id:undefined;r.modelFallback=meta.model_fallback===true;if(typeof response.model==='string'){r.actualModelStatus='observed';if(typeof meta.harness_id==='string'){r.actualConfig={harnessId:meta.harness_id,model:response.model};r.configOutcome=r.actualConfig.model===r.requestedConfig.model&&r.actualConfig.harnessId===r.requestedConfig.harnessId?'confirmed':'substituted';}}else if(isCliHarness(r.requestedConfig.harnessId)&&meta.actual_model_status==='unavailable')r.actualModelStatus='unavailable';if(isCliHarness(r.requestedConfig.harnessId))r.cliInvocation=cliInvocationFrom(meta.cli_invocation);const usage=usageFromResponse(response.usage);if(usage)r.usage={...r.usage,...usage,measured:true};if(r.roleId==='reviewer'){r.reviewerExecution={mode:typeof meta.foreman_review_mode==='string'?meta.foreman_review_mode:'',mutationAttempted:meta.reviewer_mutation_attempted===true,validation:meta.reviewer_validation};const recoveredModelMatches=reviewerModelEvidence({actualModel:response.model,actualModelStatus:r.actualModelStatus,requestedModel:r.requestedModel,selectedHarnessId:r.selectedHarnessId,reportedHarnessId:r.reportedHarnessId,modelFallback:r.modelFallback,cliInvocation:r.cliInvocation},r.requestedConfig);let valid=terminal==='succeeded'&&recoveredModelMatches&&meta.harness_id===r.requestedConfig.harnessId&&!!r.sessionId&&r.responseId!==run.workerEvidence?.responseId&&r.sessionId!==run.assignments.find(x=>x.id===run.workerEvidence?.workerAssignmentId)?.sessionId&&meta.foreman_review_mode==='read_only'&&meta.reviewer_mutation_attempted===false;try{valid=valid&&canonicalJson(meta.reviewer_validation)===canonicalJson(this.reviewerEvidencePackage(run).controllerValidation);}catch{valid=false;}if(!valid){r.status='failed';r.error='Reconciled Reviewer response did not prove the complete read-only evidence package';s.events.push(event('reviewer.execution_rejected','assignment',r.id,{responseId:r.responseId,sessionId:r.sessionId,actualModel:response.model,source:'reconciled-response'}));}}if(r.roleId==='planner'||r.roleId==='orchestrator'){const session=run.sessions[r.roleId];session.responseId=response.id;if(r.sessionId)session.uhpSessionId=r.sessionId;if(r.roleId==='planner')run.plannerSessionId=session.uhpSessionId;else run.orchestratorSessionId=session.uhpSessionId;}if(r.roleId==='worker'){const pendingGuidance=run.guidance.some(g=>g.status==='queued'||g.status==='delivered');run.status=terminal==='failed'?'failed':terminal==='cancelled'?(pendingGuidance?'waiting_guidance':'planning'):'review';}s.events.push(event('assignment.reconciled','assignment',r.id,{responseId:r.responseId,status:r.status,requestedModel:r.requestedModel,actualModelStatus:r.actualModelStatus,cliInvocation:r.cliInvocation}));pruneHighVolumeEvents(s.events);});}else if(a.status!=='cancel_requested'&&a.status!=='running')await this.store.mutate(s=>{const {a:r}=this.findAssignment(s,assignmentId);if(r.status==='cancel_requested'||r.status==='running')return;r.status='running';s.events.push(event('assignment.reconciled','assignment',r.id,{responseId:r.responseId,status:response.status}));pruneHighVolumeEvents(s.events);});} + private async reconcileExisting(assignmentId:string):Promise{const snapshot=await this.store.load(),{a}=this.findAssignment(snapshot,assignmentId);if(!a.responseId||!this.uhp.retrieve)return;const response=await this.uhp.retrieve(a.responseId);const terminal=normalizeStatus(response.status);if(['succeeded','failed','needs_revision','cancelled'].includes(terminal)){await this.store.mutate(s=>{const {a:r,run}=this.findAssignment(s,assignmentId);const wasCancelRequested=r.status==='cancel_requested';r.status=terminal;if(terminal==='succeeded')delete r.error;else if(terminal==='failed'){const detail=typeof response.error==='string'?response.error:response.error&&typeof response.error==='object'&&typeof (response.error as {message?:unknown}).message==='string'?(response.error as {message:string}).message:undefined;if(detail?.trim())r.error=detail.replace(/[\u0000-\u001f\u007f]/g,' ').trim().slice(0,2000);}if(wasCancelRequested&&terminal!=='cancelled')s.events.push(event('assignment.cancel_raced_terminal','assignment',r.id,{status:terminal,source:'reconciled-response'}));r.result=response.output_text??response.output??response.error;r.externalId=response.id;r.responseId=response.id;const meta=response.metadata??{};const session=meta.session_id??response.session_id??response.sessionId;if(typeof session==='string')r.sessionId=session;const requested=response.requested_model??meta.requested_model;if(typeof requested==='string')r.requestedModel=requested;r.selectedHarnessId=typeof meta.harness_id==='string'?meta.harness_id:undefined;r.reportedHarnessId=typeof meta.harness_id==='string'?meta.harness_id:undefined;r.modelFallback=meta.model_fallback===true;if(typeof response.model==='string'){r.actualModelStatus='observed';if(typeof meta.harness_id==='string'){r.actualConfig={harnessId:meta.harness_id,model:response.model};r.configOutcome=r.actualConfig.model===r.requestedConfig.model&&r.actualConfig.harnessId===r.requestedConfig.harnessId?'confirmed':'substituted';}}else if(isCliHarness(r.requestedConfig.harnessId)&&meta.actual_model_status==='unavailable')r.actualModelStatus='unavailable';if(isCliHarness(r.requestedConfig.harnessId))r.cliInvocation=cliInvocationFrom(meta.cli_invocation);const usage=usageFromResponse(response.usage);if(usage)r.usage={...r.usage,...usage,measured:true};if(r.roleId==='reviewer'){r.reviewerExecution={mode:typeof meta.foreman_review_mode==='string'?meta.foreman_review_mode:'',mutationAttempted:meta.reviewer_mutation_attempted===true,validation:meta.reviewer_validation};let evidence:Record|undefined;try{evidence=this.reviewerEvidencePackage(run);}catch{evidence=undefined;}const valid=this.reviewerProvesReadOnlyEvidence(run,evidence,{terminal,responseId:r.responseId,sessionId:r.sessionId,model:{actualModel:response.model,actualModelStatus:r.actualModelStatus,requestedModel:r.requestedModel,selectedHarnessId:r.selectedHarnessId,reportedHarnessId:r.reportedHarnessId,modelFallback:r.modelFallback,cliInvocation:r.cliInvocation},mode:meta.foreman_review_mode,mutationAttempted:meta.reviewer_mutation_attempted,validation:meta.reviewer_validation},r.requestedConfig);if(!valid){r.status='failed';r.error='Reconciled Reviewer response did not prove the complete read-only evidence package';s.events.push(event('reviewer.execution_rejected','assignment',r.id,{responseId:r.responseId,sessionId:r.sessionId,actualModel:response.model,source:'reconciled-response'}));}}if(r.roleId==='planner'||r.roleId==='orchestrator'){const session=run.sessions[r.roleId];session.responseId=response.id;if(r.sessionId)session.uhpSessionId=r.sessionId;if(r.roleId==='planner')run.plannerSessionId=session.uhpSessionId;else run.orchestratorSessionId=session.uhpSessionId;}if(r.roleId==='worker'){const pendingGuidance=run.guidance.some(g=>g.status==='queued'||g.status==='delivered');run.status=terminal==='failed'?'failed':terminal==='cancelled'?(pendingGuidance?'waiting_guidance':'planning'):'review';}s.events.push(event('assignment.reconciled','assignment',r.id,{responseId:r.responseId,status:r.status,requestedModel:r.requestedModel,actualModelStatus:r.actualModelStatus,cliInvocation:r.cliInvocation}));pruneHighVolumeEvents(s.events);});}else if(a.status!=='cancel_requested'&&a.status!=='running')await this.store.mutate(s=>{const {a:r}=this.findAssignment(s,assignmentId);if(r.status==='cancel_requested'||r.status==='running'||['succeeded','failed','needs_revision','cancelled'].includes(r.status))return;r.status='running';s.events.push(event('assignment.reconciled','assignment',r.id,{responseId:r.responseId,status:response.status}));pruneHighVolumeEvents(s.events);});} private async submitExisting(projectId:string,taskId:string,runId:string,assignmentId:string,retrying=false): Promise { const snapshot=await this.store.load(); const {a}=this.findAssignment(snapshot,assignmentId); if(a.status!=='submitting'&&a.status!=='cancel_requested') return a; if(retrying&&this.discovery&&this.discovery.capabilities.idempotency!==true){await this.store.mutate(s=>{s.events.push(event('assignment.recovery_blocked','assignment',assignmentId,{reason:'UHP idempotency is unavailable; controller will not repeat the submission'}));});return a;} @@ -741,7 +763,7 @@ export class Controller { const response=await this.uhp.submit({submissionId:a.submissionId,assignmentId:a.id,runId,roleId:a.roleId,taskId,projectId,prompt,config,idempotencyKey:a.idempotencyKey,onEvent:async progress=>{if(progress.type==='response.activity'){const resp=progress.event.response;if(resp&&typeof resp==='object'&&(resp as Record).activity&&typeof (resp as Record).activity==='object'&&((resp as Record).activity as Record).kind==='tool')toolEventCount++;}let cancel:Assignment|undefined;await this.store.mutate(s=>{const {a:r,run}=this.findAssignment(s,assignmentId);if(progress.responseId){r.externalId??=progress.responseId;r.responseId??=progress.responseId;if(r.status!=='cancel_requested'&&r.status!=='cancelled')r.status='running';if(r.status==='cancel_requested')cancel=structuredClone(r);}if(progress.sessionId)r.sessionId=progress.sessionId;if(r.roleId==='planner'||r.roleId==='orchestrator'){const session=run.sessions[r.roleId];if(progress.responseId)session.responseId=progress.responseId;if(progress.sessionId){session.uhpSessionId=progress.sessionId;session.status='active';if(r.roleId==='planner')run.plannerSessionId=progress.sessionId;else run.orchestratorSessionId=progress.sessionId;}}s.events.push(event('assignment.progress','assignment',r.id,progressEventData(progress)));pruneHighVolumeEvents(s.events);});if(cancel)await this.dispatchCancel(cancel);}}); const turnTimeoutMsForDetection=turnTimeoutSecondsForSubmit*1000; const isTimeoutKill=response.cliFailureCategory==='terminated_sigterm'||(typeof response.runtimeMs==='number'&&response.runtimeMs>=turnTimeoutMsForDetection*0.95); - return this.store.mutate(s=>{const {a:r,run}=this.findAssignment(s,assignmentId); r.externalId=response.externalId; r.responseId=response.responseId; r.sessionId=response.sessionId; r.requestedModel=response.requestedModel??r.requestedConfig.model;r.actualModelStatus=response.actualModelStatus??(response.actualModel?'observed':undefined);r.selectedHarnessId=response.selectedHarnessId;r.reportedHarnessId=response.reportedHarnessId;r.modelFallback=response.modelFallback;r.cliInvocation=response.cliInvocation?structuredClone(response.cliInvocation):undefined;if(response.actualModel&&response.selectedHarnessId)r.actualConfig={harnessId:response.selectedHarnessId,model:response.actualModel};r.configOutcome=r.actualConfig?((response.modelFallback||response.actualModel!==r.requestedConfig.model||response.selectedHarnessId!==r.requestedConfig.harnessId)?'substituted':'confirmed'):'unavailable';r.configNotes={...(response.boundsApplied!==undefined?{boundsApplied:response.boundsApplied}:{}),...(response.ignoredFields?.length?{ignoredFields:[...response.ignoredFields]}:{})};if(response.usage&&Object.keys(response.usage).length)r.usage={...response.usage,measured:true,...(response.runtimeMs!==undefined?{runtimeMs:response.runtimeMs}:{})};else if(response.runtimeMs!==undefined)r.usage={runtimeMs:response.runtimeMs,measured:true};const terminal=normalizeStatus(response.status);const reviewPackage=r.roleId==='reviewer'?this.reviewerEvidencePackage(run):undefined;const reviewerInvalid=r.roleId==='reviewer'&&(terminal!=='succeeded'||!response.responseId||!response.sessionId||!reviewerModelEvidence(response,r.requestedConfig)||response.responseId===run.workerEvidence?.responseId||response.sessionId===run.assignments.find(x=>x.id===run.workerEvidence?.workerAssignmentId)?.sessionId||response.reviewerExecution?.mode!=='read_only'||response.reviewerExecution?.mutationAttempted!==false||canonicalJson(response.reviewerExecution?.validation)!==canonicalJson(reviewPackage?.controllerValidation));r.reviewerExecution=response.reviewerExecution?.mode?{mode:response.reviewerExecution.mode,mutationAttempted:response.reviewerExecution.mutationAttempted===true,validation:response.reviewerExecution.validation}:undefined;if(reviewerInvalid){r.status='failed';r.error='Reviewer response did not prove a complete, distinct, read-only review of the controller evidence package';s.events.push(event('reviewer.execution_rejected','assignment',r.id,{responseId:r.responseId,sessionId:r.sessionId,actualModel:response.actualModel,mode:response.reviewerExecution?.mode,mutationAttempted:response.reviewerExecution?.mutationAttempted}));}else if(r.status==='cancel_requested'&&['succeeded','failed','needs_revision','cancelled'].includes(terminal)){r.status=terminal;if(terminal!=='cancelled')s.events.push(event('assignment.cancel_raced_terminal','assignment',r.id,{status:terminal,source:'submit-terminal'}));}else if(r.status!=='cancelled')r.status=terminal;r.result=terminal==='failed'||terminal==='needs_revision'?response.result??response.outputText??response.output:response.output??response.outputText??response.result; + return this.store.mutate(s=>{const {a:r,run}=this.findAssignment(s,assignmentId); r.externalId=response.externalId; r.responseId=response.responseId; r.sessionId=response.sessionId; r.requestedModel=response.requestedModel??r.requestedConfig.model;r.actualModelStatus=response.actualModelStatus??(response.actualModel?'observed':undefined);r.selectedHarnessId=response.selectedHarnessId;r.reportedHarnessId=response.reportedHarnessId;r.modelFallback=response.modelFallback;r.cliInvocation=response.cliInvocation?structuredClone(response.cliInvocation):undefined;if(response.actualModel&&response.selectedHarnessId)r.actualConfig={harnessId:response.selectedHarnessId,model:response.actualModel};r.configOutcome=r.actualConfig?((response.modelFallback||response.actualModel!==r.requestedConfig.model||response.selectedHarnessId!==r.requestedConfig.harnessId)?'substituted':'confirmed'):'unavailable';r.configNotes={...(response.boundsApplied!==undefined?{boundsApplied:response.boundsApplied}:{}),...(response.ignoredFields?.length?{ignoredFields:[...response.ignoredFields]}:{})};if(response.usage&&Object.keys(response.usage).length)r.usage={...response.usage,measured:true,...(response.runtimeMs!==undefined?{runtimeMs:response.runtimeMs}:{})};else if(response.runtimeMs!==undefined)r.usage={runtimeMs:response.runtimeMs,measured:true};const terminal=normalizeStatus(response.status);const reviewPackage=r.roleId==='reviewer'?this.reviewerEvidencePackage(run):undefined;const reviewerInvalid=r.roleId==='reviewer'&&!this.reviewerProvesReadOnlyEvidence(run,reviewPackage,{terminal,responseId:response.responseId,sessionId:response.sessionId,model:response,mode:response.reviewerExecution?.mode,mutationAttempted:response.reviewerExecution?.mutationAttempted,validation:response.reviewerExecution?.validation},r.requestedConfig);r.reviewerExecution=response.reviewerExecution?.mode?{mode:response.reviewerExecution.mode,mutationAttempted:response.reviewerExecution.mutationAttempted===true,validation:response.reviewerExecution.validation}:undefined;if(reviewerInvalid){r.status='failed';r.error='Reviewer response did not prove a complete, distinct, read-only review of the controller evidence package';s.events.push(event('reviewer.execution_rejected','assignment',r.id,{responseId:r.responseId,sessionId:r.sessionId,actualModel:response.actualModel,mode:response.reviewerExecution?.mode,mutationAttempted:response.reviewerExecution?.mutationAttempted}));}else if(r.status==='cancel_requested'&&['succeeded','failed','needs_revision','cancelled'].includes(terminal)){r.status=terminal;if(terminal!=='cancelled')s.events.push(event('assignment.cancel_raced_terminal','assignment',r.id,{status:terminal,source:'submit-terminal'}));}else if(r.status!=='cancelled')r.status=terminal;r.result=terminal==='failed'||terminal==='needs_revision'?response.result??response.outputText??response.output:response.output??response.outputText??response.result; if(terminal==='succeeded'){ const resultText=typeof r.result==='string'?r.result:(typeof response.outputText==='string'?response.outputText:''); if(!resultText.trim()){ @@ -858,6 +880,8 @@ export class Controller { return inbox; } private reviewerEvidencePackage(run:Run):Record{if(!run.workerEvidence||!run.validation?.observations)throw new Error('Verified Worker evidence and controller validation observations are required');const inbox=this.currentReviewerInbox(run),reviewDiff=inbox?.reviewDiff??run.workerEvidence.reviewDiff;const _rrBytes=Buffer.byteLength(reviewDiff,'utf8');if(_rrBytes>48_000)throw Object.assign(new Error(`Verified Worker diff is ${_rrBytes.toLocaleString('en-US')} bytes across ${run.workerEvidence.changes.length} file${run.workerEvidence.changes.length===1?'':'s'}, over the 48,000 byte evidence limit; split the task into smaller tasks.`),{statusCode:413});const fullValidation=inbox?.validation??this.controllerValidationEvidence(run),controllerValidation=this.orchestratorValidationContext(fullValidation),validationEvidenceDigest=sha256(fullValidation);const evidence={validation:'verified_by_foreman_git_comparison',scopeVerified:true,baseCommit:inbox?.pinnedBaseCommit??run.workerEvidence.pinnedBaseCommit,allowedScope:[...(inbox?.allowedScope??run.workerEvidence.allowedScope)],workerResponseId:inbox?.workerResponseId??run.workerEvidence.responseId,reviewDiff,controllerValidation,validationEvidenceDigest,...(inbox?{orchestratorInboxId:inbox.id,orchestratorEvidenceDigest:inbox.evidenceDigest}: {})};if(Buffer.byteLength(JSON.stringify(evidence),'utf8')>68_000)throw Object.assign(new Error('Controller Reviewer evidence package exceeds the 68,000 byte limit'),{statusCode:413});return evidence;} + /** Single rule for whether a Reviewer response proves a complete, distinct, read-only review of the controller evidence package. `evidence` is undefined when the package could not be built, which never proves anything. */ + private reviewerProvesReadOnlyEvidence(run:Run,evidence:Record|undefined,reply:{terminal:string;responseId?:string;sessionId?:string;model:ModelReport;mode:unknown;mutationAttempted:unknown;validation:unknown},requested:RoleConfig):boolean {if(!evidence)return false;return reply.terminal==='succeeded'&&!!reply.responseId&&!!reply.sessionId&&reviewerModelEvidence(reply.model,requested)&&reply.responseId!==run.workerEvidence?.responseId&&reply.sessionId!==run.assignments.find(x=>x.id===run.workerEvidence?.workerAssignmentId)?.sessionId&&reply.mode==='read_only'&&reply.mutationAttempted===false&&canonicalJson(reply.validation)===canonicalJson(evidence.controllerValidation);} private reviewerAssignmentMatchesEvidence(run:Run,reviewer:Assignment):boolean{try{const evidence=this.reviewerEvidencePackage(run);return canonicalJson(reviewer.reviewerExecution?.validation)===canonicalJson(evidence.controllerValidation)&&(typeof evidence.orchestratorInboxId!=='string'||reviewer.reviewerInboxId===evidence.orchestratorInboxId&&reviewer.reviewerEvidenceDigest===evidence.orchestratorEvidenceDigest);}catch{return false;}} private promotionEvidenceDigest(run:Run):string { if(!run.workerEvidence||!run.validation||!run.reviewerRecommendation)throw Object.assign(new Error('Verified Worker, validation, and Reviewer evidence are required'),{statusCode:409}); @@ -1307,6 +1331,9 @@ export function workerProposalScopeIssue( } return undefined; } +export type RecoveryStatus={status:'idle'|'running'|'failed';startedAt?:string;finishedAt?:string;lastError?:string;lastErrorAt?:string}; +/** One short line for logs, events and the UI: first line only, credentials masked, control characters removed, capped at 300 characters. */ +function safeRecoveryMessage(error:unknown):string {const raw=error instanceof Error?error.message:String(error);return raw.split(/\r?\n/)[0]!.replace(/[\u0000-\u001f\u007f]/g,' ').replace(/\b(Bearer|Basic)\s+[^\s,;]+/gi,'$1 [redacted]').replace(/((?:token|key|secret|password|authorization)["']?\s*[:=]\s*["']?)[^\s"',;&]+/gi,'$1[redacted]').replace(/\b(?:sk|pk|ghp|gho|xox[a-z])-?[A-Za-z0-9_-]{16,}/g,'[redacted]').trim().slice(0,300)||'Recovery failed';} const usageFields=['inputTokens','outputTokens','totalTokens','thinkingTokens','cachedInputTokens','runtimeMs','requestCount'] as const; function aggregate(values:Array):Usage|undefined {const measured=values.filter((x):x is Usage=>!!x&&Object.keys(x).some(k=>usageFields.includes(k as typeof usageFields[number])));if(!measured.length)return;const result:Usage={measured:true};for(const field of usageFields){const numbers=values.map(x=>x?.[field]);if(numbers.length&&numbers.every((x):x is number=>typeof x==='number'&&Number.isFinite(x)))result[field]=numbers.reduce((a,b)=>a+b,0);}return Object.keys(result).length>1?result:undefined;} function cliInvocationFrom(value:unknown):{executable:string;hostExecutable?:string;args:string[]}|undefined {if(!value||typeof value!=='object')return;const item=value as Record;if(typeof item.executable!=='string'||!Array.isArray(item.args)||!item.args.every(arg=>typeof arg==='string'))return;return {executable:item.executable,...(typeof item.host_executable==='string'?{hostExecutable:item.host_executable}:{}),args:[...item.args] as string[]};} diff --git a/src/server.ts b/src/server.ts index e695860..97f18f5 100644 --- a/src/server.ts +++ b/src/server.ts @@ -137,7 +137,7 @@ const server=createServer(async(req,res)=>{ projectControllers.set(existingId,{...runtime,workspace:saved}); }else{ const {scoped,bridge}=await createProjectRuntime(existingId,workspace); - try{await scoped.refreshDiscovery();await scoped.recover();const saved=await saveWorkspaceSetup(config.dataDir,existingId,{repoPath:workspace.repoPath,allowedScope:workspace.allowedScope,validationCommands:workspace.validationCommands});projectControllers.set(existingId,{controller:scoped,bridge,workspace:saved});configuredProjectIds.add(existingId);} + try{await scoped.refreshDiscovery();const saved=await saveWorkspaceSetup(config.dataDir,existingId,{repoPath:workspace.repoPath,allowedScope:workspace.allowedScope,validationCommands:workspace.validationCommands});projectControllers.set(existingId,{controller:scoped,bridge,workspace:saved});configuredProjectIds.add(existingId);void scoped.startRecovery(existingId);} catch(error){projectControllers.delete(existingId);await bridge.stop();throw error;} } const resumed=await projectControllers.get(existingId)!.controller.state();const project=resumed.projects.find(item=>item.id===existingId);if(!project)throw new Error('Saved project disappeared while reopening its repository');json(res,200,project);return; @@ -148,7 +148,7 @@ const server=createServer(async(req,res)=>{ catch(error){projectControllers.delete(projectId);await bridge.stop();throw error;} } const setupMatch=path.match(/^\/api\/projects\/([^/]+)\/workspace-setup$/); - if(req.method==='GET'&&setupMatch){const projectId=decodeURIComponent(setupMatch[1]!);let workspace;try{workspace=await loadWorkspaceSetup(config.dataDir,projectId);}catch{throw Object.assign(new Error('Saved repository is unavailable; reopen the repository to continue'),{statusCode:503});}if(!workspace)throw Object.assign(new Error('Workspace setup not found'),{statusCode:404});const bridgeHealth=projectControllers.get(projectId)?.bridge.health;json(res,200,{...workspace,bridgeStatus:bridgeHealth?.state??'unavailable',...(bridgeHealth?{bridgeHealth}:{})});return;} + if(req.method==='GET'&&setupMatch){const projectId=decodeURIComponent(setupMatch[1]!);let workspace;try{workspace=await loadWorkspaceSetup(config.dataDir,projectId);}catch{throw Object.assign(new Error('Saved repository is unavailable; reopen the repository to continue'),{statusCode:503});}if(!workspace)throw Object.assign(new Error('Workspace setup not found'),{statusCode:404});const runtime=projectControllers.get(projectId),bridgeHealth=runtime?.bridge.health;json(res,200,{...workspace,bridgeStatus:bridgeHealth?.state??'unavailable',...(bridgeHealth?{bridgeHealth}:{}),...(runtime?{recovery:runtime.controller.recoveryStatus()}:{})});return;} if(req.method==='GET'&&path==='/api/state'){json(res,200,stateForUi(await activeController.state()));return;} if(req.method==='GET'&&path==='/api/status'){json(res,200,await activeController.serviceStatus());return;} if(req.method==='GET'&&path==='/api/events'){ @@ -237,12 +237,13 @@ await controller.refreshDiscovery(); for(const project of startupState.projects){ let workspace;try{workspace=await loadWorkspaceSetup(config.dataDir,project.id);}catch(error){process.stderr.write(`Saved workspace for ${project.id} is unavailable: ${error instanceof Error?error.message:'invalid setup'}\n`);continue;} if(!workspace)continue;configuredProjectIds.add(project.id); - try{const {scoped}=await createProjectRuntime(project.id,workspace);await scoped.refreshDiscovery();await scoped.recover();} + try{const {scoped}=await createProjectRuntime(project.id,workspace);await scoped.refreshDiscovery();void scoped.startRecovery(project.id);} catch(error){process.stderr.write(`Could not start local workspace for ${project.id}: ${error instanceof Error?error.message:'bridge startup failed'}\n`);} } -if(configuredProjectIds.size===0)await controller.recover(); +if(configuredProjectIds.size===0)void controller.startRecovery('default project'); setInterval(()=>{if(configuredProjectIds.size===0)void controller.reconcileRunning();for(const runtime of projectControllers.values())void runtime.controller.reconcileRunning();},5000).unref(); -const shutdown=async()=>{server.close();await Promise.all([...projectControllers.values()].map(runtime=>runtime.bridge.stop()));await store.flush().catch(()=>undefined);}; +// Shutdown never awaits an in-flight recovery (it may be blocked on an external UHP call); the flush is bounded too. +const shutdown=async()=>{server.close();await Promise.all([...projectControllers.values()].map(runtime=>runtime.bridge.stop()));await Promise.race([store.flush().catch(()=>undefined),new Promise(done=>setTimeout(done,5000).unref())]);}; process.once('SIGINT',()=>{void shutdown().finally(()=>process.exit(0));}); process.once('SIGTERM',()=>{void shutdown().finally(()=>process.exit(0));}); server.listen(config.port,config.host,()=>process.stdout.write(`Foreman listening on http://${config.host}:${config.port}\n`)); diff --git a/tests/recovery.test.ts b/tests/recovery.test.ts new file mode 100644 index 0000000..9363470 --- /dev/null +++ b/tests/recovery.test.ts @@ -0,0 +1,114 @@ +import { mkdtemp, rm } from 'node:fs/promises'; +import { tmpdir } from 'node:os'; +import { join } from 'node:path'; +import { afterEach, describe, expect, it, vi } from 'vitest'; +import { Controller, type UhpAdapter } from '../src/controller.js'; +import type { Assignment } from '../src/domain.js'; +import { JsonStore } from '../src/store.js'; + +const dirs:string[]=[]; +const unhandled:unknown[]=[]; +const onUnhandled=(reason:unknown)=>{unhandled.push(reason);}; +process.on('unhandledRejection',onUnhandled); +afterEach(async()=>{vi.restoreAllMocks();await Promise.all(dirs.splice(0).map(d=>rm(d,{recursive:true,force:true})));}); +const cfg={harnessId:'fixture',model:'model-fixture'}; +const tick=(ms=20)=>new Promise(resolve=>setTimeout(resolve,ms)); +const pending=async(p:Promise)=>(await Promise.race([p.then(()=>'settled'),tick(60).then(()=>'pending')]))==='pending'; +const assignment=(id:string,status:Assignment['status'],extra:Partial={}):Assignment=>({id,roleId:'worker',status,requestedConfig:cfg,prompt:'work',submissionId:`sub-${id}`,idempotencyKey:`idem-${id}`,createdAt:new Date().toISOString(),...extra}); + +/** A persisted store holding one run, plus a gate that releases every hung UHP call. */ +async function seed(assignments:Assignment[]) { + const dir=await mkdtemp(join(tmpdir(),'foreman-'));dirs.push(dir);const path=join(dir,'state.json'),store=new JsonStore(path); + const first=new Controller(store,{submit:async()=>({externalId:'x',status:'completed'}),cancel:async()=>({status:'cancelled'})}); + const project:any=await first.createProject('Recovery'),task:any=await first.createTask(project.id,'Task'),run:any=await first.createRun(task.id); + await store.mutate(s=>{s.projects[0]!.tasks[0]!.runs[0]!.assignments.push(...assignments);}); + let release:()=>void=()=>undefined;const gate=new Promise(resolve=>{release=resolve;}); + const retrieved:string[]=[],submitted:string[]=[]; + const uhp:UhpAdapter={ + submit:async input=>{submitted.push(input.assignmentId);await gate;return {externalId:'ext',responseId:'resp',status:'completed'};}, + retrieve:async responseId=>{retrieved.push(responseId);await gate;return {id:responseId,status:'completed',output_text:'done'} as any;}, + cancel:async()=>({status:'cancelled'}), + }; + const restarted=new Controller(new JsonStore(path),uhp); + return {controller:restarted,store:restarted.store,run,release,retrieved,submitted,gate}; +} +const stateOf=async(controller:Controller)=>(await controller.store.load()); + +describe('non-blocking recovery',()=>{ + it('reports running while a UHP retrieve hangs, and startRecovery returns immediately',async()=>{ + const {controller,retrieved,release}=await seed([assignment('a1','running',{responseId:'resp-1',externalId:'resp-1'})]); + expect(controller.recoveryStatus()).toEqual({status:'idle'}); + const recovery=controller.startRecovery('test');await tick(); + expect(controller.recoveryStatus()).toMatchObject({status:'running',startedAt:expect.any(String)}); + expect(retrieved).toEqual(['resp-1']);expect(await pending(recovery)).toBe(true); + expect((await controller.serviceStatus()).recovery.status).toBe('running'); + release();await recovery; + expect(controller.recoveryStatus()).toMatchObject({status:'idle',finishedAt:expect.any(String)}); + expect((await stateOf(controller)).projects[0]!.tasks[0]!.runs[0]!.assignments[0]!.status).toBe('succeeded'); + }); + it('reports running while a UHP submit replay hangs',async()=>{ + const {controller,submitted,release}=await seed([assignment('a1','submitting')]); + const recovery=controller.startRecovery();await tick(); + expect(submitted).toEqual(['a1']);expect(controller.recoveryStatus().status).toBe('running');expect(await pending(recovery)).toBe(true); + release();await recovery;expect(controller.recoveryStatus().status).toBe('idle'); + }); + it('never runs twice concurrently for one controller',async()=>{ + const {controller,retrieved,release}=await seed([assignment('a1','running',{responseId:'resp-1',externalId:'resp-1'})]); + const first=controller.startRecovery(),second=controller.startRecovery(),direct=controller.recover();direct.catch(()=>undefined);await tick(); + expect(controller.recover()).toBe(direct);expect(retrieved).toEqual(['resp-1']); + release();await Promise.all([first,second,direct]);expect(retrieved).toEqual(['resp-1']); + // Once settled a later call starts a fresh recovery. + const fresh=controller.recover();expect(fresh).not.toBe(direct);await fresh; + }); + it('records a rejected recovery as failed with a redacted event, logs to stderr, and never leaks an unhandled rejection',async()=>{ + const {controller}=await seed([]);const stderr=vi.spyOn(process.stderr,'write').mockImplementation(()=>true); + vi.spyOn(controller as any,'migrateProjectRoleConfigs').mockRejectedValue(new Error('boom Authorization: Bearer sk-abcdefghijklmnop1234 token=hunter2\nsecond line')); + await controller.startRecovery('project-x');await tick(30); + const status=controller.recoveryStatus(); + expect(status).toMatchObject({status:'failed',lastError:expect.stringContaining('boom'),lastErrorAt:expect.any(String),finishedAt:expect.any(String)}); + expect(JSON.stringify(status)).not.toMatch(/hunter2|abcdefghijklmnop|second line/); + const events=(await stateOf(controller)).events.filter(e=>e.type==='recovery.failed'); + expect(events).toHaveLength(1);expect(JSON.stringify(events[0]!.data)).not.toMatch(/hunter2|abcdefghijklmnop/); + const logged=stderr.mock.calls.map(c=>String(c[0])).join('');expect(logged).toContain('Recovery failed for project-x');expect(logged).not.toMatch(/hunter2|abcdefghijklmnop/); + expect(unhandled).toEqual([]); + // A later successful recovery clears the failure. + (controller as any).migrateProjectRoleConfigs.mockRestore();await controller.startRecovery(); + expect(controller.recoveryStatus().status).toBe('idle');expect(controller.recoveryStatus().lastError).toBeUndefined(); + }); + it('recover() itself still rejects for direct callers',async()=>{ + const {controller}=await seed([]);vi.spyOn(controller as any,'resumePendingRetentions').mockRejectedValue(new Error('direct')); + await expect(controller.recover()).rejects.toThrow('direct');expect(controller.recoveryStatus().status).toBe('failed');expect(unhandled).toEqual([]); + }); + it('reconcileRunning skips assignments recovery still owns but reconciles new ones',async()=>{ + const {controller,store,retrieved,release}=await seed([assignment('a1','running',{responseId:'resp-1',externalId:'resp-1'})]); + const recovery=controller.startRecovery();await tick();expect(retrieved).toEqual(['resp-1']); + await store.mutate(s=>{s.projects[0]!.tasks[0]!.runs[0]!.assignments.push(assignment('a2','running',{responseId:'resp-2',externalId:'resp-2'}));}); + const reconciling=controller.reconcileRunning();await tick(); + // a1 is owned by recovery (no second retrieve); a2 was created after recovery started and is reconciled normally (its retrieve is gated too). + expect(retrieved).toEqual(['resp-1','resp-2']); + release();await Promise.all([recovery,reconciling]); + const assignments=(await stateOf(controller)).projects[0]!.tasks[0]!.runs[0]!.assignments; + expect(assignments.map(a=>a.status)).toEqual(['succeeded','succeeded']); + expect((await stateOf(controller)).events.filter(e=>e.type==='assignment.reconciled'&&e.entityId==='a1')).toHaveLength(1); + // After recovery settles the loop covers everything again. + await store.mutate(s=>{const a=s.projects[0]!.tasks[0]!.runs[0]!.assignments[0]!;a.status='running';});await controller.reconcileRunning();expect(retrieved.filter(r=>r==='resp-1')).toHaveLength(2); + }); + it('does not stop a run that became active after recovery began',async()=>{ + const {controller,store,run,release,retrieved}=await seed([assignment('a1','running',{responseId:'resp-1',externalId:'resp-1'})]); + controller.configureVerifiedWorkspace({repoPath:'/fixture/repo',bridgeBaseUrl:'http://127.0.0.1:1',allowedScope:['README.md'],commands:[{name:'check',command:'true',args:[]}]}); + const other:any=await controller.createRun((await stateOf(controller)).projects[0]!.tasks[0]!.id); + await store.mutate(s=>{const runs=s.projects[0]!.tasks[0]!.runs;runs.find(r=>r.id===run.id)!.controller={startedAt:new Date().toISOString(),active:true,phase:'working',budgets:{roleTurns:{planner:1,orchestrator:1,worker:1,reviewer:1},workerAttempts:1}} as any;}); + const recovery=controller.startRecovery();await tick();expect(retrieved).toEqual(['resp-1']); + await store.mutate(s=>{s.projects[0]!.tasks[0]!.runs.find(r=>r.id===other.id)!.controller={startedAt:new Date().toISOString(),active:true,phase:'working',budgets:{roleTurns:{planner:1,orchestrator:1,worker:1,reviewer:1},workerAttempts:1}} as any;}); + release();await recovery; + const runs=(await stateOf(controller)).projects[0]!.tasks[0]!.runs; + expect(runs.find(r=>r.id===run.id)!.controller?.active).toBe(false); + expect(runs.find(r=>r.id===other.id)!.controller?.active).toBe(true); + }); + it('does not overwrite a terminal assignment with a stale running observation',async()=>{ + const {controller,store}=await seed([assignment('a1','submitting',{responseId:'resp-1',externalId:'resp-1'})]); + (controller.uhp as any).retrieve=async()=>{await store.mutate(s=>{s.projects[0]!.tasks[0]!.runs[0]!.assignments[0]!.status='succeeded';});return {id:'resp-1',status:'in_progress'};}; + await controller.refreshAssignment('a1'); + expect((await stateOf(controller)).projects[0]!.tasks[0]!.runs[0]!.assignments[0]!.status).toBe('succeeded'); + }); +}); diff --git a/tests/reviewer-evidence.test.ts b/tests/reviewer-evidence.test.ts new file mode 100644 index 0000000..ba1bac8 --- /dev/null +++ b/tests/reviewer-evidence.test.ts @@ -0,0 +1,57 @@ +import { mkdtemp, rm } from 'node:fs/promises'; +import { tmpdir } from 'node:os'; +import { join } from 'node:path'; +import { afterEach, describe, expect, it } from 'vitest'; +import { Controller, type UhpAdapter } from '../src/controller.js'; +import type { Assignment } from '../src/domain.js'; +import { JsonStore } from '../src/store.js'; + +const dirs:string[]=[]; +afterEach(async()=>{await Promise.all(dirs.splice(0).map(d=>rm(d,{recursive:true,force:true})));}); +const cfg={harnessId:'fixture',model:'model-fixture'}; + +// Seeds a run holding verified Worker evidence and passed validation, so the controller can build the Reviewer evidence package. +async function seed(adapter:Partial={}) { + const dir=await mkdtemp(join(tmpdir(),'foreman-'));dirs.push(dir);const store=new JsonStore(join(dir,'state.json')); + await store.mutate(s=>{for(const role of s.roles){role.enabled=true;role.availableConfigs=[cfg];role.config=cfg;}}); + const uhp:UhpAdapter={submit:async()=>({externalId:'x',status:'completed'}),cancel:async()=>({status:'cancelled'}),...adapter}; + const controller=new Controller(store,uhp); + controller.configureVerifiedWorkspace({repoPath:'/fixture/repo',bridgeBaseUrl:'http://127.0.0.1:1',allowedScope:['README.md'],commands:[{name:'fixture check',command:'true',args:[]}]}); + const project:any=await controller.createProject('Reviewer evidence'),task:any=await controller.createTask(project.id,'Task'),run:any=await controller.createRun(task.id); + await store.mutate(s=>{const current=s.projects[0]!.tasks[0]!.runs[0]!;current.pinnedBaseCommit='a'.repeat(40);current.workspaceId='ws'; + const worker:Assignment={id:'worker-1',roleId:'worker',status:'succeeded',requestedConfig:cfg,responseId:'worker-response',externalId:'worker-response',sessionId:'worker-session',prompt:'work',submissionId:'sub-w',idempotencyKey:'idem-w',createdAt:new Date().toISOString()}; + current.assignments.push(worker); + current.workerEvidence={provenance:'recorded_replay',workerAssignmentId:worker.id,responseId:worker.responseId!,pinnedBaseCommit:current.pinnedBaseCommit,completeSnapshot:{reportedComplete:true,reportedErrors:0,entryCount:1},scopeVerified:true,allowedScope:['README.md'],entries:[],changes:[{path:'README.md',kind:'modified',summary:'note'}],reviewDiff:'diff --git a/README.md b/README.md\n+note',acceptance:'not_decided'} as any; + current.validation={id:'v1',status:'passed',passed:true,reportedPassed:true,checks:[{name:'fixture check',passed:true}],observations:[{name:'fixture check',command:'true',args:[],exitCode:0,timedOut:false,output:'passed',outputTruncated:false,passed:true}],policy:{requireAllChecksPass:true,configuredCheckCount:1},gitEvidence:{status:'verified',commit:current.pinnedBaseCommit,changedPaths:['README.md'],submittedAt:new Date().toISOString()},createdAt:new Date().toISOString()} as any;}); + const validation=(controller as any).reviewerEvidencePackage((await store.load()).projects[0]!.tasks[0]!.runs[0]!).controllerValidation; + return {store,controller,uhp,run,validation}; +} +type Reply={terminal:string;sessionId?:string;responseId:string;mode:string;mutationAttempted:boolean;validation:unknown;model:string}; +const good=(validation:unknown):Reply=>({terminal:'completed',sessionId:'reviewer-session',responseId:'rev-response',mode:'read_only',mutationAttempted:false,validation,model:'model-fixture'}); +const cases:Array<{name:string;patch:(validation:unknown)=>Partial}>=[ + {name:'wrong mode',patch:()=>({mode:'write'})}, + {name:'mutation attempted',patch:()=>({mutationAttempted:true})}, + {name:'same session as the Worker',patch:()=>({sessionId:'worker-session'})}, + {name:'same response as the Worker',patch:()=>({responseId:'worker-response'})}, + {name:'validation mismatch',patch:()=>({validation:{different:true}})}, + {name:'model mismatch',patch:()=>({model:'other-model'})}, + {name:'non-succeeded terminal status',patch:()=>({terminal:'failed'})}, + {name:'missing session',patch:()=>({sessionId:undefined})}, +]; +const live=(o:Reply)=>({externalId:o.responseId,responseId:o.responseId,sessionId:o.sessionId,status:o.terminal,outputText:'review',actualModel:o.model,requestedModel:'model-fixture',selectedHarnessId:'fixture',reviewerExecution:{mode:o.mode,mutationAttempted:o.mutationAttempted,validation:o.validation}}); +const retrieved=(o:Reply)=>({id:o.responseId,status:o.terminal,output_text:'review',model:o.model,requested_model:'model-fixture',session_id:o.sessionId,metadata:{harness_id:'fixture',foreman_review_mode:o.mode,reviewer_mutation_attempted:o.mutationAttempted,reviewer_validation:o.validation}}); + +describe('Reviewer read-only evidence rule',()=>{ + describe('live submit path',()=>{ + async function run(patch:(validation:unknown)=>Partial){let response:any;const s=await seed({submit:async()=>response});response=live({...good(s.validation),...patch(s.validation)});const a=await s.controller.assign(s.run.id,'reviewer',(s.controller as any).reviewerPrompt());return {a,events:(await s.store.load()).events.filter(e=>e.type==='reviewer.execution_rejected')};} + it('accepts a complete, distinct, read-only response',async()=>{const {a,events}=await run(()=>({}));expect(a.status).toBe('succeeded');expect(events).toHaveLength(0);}); + for(const c of cases)it(`rejects ${c.name}`,async()=>{const {a,events}=await run(c.patch);expect(a.status).toBe('failed');expect(a.error).toBe('Reviewer response did not prove a complete, distinct, read-only review of the controller evidence package');expect(events).toHaveLength(1);expect(events[0]!.data).toHaveProperty('actualModel');}); + }); + describe('reconciled path',()=>{ + async function run(patch:(validation:unknown)=>Partial){const s=await seed();let response:any;(s.uhp as any).retrieve=async()=>response;response={...retrieved({...good(s.validation),...patch(s.validation)})}; + await s.store.mutate(st=>{st.projects[0]!.tasks[0]!.runs[0]!.assignments.push({id:'rev-1',roleId:'reviewer',status:'running',requestedConfig:cfg,responseId:'rev-response',externalId:'rev-response',prompt:'review',submissionId:'sub-r',idempotencyKey:'idem-r',createdAt:new Date().toISOString()});}); + const a:any=await s.controller.refreshAssignment('rev-1');return {a,events:(await s.store.load()).events.filter(e=>e.type==='reviewer.execution_rejected')};} + it('accepts a complete, distinct, read-only response',async()=>{const {a,events}=await run(()=>({}));expect(a.status).toBe('succeeded');expect(a.error).toBeUndefined();expect(events).toHaveLength(0);}); + for(const c of cases)it(`rejects ${c.name}`,async()=>{const {a,events}=await run(c.patch);expect(a.status).toBe('failed');expect(a.error).toBe('Reconciled Reviewer response did not prove the complete read-only evidence package');expect(events).toHaveLength(1);expect(events[0]!.data).toMatchObject({source:'reconciled-response',actualModel:expect.anything()});}); + }); +}); From 6d16080b6f8c923f6cd7c1878c584eb9027726ba Mon Sep 17 00:00:00 2001 From: Claude Date: Mon, 28 Sep 2026 22:11:07 +0000 Subject: [PATCH 3/3] Seed bridge workspaces in one git process and clean them up Seeding ran two git cat-file processes per file, and bridge workspaces were never deleted. - seedWorkspace streams every blob through one `git cat-file --batch` process, parsed incrementally from raw buffers with backpressure. All existing limits and error messages are kept; blob size is checked from the record header before the body is buffered, and the child has a timeout. On a 3000-file repository seeding took a median 2.7 s instead of 24.2 s. - DELETE /extensions/foreman-workspace/v1/workspaces/:id removes a workspace and its sibling dirs: 204, 404 for an unknown id, 409 while a response is running in it. - At startup the bridge removes bridge-named directories under its work root that no persisted state references. - Idle workspaces are swept after LOCAL_CLI_UHP_WORKSPACE_TTL_MS (default 24 h, 0 disables); one with a running response is never removed. - Removal resolves paths and never follows a symlink out of the work root. State writes are serialized so concurrent persists cannot race. Co-Authored-By: Claude Opus 5.5 Claude-Session: https://claude.ai/code/session_01DXCvrGWmmRNDLmy7nYS2QP --- investigations/local-cli-uhp/README.md | 15 +++ investigations/local-cli-uhp/server.mjs | 163 ++++++++++++++++++++---- investigations/local-cli-uhp/test.mjs | 134 ++++++++++++++++++- 3 files changed, 286 insertions(+), 26 deletions(-) diff --git a/investigations/local-cli-uhp/README.md b/investigations/local-cli-uhp/README.md index a948a5f..0928509 100644 --- a/investigations/local-cli-uhp/README.md +++ b/investigations/local-cli-uhp/README.md @@ -163,7 +163,22 @@ operation over a full commit SHA; requests do not accept a host path: - `POST /extensions/foreman-workspace/v1/workspaces` with `{"base_commit":""}` seeds a fresh workspace from the configured source repository and returns its `workspace_id` and exact base. + Blobs are read with a single `git cat-file --batch` process (16 MiB per blob, + 256 MiB total; a missing or oversized blob fails the seed and removes the + partial workspace). - Submit the Worker request with that `workspace_id` in response metadata. +- `DELETE /extensions/foreman-workspace/v1/workspaces/{workspace_id}` removes + the workspace and its per-workspace directories (`204`; `404` for an unknown + id; `409` while a response is running in it or uses it as a read-only + workspace). +- Cleanup is also automatic. At startup the bridge removes bridge-named entries + under `LOCAL_CLI_UHP_WORK` that no persisted state references (workspaces, + Reviewer scratch, preflight dirs, role-session state no stored response + points at); unrecognised files are left alone. Every 1/4 of + `LOCAL_CLI_UHP_WORKSPACE_TTL_MS` (default 24 h, `0` disables; checked at most + every 10 minutes) an unref'd timer removes workspaces with no running + response whose last use (seed, overlay, snapshot, submit, finish) is older + than the TTL. Removal never follows a symlink out of the work root. - `GET /extensions/foreman-workspace/v1/workspaces/{workspace_id}/snapshot` returns a complete tree, exact file bytes (base64) and SHA-256, Git mode, symlink target, and an explicit `errors` list. Dotfiles and instruction diff --git a/investigations/local-cli-uhp/server.mjs b/investigations/local-cli-uhp/server.mjs index 7ce18d4..1ee2877 100644 --- a/investigations/local-cli-uhp/server.mjs +++ b/investigations/local-cli-uhp/server.mjs @@ -11,6 +11,7 @@ import { tmpdir, homedir } from 'node:os'; import { join, resolve, dirname } from 'node:path'; import { posix } from 'node:path'; import { TextDecoder } from 'node:util'; +import { once } from 'node:events'; import { createRequire } from 'node:module'; import { roleSessionBinding, roleStatePath, resolvePreviousRoleSession, withRoleSession } from './role-sessions.mjs'; import { claudeCodeCliArgs, codexCliArgs } from './cli-args.mjs'; @@ -51,6 +52,7 @@ const HARNESS = { codex: { id: 'codex-cli', bin: process.env.CODEX_BIN ?? 'codex', authDir: process.env.CODEX_HOME ?? join(homedir(), '.codex'), model: process.env.CODEX_MODEL ?? 'gpt-6-sol' }, agy: { id: 'antigravity-cli', bin: process.env.AGY_BIN ?? 'agy', authDir: process.env.AGY_CONFIG_DIR ?? join(homedir(), '.gemini', 'antigravity-cli'), model: process.env.AGY_MODEL ?? 'gemini-3.8-flash-low' }, }; +const WORKSPACE_TTL_MS = (() => { const value = Number(process.env.LOCAL_CLI_UHP_WORKSPACE_TTL_MS); return process.env.LOCAL_CLI_UHP_WORKSPACE_TTL_MS !== undefined && process.env.LOCAL_CLI_UHP_WORKSPACE_TTL_MS !== '' && Number.isFinite(value) && value >= 0 ? value : 24 * 3600_000; })(); const tasks = new Map(); const workspaces = new Map(); let agyModelsCache = { expires: 0, models: [] }; @@ -63,7 +65,9 @@ function agyDiscoveryEnvironment(source = process.env) { return Object.fromEntries(Object.entries(source).filter(([name, value]) => allowed.has(name) && typeof value === 'string')); } -async function persist() { +let persistQueue = Promise.resolve(); +function persist() { const run = persistQueue.then(writeState); persistQueue = run.catch(() => undefined); return run; } +async function writeState() { await mkdir(resolve(STATE, '..'), { recursive: true }); const temp = `${STATE}.${process.pid}.tmp`; await writeFile(temp, JSON.stringify(state), { mode: 0o600 }); @@ -75,12 +79,65 @@ async function load() { for (const [id, item] of Object.entries(state.workspaces)) { if (!/^ws_[0-9a-f-]{36}$/.test(id) || !/^(?:[0-9a-f]{40}|[0-9a-f]{64})$/.test(item.baseCommit) || !SOURCE_REPO) continue; const dir = join(ROOT, id); - try { if ((await lstat(dir)).isDirectory()) workspaces.set(id, { id, baseCommit: item.baseCommit, dir, sourceRepo: await realpath(SOURCE_REPO), responseId: item.responseId }); } catch {} + try { if ((await lstat(dir)).isDirectory()) workspaces.set(id, { id, baseCommit: item.baseCommit, dir, sourceRepo: await realpath(SOURCE_REPO), responseId: item.responseId, lastUsed: Number.isFinite(item.lastUsed) ? item.lastUsed : Date.now() }); } catch {} } for (const [id, r] of Object.entries(state.responses)) { - if (r.status === 'in_progress') { r.status = 'failed'; r.error = { message: 'Server restarted before CLI completion; replay will not spawn a duplicate' }; await persist(); } + if (r.status === 'in_progress') { touchWorkspace(workspaces.get(r.metadata?.workspace_id)); touchWorkspace(workspaces.get(r.metadata?.foreman_read_only_workspace_id)); r.status = 'failed'; r.error = { message: 'Server restarted before CLI completion; replay will not spawn a duplicate' }; await persist(); } tasks.set(id, { response: r }); } + await sweepOrphans(); +} +// Everything below only ever removes entries of ROOT (never follows a symlink out of it), and only names this bridge creates. +const ROOT_ENTRY = /^(?:ws_[0-9a-f-]{36}(?:\..+)?|review-[0-9a-f-]{36}|resp_[0-9a-f-]{36}\.review-codex-home|(?:preflight|codex-preflight|agy-preflight|planner-auth-preflight)-[0-9a-f-]{36})$/; +async function removeInsideRoot(path) { + const root = await realpath(ROOT); + let info; + try { info = await lstat(path); } catch (e) { if (e.code === 'ENOENT') return false; throw e; } + if (info.isSymbolicLink()) { await rm(path, { force: true }); return true; } // the link itself, never its target + const real = await realpath(path); + if (dirname(real) === real || !real.startsWith(`${root}/`)) throw Error('Refusing to remove a path outside the work root'); + if (info.isDirectory()) await chmod(real, 0o700).catch(() => undefined); + await rm(real, { recursive: true, force: true }); + return true; +} +function referencedRoleStates() { + const paths = new Set(); + for (const r of Object.values(state.responses)) for (const p of [r.metadata?.role_session_state_path, r.metadata?.role_session?.state_path]) if (typeof p === 'string') paths.add(resolve(p)); + return paths; +} +async function sweepOrphans() { + if (ROOT === tmpdir()) return; + const keepWorkspace = name => Object.keys(state.workspaces).some(id => name === id || name.startsWith(`${id}.`)); + for (const name of await readdir(ROOT).catch(() => [])) { + if (ROOT_ENTRY.test(name) && !keepWorkspace(name)) await removeInsideRoot(join(ROOT, name)).catch(() => undefined); + } + const roles = join(ROOT, 'role-sessions'), keep = referencedRoleStates(); + for (const name of await readdir(roles).catch(() => [])) if (!keep.has(join(roles, name))) await removeInsideRoot(join(roles, name)).catch(() => undefined); + let changed = false; + for (const id of Object.keys(state.workspaces)) if (!(await lstat(join(ROOT, id)).catch(() => undefined))) { delete state.workspaces[id]; changed = true; } + if (changed) await persist(); +} +function touchWorkspace(ws) { if (!ws) return; ws.lastUsed = Date.now(); if (state.workspaces[ws.id]) state.workspaces[ws.id].lastUsed = ws.lastUsed; } +function workspaceActive(ws) { + return ws.ops > 0 || Object.values(state.responses).some(r => r.status === 'in_progress' && (r.metadata?.workspace_id === ws.id || r.metadata?.foreman_read_only_workspace_id === ws.id)); +} +// Runs one workspace request (overlay, snapshot) so cleanup cannot remove the workspace underneath it. +async function withWorkspaceOp(ws, fn) { ws.ops = (ws.ops ?? 0) + 1; try { return await fn(); } finally { ws.ops--; touchWorkspace(ws); persist().catch(() => undefined); } } +// The synchronous part (activity check plus unregistering) is what stops a new response from binding to a workspace being removed. +async function removeWorkspace(id) { + const ws = workspaces.get(id); + if (!ws || workspaceActive(ws)) return false; + workspaces.delete(id); delete state.workspaces[id]; + await persist(); + for (const name of await readdir(ROOT).catch(() => [])) if (name === id || name.startsWith(`${id}.`)) await removeInsideRoot(join(ROOT, name)).catch(error => console.warn(`workspace cleanup failed for ${name}: ${error.message}`)); + return true; +} +async function sweepIdleWorkspaces() { + if (!WORKSPACE_TTL_MS) return; + for (const ws of [...workspaces.values()]) { + if (workspaceActive(ws)) { touchWorkspace(ws); continue; } + if (Date.now() - (ws.lastUsed ?? 0) > WORKSPACE_TTL_MS) await removeWorkspace(ws.id).catch(error => console.warn(`workspace sweep failed for ${ws.id}: ${error.message}`)); + } } function send(res, status, body, extra = {}) { res.writeHead(status, { 'content-type': 'application/json', 'UHP-Version': VERSION, ...extra }); res.end(JSON.stringify(body)); @@ -479,6 +536,60 @@ function git(repo, args, maxBytes = 32 * 1024 * 1024) { child.once('close', code => code === 0 ? resolvePromise(Buffer.concat(out, size)) : reject(Error(`git ${args[0]} failed (${code}): ${err.slice(0, 1000)}`))); }); } +const GIT_BATCH_TIMEOUT_MS = 300_000; +const MAX_BLOB = 16 * 1024 * 1024, MAX_SEED_TOTAL = 256 * 1024 * 1024; +// One `git cat-file --batch` for every blob: oids in on stdin, ` \n\n` records out. Both callbacks +// run in order and are awaited before the next chunk is read, which is the backpressure on git's stdout. +async function gitBatchBlobs(repo, entries, onHeader, onBlob) { + const child = spawn('git', ['-C', repo, 'cat-file', '--batch'], { stdio: ['pipe', 'pipe', 'pipe'], shell: false }); + let err = ''; let timedOut = false; const abort = new AbortController(); + child.stderr.on('data', chunk => { if (err.length < 4000) err += chunk.toString('utf8').slice(0, 4000 - err.length); }); + child.stdin.on('error', () => {}); + const closed = new Promise((resolveClose, rejectClose) => { child.once('error', rejectClose); child.once('close', resolveClose); }); + closed.catch(() => {}); + const timer = setTimeout(() => { timedOut = true; child.kill('SIGKILL'); }, GIT_BATCH_TIMEOUT_MS); + const feed = (async () => { + let pending = ''; + for (const entry of entries) { + pending += `${entry.oid}\n`; + if (pending.length >= 65_536) { if (!child.stdin.write(pending)) await once(child.stdin, 'drain', { signal: abort.signal }); pending = ''; } + } + child.stdin.end(pending); + })().catch(() => {}); + try { + let buf = Buffer.alloc(0), index = 0, current; + for await (const chunk of child.stdout) { + buf = buf.length ? Buffer.concat([buf, chunk]) : chunk; + while (buf.length) { + if (!current) { + const nl = buf.indexOf(10); + if (nl < 0) { if (buf.length > 4096) throw Error('Git cat-file header too long'); break; } + const entry = entries[index]; + if (!entry) throw Error('Git cat-file returned more records than requested'); + const header = buf.toString('ascii', 0, nl); buf = buf.subarray(nl + 1); + if (header === `${entry.oid} missing`) throw Error(`Seed snapshot blob is missing: ${entry.path}`); + const match = /^([0-9a-f]{40}|[0-9a-f]{64}) (\w+) (\d+)$/.exec(header); + if (!match || match[1] !== entry.oid || match[2] !== 'blob') throw Error(`Unexpected Git cat-file header for ${entry.path}`); + const size = Number(match[3]); + await onHeader(entry, size); + current = { entry, size, parts: [], got: 0 }; + } + const take = buf.subarray(0, Math.min(current.size + 1 - current.got, buf.length)); + current.parts.push(take); current.got += take.length; buf = buf.subarray(take.length); + if (current.got < current.size + 1) break; + const record = Buffer.concat(current.parts, current.got); + if (record[current.size] !== 10) throw Error(`Git cat-file record for ${current.entry.path} is malformed`); + const done = current; current = undefined; index++; + await onBlob(done.entry, record.subarray(0, done.size)); + } + } + if (timedOut) throw Error('git cat-file timed out'); + if (current || index !== entries.length) throw Error('Git cat-file output was truncated'); + const code = await closed; + if (code !== 0) throw Error(`git cat-file failed (${code}): ${err.slice(0, 1000)}`); + } catch (error) { throw timedOut ? Error('git cat-file timed out') : error; } + finally { clearTimeout(timer); abort.abort(); child.kill('SIGKILL'); await feed; } +} async function seedWorkspace(baseCommit) { if (!SOURCE_REPO) throw Error('Workspace seeding requires LOCAL_CLI_UHP_SOURCE_REPO'); if (typeof baseCommit !== 'string' || !/^(?:[0-9a-f]{40}|[0-9a-f]{64})$/i.test(baseCommit)) throw Error('A full base commit SHA is required'); @@ -491,35 +602,29 @@ async function seedWorkspace(baseCommit) { try { treeText = utf8.decode(tree); } catch { throw Error('Git tree contains a non-UTF-8 path'); } const records = tree.length ? treeText.split('\0').filter(Boolean) : []; if (records.length > 100_000) throw Error('Seed snapshot entry limit exceeded'); - let total = 0; const parsed = []; for (const record of records) { const tab = record.indexOf('\t'); const [mode, type, oid] = record.slice(0, tab).split(' '); const path = record.slice(tab + 1); if (!validRelativePath(path) || type !== 'blob' || !['100644', '100755', '120000'].includes(mode)) throw Error(`Unsupported or unsafe Git tree entry: ${path}`); - const sizeText = (await git(repo, ['cat-file', '-s', oid], 64)).toString('ascii').trim(); - const size = Number(sizeText); - if (!/^\d+$/.test(sizeText) || !Number.isSafeInteger(size) || size > 16 * 1024 * 1024 || (total += size) > 256 * 1024 * 1024) throw Error(`Seed snapshot size limit exceeded: ${path}`); - parsed.push({ mode, oid, path, size }); - if (mode === '120000') { - let target; - try { target = utf8.decode(await git(repo, ['cat-file', 'blob', oid], 16 * 1024 * 1024)); } catch { throw Error(`Non-UTF-8 symlink target: ${path}`); } - if (linkEscapes(path, target)) throw Error(`Seed symlink escapes workspace: ${path}`); - } + parsed.push({ mode, oid, path }); } const id = `ws_${randomUUID()}`; const dir = join(ROOT, id); await mkdir(dir, { recursive: false, mode: 0o700 }); try { - for (const entry of parsed) { + let total = 0; + await gitBatchBlobs(repo, parsed, (entry, size) => { if (size > MAX_BLOB || (total += size) > MAX_SEED_TOTAL) throw Error(`Seed snapshot size limit exceeded: ${entry.path}`); }, async (entry, bytes) => { const target = join(dir, ...entry.path.split('/')); await mkdir(dirname(target), { recursive: true, mode: 0o700 }); - const bytes = await git(repo, ['cat-file', 'blob', entry.oid], 16 * 1024 * 1024); - if (bytes.length !== entry.size) throw Error(`Pinned Git blob changed or was truncated: ${entry.path}`); - if (entry.mode === '120000') await symlink(bytes.toString('utf8'), target); - else { await writeFile(target, bytes, { mode: entry.mode === '100755' ? 0o755 : 0o644, flag: 'wx' }); await chmod(target, entry.mode === '100755' ? 0o755 : 0o644); } - } - const ws = { id, baseCommit: sha, dir, sourceRepo: repo }; + if (entry.mode === '120000') { + let linkTarget; + try { linkTarget = utf8.decode(bytes); } catch { throw Error(`Non-UTF-8 symlink target: ${entry.path}`); } + if (linkEscapes(entry.path, linkTarget)) throw Error(`Seed symlink escapes workspace: ${entry.path}`); + await symlink(linkTarget, target); + } else { await writeFile(target, bytes, { mode: entry.mode === '100755' ? 0o755 : 0o644, flag: 'wx' }); await chmod(target, entry.mode === '100755' ? 0o755 : 0o644); } + }); + const ws = { id, baseCommit: sha, dir, sourceRepo: repo, lastUsed: Date.now() }; workspaces.set(id, ws); - state.workspaces[id] = { baseCommit: sha }; await persist(); + state.workspaces[id] = { baseCommit: sha, lastUsed: ws.lastUsed }; await persist(); return ws; } catch (error) { await rm(dir, { recursive: true, force: true }); throw error; } } @@ -1070,6 +1175,7 @@ async function runTask(record, prompt) { record.metadata.role_session = { binding: record.metadata.role_session_binding, cli_session_id: parsed.session, state_path: record.metadata.role_session_state_path }; } await persist(); + touchWorkspace(ws); touchWorkspace(roWs); for (const finish of active.finishListeners ?? []) finish(); if (kind === 'codex-cli' && !persistentRole) { const codexHome = reviewer ? reviewerState.codexHome : taskWorkspace.codexHome; @@ -1314,6 +1420,13 @@ const server = createServer(async (req, res) => { const b = await body(req); const ws = await seedWorkspace(b.base_commit); return send(res, 201, { workspace_id: ws.id, base_commit: ws.baseCommit }); } + const workspacePath = url.pathname.match(/^\/extensions\/foreman-workspace\/v1\/workspaces\/([^/]+)$/); + if (req.method === 'DELETE' && workspacePath) { + const ws = workspaces.get(decodeURIComponent(workspacePath[1])); + if (!ws) return send(res, 404, { error: { code: 'workspace_not_found' } }); + if (!(await removeWorkspace(ws.id))) return workspaces.has(ws.id) ? send(res, 409, { error: { code: 'workspace_task_in_progress' } }) : send(res, 404, { error: { code: 'workspace_not_found' } }); + res.writeHead(204, { 'UHP-Version': VERSION }); return res.end(); + } const overlayPath = url.pathname.match(/^\/extensions\/foreman-workspace\/v1\/workspaces\/([^/]+)\/overlay$/); if (req.method === 'POST' && overlayPath) { const ws = workspaces.get(decodeURIComponent(overlayPath[1])); @@ -1321,7 +1434,7 @@ const server = createServer(async (req, res) => { const response = ws.responseId ? state.responses[ws.responseId] : undefined; if (response?.status === 'in_progress') return send(res, 409, { error: { code: 'workspace_task_in_progress' } }); const b = await body(req); - const result = await overlayWorkspace(ws, b.entries); + const result = await withWorkspaceOp(ws, () => overlayWorkspace(ws, b.entries)); return send(res, 200, { workspace_id: ws.id, applied: result.applied }); } const snapshotPath = url.pathname.match(/^\/extensions\/foreman-workspace\/v1\/workspaces\/([^/]+)\/snapshot$/); @@ -1330,7 +1443,7 @@ const server = createServer(async (req, res) => { if (!ws) return send(res, 404, { error: { code: 'workspace_not_found' } }); const response = ws.responseId ? state.responses[ws.responseId] : undefined; if (response?.status === 'in_progress') return send(res, 409, { error: { code: 'workspace_task_in_progress' } }); - const snapshot = await snapshotWorkspace(ws); + const snapshot = await withWorkspaceOp(ws, () => snapshotWorkspace(ws)); if (ws.responseId && response?.status !== 'completed') { snapshot.complete = false; snapshot.errors.push({ path: '.', error: `task_status_${response?.status ?? 'unknown'}` }); @@ -1408,7 +1521,8 @@ const server = createServer(async (req, res) => { if (ws?.responseId) return send(res, 409, { error: { code: 'workspace_already_used' } }); const id = `resp_${randomUUID()}`; const record = { id, object: 'response', status: 'in_progress', requested_model: b.model, metadata: { ...b.metadata, harness_id: h.id, ...(workspaceId ? { workspace_id: workspaceId } : {}) }, timeout_seconds: timeout, max_step: steps }; - if (ws) { ws.responseId = id; state.workspaces[ws.id].responseId = id; } + if (ws) { ws.responseId = id; state.workspaces[ws.id].responseId = id; touchWorkspace(ws); } + touchWorkspace(workspaces.get(roWorkspaceId)); state.keys[key] = id; state.responses[id] = record; const task = { response: record, finishListeners: new Set(), activity: [], activityTotal: 0, activityListeners: new Set(), activityRemainder: '' }; recordActivity(task, 'Preparing isolated runtime'); tasks.set(id, task); await persist(); // durable idempotency intent before spawning any CLI process runTask(record, b.input).catch(async error => { record.status = 'failed'; record.error = { message: 'CLI task failed internally' }; if(task.reviewerWorkDir) { await rm(task.reviewerWorkDir,{recursive:true,force:true}).catch(()=>undefined); task.reviewerWorkDir=undefined; } const ws = workspaces.get(record.metadata.workspace_id); if (ws) { record.metadata.execution_boundary = ws.boundary ?? { proven: false, error: 'execution_boundary_unavailable' }; record.metadata.execution_stage = ws.executionStage ?? 'task_setup'; if (ws.boundaryDiagnostic) record.metadata.execution_boundary_diagnostic = ws.boundaryDiagnostic; else if (/^boundary_probe_failed:/.test(String(error?.message))) record.metadata.execution_boundary_diagnostic = { category: String(error.message).split(':')[1] ?? 'probe_failed', exit_code: Number(String(error.message).split(':')[2]) || null, signal: null }; if (record.metadata.harness_id === 'codex-cli' && record.metadata.foreman_review_mode !== 'read_only') { record.metadata.cli_exit = { exit_code: null, signal: null }; record.metadata.cli_failure_category = record.metadata.execution_stage === 'resolve_auth' ? 'host_auth_unavailable' : record.metadata.execution_stage === 'boundary_probe' ? 'boundary_setup' : record.metadata.execution_stage === 'runtime_mount' ? 'runtime_setup' : 'cli_setup'; await rm(ws.codexHome, { recursive: true, force: true }).catch(() => undefined); } } await persist(); for (const finish of task.finishListeners) finish(); }); @@ -1475,6 +1589,7 @@ if (process.argv[2] === '--preflight-claude-runtime') { console.log(await preflightAgyRuntime()); } else { await load(); + if (WORKSPACE_TTL_MS) setInterval(() => sweepIdleWorkspaces().catch(() => undefined), Math.min(Math.max(WORKSPACE_TTL_MS / 4, 100), 600_000)).unref(); server.on('error', error => { console.error(`local CLI UHP could not listen on 127.0.0.1:${PORT}: ${error.message}`); process.exit(1); }); if (!AUTH_TOKEN) console.warn('WARNING: LOCAL_CLI_UHP_TOKEN is not set; this bridge accepts unauthenticated requests from any local process (set it to require "Authorization: Bearer ").'); server.listen(PORT, '127.0.0.1', () => console.log(`experimental local CLI UHP listening on 127.0.0.1:${PORT}${AUTH_TOKEN ? ' (bearer token required)' : ''}`)); diff --git a/investigations/local-cli-uhp/test.mjs b/investigations/local-cli-uhp/test.mjs index 4ed7834..7044784 100644 --- a/investigations/local-cli-uhp/test.mjs +++ b/investigations/local-cli-uhp/test.mjs @@ -1,9 +1,10 @@ import test from 'node:test'; import assert from 'node:assert/strict'; -import { mkdtemp, writeFile, chmod, readFile, rm, truncate, mkdir, readdir, stat } from 'node:fs/promises'; +import { mkdtemp, writeFile, chmod, readFile, rm, truncate, mkdir, readdir, stat, lstat, symlink, readlink } from 'node:fs/promises'; import { tmpdir } from 'node:os'; import { join, dirname, resolve } from 'node:path'; -import { spawn } from 'node:child_process'; +import { spawn, execFile } from 'node:child_process'; +import { randomBytes, randomUUID } from 'node:crypto'; import { request as httpRequest } from 'node:http'; import { createServer as createNetServer } from 'node:net'; import { fileURLToPath } from 'node:url'; @@ -45,6 +46,7 @@ async function setup(t, options = {}) { await mkdir(join(dir,'codex-auth')); await writeFile(join(dir,'codex-auth','auth.json'),'{"fixture":true}',{mode:0o600}); const env = { ...process.env, ANTHROPIC_API_KEY:'fixture-only-do-not-forward', OPENAI_API_KEY:'fixture-only-do-not-forward', AWS_ACCESS_KEY_ID:'fixture-only-do-not-forward', GOOGLE_API_KEY:'fixture-only-do-not-forward', CLAUDE_CODE_USE_BEDROCK:'1', CLAUDE_CODE_USE_VERTEX:'1', CLAUDE_CODE_USE_FOUNDRY:'1', CODEX_API_KEY:'fixture-only-do-not-forward', ...(options.claudeNetworkRequirements ? {HTTP_PROXY:'http://fixture-proxy.invalid:8080'} : {}), ...(Number.isFinite(options.keepaliveMs) ? {LOCAL_CLI_UHP_KEEPALIVE_MS:String(options.keepaliveMs)} : {}), ...(options.agyWorkerEffort ? {AGY_WORKER_EFFORT:options.agyWorkerEffort} : {}), LOCAL_CLI_UHP_PORT: String(port), LOCAL_CLI_UHP_STATE: join(dir, 'state.json'), LOCAL_CLI_UHP_WORK: join(dir, 'work'), FOREMAN_CLAUDE_USAGE_CACHE: join(dir,'claude-usage-cache.json'), CLAUDE_CONFIG_DIR: join(dir, 'claude-auth'), CODEX_HOME: join(dir, 'codex-auth'), CLAUDE_MODEL: options.noClaudeModel ? '' : options.claudeModel ?? 'claude-requested', CODEX_MODEL: 'codex-requested', CLAUDE_BIN: options.claudeBin ?? (options.spawnError ? join(dir,'missing-cli') : claude), CODEX_BIN: codex, AGY_BIN:join(dir,'missing-agy'), ...(options.agyEnabled ? { AGY_CONFIG_DIR: join(dir,'agy-auth'), ...(options.noAgyModel ? {} : {AGY_MODEL:options.agyModel ?? 'gemini-3.8-flash-medium'}), AGY_BIN:agy } : {}), ...agyDiscoveryEnv, LOCAL_CLI_UHP_SOURCE_REPO: options.sourceRepo ?? fixture.repo, LOCAL_CLI_UHP_BWRAP: options.bwrapBin ?? 'bwrap' }; if (options.noAgyModel) delete env.AGY_MODEL; + if (options.extraEnv) Object.assign(env, options.extraEnv); if (!options.agyWorkerEffort) delete env.AGY_WORKER_EFFORT; if (options.agyEnabled) Object.assign(env,{AGY_CONFIG_DIR:join(dir,'agy-auth'),...(options.noAgyModel?{}:{AGY_MODEL:options.agyModel ?? 'gemini-3.8-flash-medium'}),AGY_BIN:agy}); // Tests run without a bridge token unless one is requested, whatever the developer's shell exports. @@ -1050,3 +1052,131 @@ test('an authenticated request with a malformed URL is rejected without crashing const result = await client.submit({ submissionId: 'sub-env', assignmentId: 'asg-env', runId: 'run-env', roleId: 'planner', taskId: 'task-env', projectId: 'prj-env', prompt: 'Report your environment.', config: { harnessId: 'claude-code', model: 'claude-requested', timeoutSeconds: 5 }, idempotencyKey: 'token-env-key' }); assert.equal(result.outputText, 'token_visible=false;env_entries=true'); }); + +// Batch seeding and workspace cleanup. +const gitIn = (repo, ...args) => new Promise((resolveGit, reject) => execFile('git', ['-C', repo, ...args], { encoding: 'buffer', maxBuffer: 1 << 28 }, (error, out) => error ? reject(error) : resolveGit(out))); +async function makeRepo(t, populate) { + const repo = await mkdtemp(join(tmpdir(), 'uhp-batch-repo-')); t.after(() => rm(repo, { recursive: true, force: true })); + await gitIn(repo, 'init', '-q'); await gitIn(repo, 'config', 'user.name', 'Fixture'); await gitIn(repo, 'config', 'user.email', 'fixture@example.invalid'); + await populate(repo); + await gitIn(repo, 'add', '--all'); await gitIn(repo, 'commit', '-qm', 'fixture'); + return { repo, baseCommit: (await gitIn(repo, 'rev-parse', 'HEAD')).toString().trim() }; +} +const seedRaw = (base, baseCommit, headers = {}) => fetch(`${base}/extensions/foreman-workspace/v1/workspaces`, { method: 'POST', headers: { 'Content-Type': 'application/json', ...headers }, body: JSON.stringify({ base_commit: baseCommit }) }); +const deleteWorkspace = (base, id, headers = {}) => fetch(`${base}/extensions/foreman-workspace/v1/workspaces/${id}`, { method: 'DELETE', headers }); +async function bridgeGitChildren(repo) { + const found = []; + for (const pid of (await readdir('/proc')).filter(name => /^\d+$/.test(name))) { try { if ((await readFile(`/proc/${pid}/cmdline`, 'utf8')).includes(`${repo}\0cat-file`)) found.push(pid); } catch {} } + return found; +} +const slowClaude = ms => `process.stdin.resume();process.stdin.on('end',()=>setTimeout(()=>{console.log(JSON.stringify({type:'system',subtype:'init',model:'claude-requested',session_id:'slow-session'}));console.log(JSON.stringify({type:'result',subtype:'success',result:'slow bounded answer',model:'claude-requested',session_id:'slow-session',usage:{input_tokens:3,output_tokens:2}}));},${ms}));`; +async function startResponse(base, headers, key, workspaceId) { + const r = await fetch(`${base}/v1/responses`, { method: 'POST', headers: { 'Content-Type': 'application/json', Accept: 'text/event-stream', 'UHP-Version': '2026-09-12', 'Idempotency-Key': key, ...headers }, body: JSON.stringify({ input: 'Say bounded answer', model: 'claude-requested', metadata: { harness_id: 'claude-code', workspace_id: workspaceId }, stream: true, timeout_seconds: 10, max_step: 1 }) }); + assert.equal(r.status, 200); return r; +} +async function until(check, ms = 8000) { const end = Date.now() + ms; while (Date.now() < end) { if (await check()) return true; await new Promise(r => setTimeout(r, 50)); } return false; } +const exists = path => lstat(path).then(() => true, () => false); + +test('batch seeding reproduces binary, empty, executable and symlink blobs across more than 200 files', async t => { + const files = new Map(); + const { repo, baseCommit } = await makeRepo(t, async repo => { + for (let i = 0; i < 250; i++) { const path = `dir${i % 7}/sub/file${i}.txt`; files.set(path, Buffer.from(`file ${i}\n`.repeat(1 + (i % 5)))); } + files.set('bin/random.bin', randomBytes(300_000)); + files.set('bin/all.bin', Buffer.concat([Buffer.from(Array.from({ length: 256 }, (_, i) => i)), Buffer.from('\n\n0123456789abcdef 1 2\n'), Buffer.alloc(70_000, 10)])); + files.set('empty.txt', Buffer.alloc(0)); + for (const [path, bytes] of files) { await mkdir(dirname(join(repo, path)), { recursive: true }); await writeFile(join(repo, path), bytes); } + await writeFile(join(repo, 'run.sh'), '#!/bin/sh\necho hi\n', { mode: 0o755 }); await chmod(join(repo, 'run.sh'), 0o755); + await symlink('dir0/sub/file0.txt', join(repo, 'link')); + }); + const { base, env } = await setup(t, { sourceRepo: repo }); + const seed = await seedRaw(base, baseCommit); assert.equal(seed.status, 201); + const work = join(env.LOCAL_CLI_UHP_WORK, (await seed.json()).workspace_id); + for (const [path, bytes] of files) assert.ok((await readFile(join(work, path))).equals(bytes), path); + assert.equal((await stat(join(work, 'run.sh'))).mode & 0o777, 0o755); assert.equal((await stat(join(work, 'empty.txt'))).mode & 0o777, 0o644); + assert.equal(await readlink(join(work, 'link')), 'dir0/sub/file0.txt'); + assert.deepEqual(await bridgeGitChildren(repo), []); +}); + +test('batch seeding fails cleanly on a blob over the cap, a missing blob and an escaping symlink, and leaves no workspace or git child', async t => { + const big = await makeRepo(t, async repo => { await writeFile(join(repo, 'a.txt'), 'small\n'); await writeFile(join(repo, 'big.bin'), Buffer.alloc(16 * 1024 * 1024 + 1)); }); + const missing = await makeRepo(t, async repo => { await writeFile(join(repo, 'a.txt'), 'small\n'); await writeFile(join(repo, 'gone.txt'), 'this blob will be deleted\n'); }); + const oid = (await gitIn(missing.repo, 'rev-parse', `${missing.baseCommit}:gone.txt`)).toString().trim(); + await rm(join(missing.repo, '.git', 'objects', oid.slice(0, 2), oid.slice(2)), { force: true }); + const escape = await makeRepo(t, async repo => { await writeFile(join(repo, 'a.txt'), 'small\n'); await symlink('../outside', join(repo, 'evil')); }); + const { base, env } = await setup(t, { sourceRepo: big.repo }); + const r1 = await seedRaw(base, big.baseCommit); assert.equal(r1.status, 500); assert.match((await r1.json()).error.message, /size limit exceeded: big\.bin/); + assert.deepEqual(await readdir(env.LOCAL_CLI_UHP_WORK), []); assert.deepEqual(await bridgeGitChildren(big.repo), []); + const other = await setup(t, { sourceRepo: missing.repo }); + const r2 = await seedRaw(other.base, missing.baseCommit); assert.equal(r2.status, 500); assert.match((await r2.json()).error.message, /blob is missing: gone\.txt/); + assert.deepEqual(await readdir(other.env.LOCAL_CLI_UHP_WORK), []); assert.deepEqual(await bridgeGitChildren(missing.repo), []); + const third = await setup(t, { sourceRepo: escape.repo }); + const r3 = await seedRaw(third.base, escape.baseCommit); assert.equal(r3.status, 500); assert.match((await r3.json()).error.message, /symlink escapes workspace: evil/); + assert.deepEqual(await readdir(third.env.LOCAL_CLI_UHP_WORK), []); + assert.equal((await fetch(`${other.base}/v1/uhp`)).status, 200); +}); + +test('DELETE removes a workspace and its sibling dirs: 404 unknown, 401 without the token, 409 while a response runs, 204 after', async t => { + const { base, env, baseCommit, authHeaders } = await setup(t, { token: BRIDGE_TOKEN, claudeBody: slowClaude(1500) }); + const seed = await (await seedRaw(base, baseCommit, authHeaders)).json(); const id = seed.workspace_id; + const work = join(env.LOCAL_CLI_UHP_WORK, id); const codexHome = `${work}.codex-home`; await mkdir(codexHome); + assert.equal((await deleteWorkspace(base, `ws_${randomUUID()}`, authHeaders)).status, 404); + for (const headers of [{}, { authorization: 'Bearer wrong' }]) assert.equal((await deleteWorkspace(base, id, headers)).status, 401); + assert.ok(await exists(work)); + const running = await startResponse(base, authHeaders, 'delete-running', id); + const blocked = await deleteWorkspace(base, id, authHeaders); assert.equal(blocked.status, 409); assert.equal((await blocked.json()).error.code, 'workspace_task_in_progress'); + assert.ok(await exists(work)); + assert.equal(terminalEvent((await running.text()).split('\n').filter(x => x.startsWith('data: ')).map(x => JSON.parse(x.slice(6))))?.type, 'response.completed'); + const removed = await deleteWorkspace(base, id, authHeaders); assert.equal(removed.status, 204); assert.equal(await removed.text(), ''); + assert.ok(!(await exists(work))); assert.ok(!(await exists(codexHome))); + assert.equal(JSON.parse(await readFile(env.LOCAL_CLI_UHP_STATE, 'utf8')).workspaces[id], undefined); + assert.equal((await deleteWorkspace(base, id, authHeaders)).status, 404); + assert.equal((await fetch(`${base}/extensions/foreman-workspace/v1/workspaces/${id}/snapshot`, { headers: authHeaders })).status, 404); +}); + +test('DELETE and the startup sweep never remove anything outside the work root through a symlinked workspace dir', async t => { + const { base, env, baseCommit, restart } = await setup(t); + const outside = await mkdtemp(join(tmpdir(), 'uhp-outside-')); t.after(() => rm(outside, { recursive: true, force: true })); + await writeFile(join(outside, 'precious.txt'), 'keep me'); + const id = (await (await seedRaw(base, baseCommit)).json()).workspace_id; const work = join(env.LOCAL_CLI_UHP_WORK, id); + await rm(work, { recursive: true, force: true }); await symlink(outside, work); await symlink(outside, `${work}.codex-home`); + assert.equal((await deleteWorkspace(base, id)).status, 204); + assert.ok(!(await exists(work))); assert.ok(!(await exists(`${work}.codex-home`))); + assert.equal(await readFile(join(outside, 'precious.txt'), 'utf8'), 'keep me'); + const orphan = join(env.LOCAL_CLI_UHP_WORK, `ws_${randomUUID()}`); await symlink(outside, orphan); await symlink(outside, join(env.LOCAL_CLI_UHP_WORK, `review-${randomUUID()}`)); + await restart(); + assert.ok(!(await exists(orphan))); assert.deepEqual(await readdir(env.LOCAL_CLI_UHP_WORK), []); + assert.equal(await readFile(join(outside, 'precious.txt'), 'utf8'), 'keep me'); +}); + +test('startup removes work-root entries no persisted state references and keeps everything that is referenced', async t => { + const { base, env, baseCommit, restart } = await setup(t); + const root = env.LOCAL_CLI_UHP_WORK; + const kept = (await (await seedRaw(base, baseCommit)).json()).workspace_id; await mkdir(join(root, `${kept}.codex-home`)); + const orphan = `ws_${randomUUID()}`; const orphanFiles = [orphan, `${orphan}.codex-home`, `review-${randomUUID()}`, `resp_${randomUUID()}.review-codex-home`, `codex-preflight-${randomUUID()}`]; + for (const name of orphanFiles) { await mkdir(join(root, name)); await writeFile(join(root, name, 'x'), 'x'); } + await chmod(join(root, orphanFiles[2]), 0o500); + for (const name of ['kept-role', 'orphan-role']) { await mkdir(join(root, 'role-sessions', name), { recursive: true }); await writeFile(join(root, 'role-sessions', name, 'x'), 'x'); } + await writeFile(join(root, 'not-ours.txt'), 'foreign'); await mkdir(join(root, 'ws_not-a-uuid')); + const statePath = env.LOCAL_CLI_UHP_STATE; const persisted = JSON.parse(await readFile(statePath, 'utf8')); + persisted.responses.resp_role = { id: 'resp_role', object: 'response', status: 'completed', metadata: { role_session_state_path: join(root, 'role-sessions', 'kept-role') } }; + await writeFile(statePath, JSON.stringify(persisted)); + await restart(); + assert.deepEqual((await readdir(root)).sort(), [kept, `${kept}.codex-home`, 'not-ours.txt', 'role-sessions', 'ws_not-a-uuid'].sort()); + assert.deepEqual(await readdir(join(root, 'role-sessions')), ['kept-role']); + assert.equal((await fetch(`${base}/extensions/foreman-workspace/v1/workspaces/${kept}/snapshot`)).status, 200); + await rm(join(root, kept), { recursive: true, force: true }); await restart(); + assert.equal(JSON.parse(await readFile(statePath, 'utf8')).workspaces[kept], undefined); +}); + +test('the TTL sweep removes idle workspaces but never one with a running response', async t => { + const { base, env, baseCommit } = await setup(t, { claudeBody: slowClaude(3000), extraEnv: { LOCAL_CLI_UHP_WORKSPACE_TTL_MS: '800' } }); + const idle = (await (await seedRaw(base, baseCommit)).json()).workspace_id; const busy = (await (await seedRaw(base, baseCommit)).json()).workspace_id; + const busyDir = join(env.LOCAL_CLI_UHP_WORK, busy); + const running = await startResponse(base, {}, 'ttl-running', busy); let finished = false; const body = running.text().then(text => { finished = true; return text; }); + assert.ok(await until(async () => !(await exists(join(env.LOCAL_CLI_UHP_WORK, idle)))), 'the idle workspace was never swept'); + assert.equal(finished, false); assert.ok(await exists(busyDir), 'the busy workspace was swept mid-run'); + assert.match(await body, /response\.completed/); + assert.ok(await exists(busyDir)); + assert.ok(await until(async () => !(await exists(busyDir))), 'the finished workspace was never swept once idle'); + const persisted = JSON.parse(await readFile(env.LOCAL_CLI_UHP_STATE, 'utf8')).workspaces; assert.equal(persisted[idle], undefined); assert.equal(persisted[busy], undefined); +});