diff --git a/src/cli/fleet.test.ts b/src/cli/fleet.test.ts index 17558e1..0525462 100644 --- a/src/cli/fleet.test.ts +++ b/src/cli/fleet.test.ts @@ -1329,8 +1329,10 @@ describe('fleet CLI runtime', () => { it('does not infer or preflight clone paths for a maintenance command', async () => { const root = await mkdtemp(join(tmpdir(), 'fleet-cli-maintenance-clone-')) try { + const heartbeatPath = join(root, 'heartbeat.json') const configPath = await writeConfig(root, { repos: { org: 'AgentWorkforce', names: ['pear'] }, + loop: { heartbeatPath, heartbeatStaleMs: 10_000 }, }) const git = vi.fn(async () => { throw new Error('status must not inspect local git state') @@ -1354,7 +1356,13 @@ describe('fleet CLI runtime', () => { }) expect(code).toBe(0) - expect(JSON.parse(output.text())).toEqual(factoryStatus) + expect(JSON.parse(output.text())).toEqual({ + ...factoryStatus, + eventListener: { + state: 'not-listening', + reason: 'heartbeat missing', + }, + }) expect(git).not.toHaveBeenCalled() expect(integrations.getStatus).not.toHaveBeenCalled() } finally { @@ -1828,7 +1836,8 @@ describe('fleet CLI runtime', () => { it('prints factory status from the top-level status command', async () => { const root = await mkdtemp(join(tmpdir(), 'fleet-cli-status-')) try { - const configPath = await writeConfig(root) + const heartbeatPath = join(root, 'heartbeat.json') + const configPath = await writeConfig(root, { loop: { heartbeatPath, heartbeatStaleMs: 10_000 } }) const output = buffer() const factoryStatus = { inFlight: [], queued: [], counters: { pulled: 0 } } const factory = { @@ -1856,7 +1865,131 @@ describe('fleet CLI runtime', () => { }) expect(code).toBe(0) - expect(JSON.parse(output.text())).toEqual(factoryStatus) + expect(JSON.parse(output.text())).toEqual({ + ...factoryStatus, + eventListener: { + state: 'not-listening', + reason: 'heartbeat missing', + }, + }) + } finally { + await rm(root, { recursive: true, force: true }) + } + }) + + it('surfaces a stale registered workspace mirror in factory status', async () => { + const root = await mkdtemp(join(tmpdir(), 'fleet-cli-stale-status-')) + try { + const heartbeatPath = join(root, 'heartbeat.json') + const configPath = await writeConfig(root, { loop: { heartbeatPath, heartbeatStaleMs: 10_000 } }) + const output = buffer() + const mirror = join(root, 'chief', '.integrations') + const factory = { + start: vi.fn(), + stop: vi.fn(), + runLoop: vi.fn(async () => []), + runOnce: vi.fn(), + status: vi.fn(() => ({ inFlight: [], queued: [], counters: {} })), + triageIssue: vi.fn(), + dispatch: vi.fn(), + on: vi.fn(), + dispose: vi.fn(), + } as unknown as Factory + const mount = Object.assign(new FakeMountClient(), { + getLocalMountHealth: () => ({ + degraded: true, + reason: 'last reconcile 5m ago', + localDir: mirror, + }), + }) + const now = Date.now() + await writeFile(heartbeatPath, JSON.stringify({ + pid: process.pid, + status: 'running', + iteration: 0, + maxIterations: 0, + updatedAt: new Date(now).toISOString(), + updatedAtMs: now, + eventListener: { state: 'subscribed' }, + })) + + const code = await runFleetCli(['status', '--config', configPath], { + fleet: new FakeFleetClient(), + mount, + createFactory: () => factory, + stdout: output, + stderr: buffer(), + }) + + expect(code).toBe(0) + expect(JSON.parse(output.text())).toMatchObject({ + eventListener: { + state: 'subscribed', + }, + localMountDegraded: true, + localMountDegradedReason: 'last reconcile 5m ago', + localMountRoot: mirror, + localMountEventFeed: { + state: 'degraded', + livenessSignal: '.integrations/.relay/state.json', + reason: 'last reconcile 5m ago', + root: mirror, + }, + }) + } finally { + await rm(root, { recursive: true, force: true }) + } + }) + + it('distinguishes a healthy quiet mount event feed from a daemon that is not listening', async () => { + const root = await mkdtemp(join(tmpdir(), 'fleet-cli-event-status-')) + try { + const heartbeatPath = join(root, 'heartbeat.json') + const configPath = await writeConfig(root, { loop: { heartbeatPath, heartbeatStaleMs: 10_000 } }) + const output = buffer() + const mirror = join(root, 'chief', '.integrations') + const factory = { + start: vi.fn(), + stop: vi.fn(), + runLoop: vi.fn(async () => []), + runOnce: vi.fn(), + status: vi.fn(() => ({ inFlight: [], queued: [], counters: {} })), + triageIssue: vi.fn(), + dispatch: vi.fn(), + on: vi.fn(), + dispose: vi.fn(), + } as unknown as Factory + const mount = Object.assign(new FakeMountClient(), { + getLocalMountHealth: () => ({ degraded: false, localDir: mirror }), + }) + const now = Date.now() + await writeFile(heartbeatPath, JSON.stringify({ + pid: process.pid, + status: 'running', + iteration: 0, + maxIterations: 0, + updatedAt: new Date(now).toISOString(), + updatedAtMs: now, + eventListener: { state: 'subscribed' }, + })) + + const code = await runFleetCli(['status', '--config', configPath], { + fleet: new FakeFleetClient(), + mount, + createFactory: () => factory, + stdout: output, + stderr: buffer(), + }) + + expect(code).toBe(0) + expect(JSON.parse(output.text())).toMatchObject({ + eventListener: { state: 'subscribed' }, + localMountEventFeed: { + state: 'healthy', + livenessSignal: '.integrations/.relay/state.json', + root: mirror, + }, + }) } finally { await rm(root, { recursive: true, force: true }) } @@ -2008,8 +2141,8 @@ describe('fleet CLI runtime', () => { expect(code).toBe(0) expect(integrations.getStatus).toHaveBeenCalledWith('github') - expect(mountCalls).toEqual([process.cwd(), clonePath]) - expect(errors.text()).toContain(`warning: could not start relayfile mount for standalone babysitter at ${clonePath}`) + expect(mountCalls).toEqual([process.cwd()]) + expect(errors.text()).toBe('') expect(fleet.spawns).toHaveLength(1) expect(fleet.preservedInfrastructure).toBe(1) expect(fleet.spawns[0]).toMatchObject({ @@ -2309,11 +2442,12 @@ describe('fleet CLI runtime', () => { } }) - it('summarizes stale clone-path mount refreshes once and keeps path details at debug verbosity', async () => { + it('refreshes one stale registered workspace mirror once, regardless of routed clone count', async () => { const root = await mkdtemp(join(tmpdir(), 'fleet-cli-stale-mounts-')) const previousCwd = process.cwd() try { const clonePaths = [join(root, 'pear'), join(root, 'relay')] + const mirrorDir = join(root, 'chief', '.integrations') await Promise.all(clonePaths.map((clonePath) => mkdir(clonePath))) const configPath = await writeConfig(root, { repos: { @@ -2358,14 +2492,17 @@ describe('fleet CLI runtime', () => { dispose: vi.fn(), } as unknown as Factory const errors = buffer() + const mount = Object.assign(new FakeMountClient(), { + getLocalMountRoot: () => mirrorDir, + }) const code = await runFleetCli(['start', '--config', configPath], { fleet: new FakeFleetClient(), - mount: new FakeMountClient(), + mount, createFactory: vi.fn(() => factory), ensureLocalMount, waitForStopSignal: vi.fn(async () => { await vi.waitFor(() => { - expect(errors.text()).toContain('[factory] refreshed 2 stale local mount(s)') + expect(errors.text()).toContain('[factory] refreshed 1 stale local mount(s)') }) }), env: debug ? { FACTORY_LOG_LEVEL: 'debug' } : {}, @@ -2377,20 +2514,17 @@ describe('fleet CLI runtime', () => { } const staleAt = Date.now() - await writeMountState(clonePaths[0]!, new Date(staleAt - 30 * 60 * 1000).toISOString()) - await writeMountState(clonePaths[1]!, new Date(staleAt - 31 * 60 * 1000).toISOString()) + await writeMountState(dirname(mirrorDir), new Date(staleAt - 31 * 60 * 1000).toISOString()) const normalOutput = await runStart(false) - expect(normalOutput).toContain('[factory] refreshed 2 stale local mount(s) (last reconcile ~31m ago)') + expect(normalOutput).toContain('[factory] refreshed 1 stale local mount(s) (last reconcile ~31m ago)') expect(normalOutput).not.toContain('local mount is stale') expect(normalOutput).not.toContain('[factory] debug:') expect(normalOutput.match(/stale local mount/gu)).toHaveLength(1) - await writeMountState(clonePaths[0]!, new Date(staleAt - 30 * 60 * 1000).toISOString()) - await writeMountState(clonePaths[1]!, new Date(staleAt - 31 * 60 * 1000).toISOString()) + await writeMountState(dirname(mirrorDir), new Date(staleAt - 31 * 60 * 1000).toISOString()) const debugOutput = await runStart(true) - expect(debugOutput).toContain(`[factory] debug: refreshed stale local mount at ${clonePaths[0]}`) - expect(debugOutput).toContain(`[factory] debug: refreshed stale local mount at ${clonePaths[1]}`) - expect(debugOutput).toContain('[factory] refreshed 2 stale local mount(s)') + expect(debugOutput).toContain(`[factory] debug: refreshed stale local mount at ${mirrorDir}`) + expect(debugOutput).toContain('[factory] refreshed 1 stale local mount(s)') } finally { process.chdir(previousCwd) await rm(root, { recursive: true, force: true }) @@ -2443,9 +2577,7 @@ describe('fleet CLI runtime', () => { expect(ensureLocalMount).toHaveBeenCalledWith('rw_7ccfea89', process.cwd(), { acceptableWorkspaceIds: ['50587328-441d-4acb-b8f3-dbe1b3c5de99'], }) - expect(ensureLocalMount).toHaveBeenCalledWith('rw_7ccfea89', '/work/pear', { - acceptableWorkspaceIds: ['50587328-441d-4acb-b8f3-dbe1b3c5de99'], - }) + expect(ensureLocalMount).toHaveBeenCalledTimes(1) } finally { await rm(root, { recursive: true, force: true }) } @@ -2488,9 +2620,7 @@ describe('fleet CLI runtime', () => { expect(ensureSdkMount).toHaveBeenCalledWith(process.cwd(), { acceptableWorkspaceIds: undefined, }) - expect(ensureSdkMount).toHaveBeenCalledWith('/work/pear', { - acceptableWorkspaceIds: undefined, - }) + expect(ensureSdkMount).toHaveBeenCalledTimes(1) expect(disposeMount).toHaveBeenCalledTimes(1) } finally { await rm(root, { recursive: true, force: true }) @@ -2577,9 +2707,7 @@ describe('fleet CLI runtime', () => { expect(ensureLocalMount).toHaveBeenCalledWith('factory-cli-test', process.cwd(), { acceptableWorkspaceIds: undefined, }) - expect(ensureLocalMount).toHaveBeenCalledWith('factory-cli-test', '/work/pear', { - acceptableWorkspaceIds: undefined, - }) + expect(ensureLocalMount).toHaveBeenCalledTimes(1) expect(createFactory).toHaveBeenCalledTimes(1) expect(createFactory.mock.calls[0]?.[1].stateStore).toBeInstanceOf(FileStateStore) expect(factory.start).toHaveBeenCalledWith({ mode: 'live' }) @@ -2591,11 +2719,11 @@ describe('fleet CLI runtime', () => { } }) - it('warms configured clone mounts with bounded concurrency without blocking live start', async () => { + it('warms one registered workspace mirror without blocking live start for sixteen routes', async () => { const root = await mkdtemp(join(tmpdir(), 'fleet-cli-start-mount-concurrency-')) try { const clonePaths = Object.fromEntries( - Array.from({ length: 9 }, (_, index) => [`AgentWorkforce/repo-${index}`, join(root, `repo-${index}`)]), + Array.from({ length: 16 }, (_, index) => [`AgentWorkforce/repo-${index}`, join(root, `repo-${index}`)]), ) const configPath = await writeConfig(root, { repos: { @@ -2605,6 +2733,9 @@ describe('fleet CLI runtime', () => { }, }) const mounted: string[] = [] + const mirrorDir = join(root, 'chief', '.integrations') + let releaseMount!: () => void + const mountReleased = new Promise((resolve) => { releaseMount = resolve }) let mountedWhenFactoryStarted = -1 const factory = { start: vi.fn(async () => { mountedWhenFactoryStarted = mounted.length }), @@ -2617,38 +2748,181 @@ describe('fleet CLI runtime', () => { on: vi.fn(), dispose: vi.fn(), } as unknown as Factory - let active = 0 - let maxActive = 0 const ensureLocalMount = vi.fn(async (_workspaceId: string, startDir: string) => { - if (startDir === process.cwd()) return + await mountReleased mounted.push(startDir) - active += 1 - maxActive = Math.max(maxActive, active) - await new Promise((resolve) => setTimeout(resolve, 10)) - active -= 1 + }) + const mount = Object.assign(new FakeMountClient(), { + getLocalMountRoot: () => mirrorDir, }) await runFleetCli(['start', '--config', configPath], { fleet: new FakeFleetClient(), - mount: new FakeMountClient(), + mount, createFactory: vi.fn(() => factory), ensureLocalMount, waitForStopSignal: vi.fn(async () => { - await vi.waitFor(() => expect(mounted).toHaveLength(9)) + releaseMount() + await vi.waitFor(() => expect(mounted).toHaveLength(1)) }), stdout: buffer(), stderr: buffer(), }) - expect(maxActive).toBe(4) - expect(mounted.sort()).toEqual(Object.values(clonePaths).sort()) - expect(mountedWhenFactoryStarted).toBeLessThan(Object.keys(clonePaths).length) + expect(mounted).toEqual([dirname(mirrorDir)]) + expect(ensureLocalMount).toHaveBeenCalledTimes(1) + expect(mountedWhenFactoryStarted).toBeLessThan(1) expect(factory.start).toHaveBeenCalledWith({ mode: 'live' }) } finally { await rm(root, { recursive: true, force: true }) } }) + it('resolves an unknown workspace mirror before enabling a live factory', async () => { + const root = await mkdtemp(join(tmpdir(), 'fleet-cli-resolve-mirror-before-start-')) + try { + const configPath = await writeConfig(root) + const events: string[] = [] + let registeredRoot: string | undefined + const mount = Object.assign(new FakeMountClient(), { + getLocalMountRoot: () => registeredRoot, + }) + const factory = { + start: vi.fn(async () => { events.push('factory-start') }), + stop: vi.fn(async () => {}), + runLoop: vi.fn(async () => []), + runOnce: vi.fn(), + status: vi.fn(), + triageIssue: vi.fn(), + dispatch: vi.fn(), + on: vi.fn(), + dispose: vi.fn(), + } as unknown as Factory + const ensureLocalMount = vi.fn(async () => { + events.push('mount-resolved') + registeredRoot = join(root, 'chief', '.integrations') + }) + + const code = await runFleetCli(['start', '--config', configPath], { + fleet: new FakeFleetClient(), + mount, + createFactory: vi.fn(() => factory), + ensureLocalMount, + waitForStopSignal: vi.fn(async () => undefined), + stdout: buffer(), + stderr: buffer(), + }) + + expect(code).toBe(0) + expect(events).toEqual(['mount-resolved', 'factory-start']) + expect(ensureLocalMount).toHaveBeenCalledTimes(1) + } finally { + await rm(root, { recursive: true, force: true }) + } + }) + + it('handles SIGTERM gracefully while an unknown workspace mirror is still resolving', async () => { + const root = await mkdtemp(join(tmpdir(), 'fleet-cli-resolve-mirror-sigterm-')) + try { + const configPath = await writeConfig(root) + const listeners = new Map void>() + const processLike = { + once(signal: string, listener: () => void) { + listeners.set(signal, listener) + return processLike + }, + off(signal: string, listener: () => void) { + if (listeners.get(signal) === listener) listeners.delete(signal) + return processLike + }, + } + const calls: string[] = [] + let registeredRoot: string | undefined + let releaseMount!: () => void + const mountReleased = new Promise((resolve) => { releaseMount = resolve }) + const mount = Object.assign(new FakeMountClient(), { + getLocalMountRoot: () => registeredRoot, + }) + const factory = { + start: vi.fn(async () => {}), + stop: vi.fn(async () => { calls.push('stop') }), + runLoop: vi.fn(async () => []), + runOnce: vi.fn(), + status: vi.fn(), + triageIssue: vi.fn(), + dispatch: vi.fn(), + on: vi.fn(), + dispose: vi.fn(), + } as unknown as Factory + const ensureLocalMount = vi.fn(async () => { + await mountReleased + registeredRoot = join(root, 'chief', '.integrations') + }) + + const run = runFleetCli(['start', '--config', configPath], { + fleet: new FakeFleetClient(), + mount, + createFactory: vi.fn(() => factory), + ensureLocalMount, + waitForStopSignal: vi.fn(async () => undefined), + stopSignalProcessLike: processLike as unknown as Pick, + flushDaemonOutput: async () => { calls.push('flush') }, + stdout: buffer(), + stderr: buffer(), + }) + + await vi.waitFor(() => { + expect(ensureLocalMount).toHaveBeenCalledTimes(1) + expect(listeners.has('SIGTERM')).toBe(true) + }) + listeners.get('SIGTERM')?.() + await vi.waitFor(() => expect(calls).toEqual(['stop', 'flush'])) + releaseMount() + + await expect(run).resolves.toBe(0) + expect(factory.start).not.toHaveBeenCalled() + expect(factory.stop).toHaveBeenCalledTimes(1) + expect(listeners.size).toBe(0) + } finally { + await rm(root, { recursive: true, force: true }) + } + }) + + it('does not start a live factory when an unknown workspace mirror cannot be resolved', async () => { + const root = await mkdtemp(join(tmpdir(), 'fleet-cli-unresolved-mirror-')) + try { + const configPath = await writeConfig(root) + const factory = { + start: vi.fn(async () => {}), + stop: vi.fn(async () => {}), + runLoop: vi.fn(async () => []), + runOnce: vi.fn(), + status: vi.fn(), + triageIssue: vi.fn(), + dispatch: vi.fn(), + on: vi.fn(), + dispose: vi.fn(), + } as unknown as Factory + const errors = buffer() + + const code = await runFleetCli(['start', '--config', configPath], { + fleet: new FakeFleetClient(), + mount: Object.assign(new FakeMountClient(), { getLocalMountRoot: () => undefined }), + createFactory: vi.fn(() => factory), + ensureLocalMount: vi.fn(async () => { throw new Error('admission refused') }), + waitForStopSignal: vi.fn(async () => undefined), + stdout: buffer(), + stderr: errors, + }) + + expect(code).toBe(1) + expect(factory.start).not.toHaveBeenCalled() + expect(errors.text()).toContain('aborting startup: Relayfile workspace mirror could not be resolved') + } finally { + await rm(root, { recursive: true, force: true }) + } + }) + it('uses ./factory.config.json by default for factory commands', async () => { const root = await mkdtemp(join(tmpdir(), 'fleet-cli-default-config-')) const previousCwd = process.cwd() diff --git a/src/cli/fleet.ts b/src/cli/fleet.ts index fd6fc12..63fd480 100644 --- a/src/cli/fleet.ts +++ b/src/cli/fleet.ts @@ -133,8 +133,6 @@ interface LoadedConfig { } const autoDetectedIssueSources = new WeakSet() -const CLONE_MOUNT_PREFLIGHT_CONCURRENCY = 4 - type ParsedCommand = | { kind: 'spawn'; input: { capability: Capability; name?: string; node?: 'self' | string; task?: string; workflow?: string; model?: string; sessionRef?: string; cwd?: string } } | { kind: 'roster' } @@ -629,16 +627,18 @@ async function runFactoryCommand( waiter.resolve(code) } } - // Local mirrors are a writeback aid, not the source of truth for remote - // issue discovery. Start their SDK-backed supervisors immediately, but - // do not serialize durable recovery behind a stale checkout's readiness - // timeout. The mount client reports degradation and keeps retrying. + // Once a workspace mirror is known, retain background stale-mount + // supervision so durable recovery is not serialized behind a refresh. + // If there is no registered root yet, however, wait for the single + // mount/admission fallback before Factory can dispatch: agents must not + // receive a provisional checkout-local `.integrations` path. // // A MountAuthScopeError is the exception: it is terminal (the cloud // session lacks the filesystem scope the mount needs), so limping on // would only spawn agents against a read-denied mirror. Fail fast with the // remediation and resolve the command with a non-zero code. - void warmStartPathMounts( + const warmMount = () => warmStartPathMounts( + mount, mountFn, workspaceId, config, @@ -646,16 +646,16 @@ async function runFactoryCommand( mountStderr, debugMountRefreshes, ) - .catch((error: unknown) => { - if (error instanceof MountAuthScopeError) { - mountStderr.write(`${error.message}\n`) - mountStderr.write('[factory] aborting startup: local mount cannot obtain its filesystem scopes.\n') - void flushAndResolve(1) - return - } - const message = error instanceof Error ? error.message : String(error) - mountStderr.write(`[factory] warning: background relayfile mount warmup failed: ${message}\n`) - }) + const handleWarmMountError = (error: unknown): void => { + if (error instanceof MountAuthScopeError) { + mountStderr.write(`${error.message}\n`) + mountStderr.write('[factory] aborting startup: local mount cannot obtain its filesystem scopes.\n') + void flushAndResolve(1) + return + } + const message = error instanceof Error ? error.message : String(error) + mountStderr.write(`[factory] warning: background relayfile mount warmup failed: ${message}\n`) + } const removeSignalHandlers = installFactoryStopSignalHandlers(factory, { exit: (code) => { stoppedBySignal = true @@ -667,6 +667,23 @@ async function runFactoryCommand( processLike: deps.stopSignalProcessLike, }) try { + if (mount.getLocalMountRoot?.() === undefined) { + try { + const result = await warmMount() + if (!result.mounted) { + mountStderr.write('[factory] aborting startup: Relayfile workspace mirror could not be resolved.\n') + if (stoppedBySignal) return await waiter.promise + return 1 + } + } catch (error) { + handleWarmMountError(error) + if (stoppedBySignal) return await waiter.promise + return 1 + } + } else { + void warmMount().catch(handleWarmMountError) + } + if (stoppedBySignal) return await waiter.promise await factory.start({ mode: command.mode }) const code = await (deps.waitForStopSignal?.() ?? waiter.promise) return typeof code === 'number' ? code : 0 @@ -678,19 +695,18 @@ async function runFactoryCommand( } } if (command.action === 'run-once') { - await ensureClonePathMounts( + await ensureWorkspaceMount( + mount, mountFn, workspaceId, - config, acceptableMountIds, mountStderr, - debugMountRefreshes, ) writeJson(out, await factory.runOnce({ dryRun: globals.dryRun })) return 0 } if (command.action === 'status') { - writeJson(out, factory.status()) + writeJson(out, await factoryStatusWithMountHealth(factory, mount, config.loop.heartbeatPath, config.loop.heartbeatStaleMs)) return 0 } if (command.action === 'loop-status') { @@ -707,20 +723,22 @@ async function runFactoryCommand( writeJson(out, { killed: heartbeat.pid, signal: 'SIGTERM' }) return 0 } - await ensureClonePathMounts( + await ensureWorkspaceMount( + mount, mountFn, workspaceId, - config, acceptableMountIds, mountStderr, - debugMountRefreshes, ) const removeSignalHandlers = installFactoryStopSignalHandlers(factory, { processLike: deps.stopSignalProcessLike, }) try { const reports = await factory.runLoop({ dryRun: globals.dryRun }) - writeJson(out, { reports, status: factory.status() }) + writeJson(out, { + reports, + status: await factoryStatusWithMountHealth(factory, mount, config.loop.heartbeatPath, config.loop.heartbeatStaleMs), + }) } finally { removeSignalHandlers() await factory.stop() @@ -762,31 +780,27 @@ async function runFactoryCommand( } async function warmStartPathMounts( + mount: MountClient, mountFn: NonNullable, workspaceId: string, config: FactoryConfig, acceptableMountIds?: readonly string[], stderr: Pick = process.stderr, debug = process.env.FACTORY_LOG_LEVEL?.toLowerCase() === 'debug', -): Promise { - const mountOpts = { acceptableWorkspaceIds: acceptableMountIds } - const [daemonRefresh, cloneRefreshes] = await Promise.all([ - ensureMountPath(mountFn, workspaceId, process.cwd(), mountOpts, stderr), - ensureClonePathMounts( - mountFn, - workspaceId, - config, - acceptableMountIds, - stderr, - debug, - false, - ), - ]) - writeMountRefreshSummary( - [...(daemonRefresh ? [daemonRefresh] : []), ...cloneRefreshes], +): Promise { + const result = await ensureWorkspaceMount( + mount, + mountFn, + workspaceId, + acceptableMountIds, stderr, - debug, ) + writeMountRefreshSummary(result.refreshed ? [result.refreshed] : [], stderr, debug) + stderr.write( + `[factory] Relayfile workspace mirror preflight: mounted=${result.mounted ? 1 : 0} ` + + `failed=${result.mounted ? 0 : 1} routedRepos=${new Set(Object.values(config.repos.byLabel)).size}\n`, + ) + return result } async function runStandaloneBabysitCommand( @@ -803,11 +817,7 @@ async function runStandaloneBabysitCommand( const repo = resolveStandaloneBabysitRepo(command.repo, config) const clonePath = standaloneBabysitClonePath(repo, config) const mountFn = resolveLocalMountFn(deps, mount) - const mountOpts = { acceptableWorkspaceIds: acceptableMountIds } - await ensureStandaloneBabysitMount(mountFn, workspaceId, process.cwd(), mountOpts, deps.stderr) - if (clonePath && resolve(clonePath) !== resolve(process.cwd())) { - await ensureStandaloneBabysitMount(mountFn, workspaceId, clonePath, mountOpts, deps.stderr) - } + await ensureWorkspaceMount(mount, mountFn, workspaceId, acceptableMountIds, deps.stderr) const pr = await readStandalonePullRequest( mount, @@ -875,7 +885,7 @@ async function runStandaloneBabysitCommand( maintainerCanModify: pr.maintainerCanModify, }, standaloneBabysitter: { specSource }, - integrationsMountRoot: resolve(process.cwd(), '.integrations'), + integrationsMountRoot: resolveIntegrationsMountRoot(mount), testGuidance, }) const receiptBase = { @@ -915,27 +925,6 @@ async function runStandaloneBabysitCommand( return 0 } -async function ensureStandaloneBabysitMount( - mountFn: NonNullable, - workspaceId: string, - startDir: string, - options: { acceptableWorkspaceIds?: readonly string[] }, - stderr: Pick = process.stderr, -): Promise { - try { - await mountFn(workspaceId, startDir, options) - } catch (error) { - // Terminal scope shortfall: propagate so the command aborts with the - // remediation rather than silently falling back to a read-denied mirror. - if (error instanceof MountAuthScopeError) throw error - const message = error instanceof Error ? error.message : String(error) - stderr.write( - `[factory] warning: could not start relayfile mount for standalone babysitter at ${resolve(startDir)}; ` + - `the agent will use the GitHub CLI fallback: ${message}\n`, - ) - } -} - function resolveStandaloneBabysitRepo(repo: string | undefined, config: FactoryConfig): string { const configured = repo ?? config.repos.default if (!configured) { @@ -966,43 +955,31 @@ function standaloneBabysitClonePath(repo: string, config: FactoryConfig): string } /** - * Ensures the relayfile mount is running at each configured clone path so - * spawned agents can resolve `.integrations` relative to their working - * directory (the checkout path). The mount daemon started at the daemon CWD - * is not automatically accessible from a different directory, and agents need - * these paths for integration writebacks (Slack, GitHub, etc.). + * Ensures exactly one Relayfile mirror for the workspace. Agents receive this + * absolute path in their tasks, so routing more repositories never asks + * Relayfile to re-home the mirror into each checkout. */ -async function ensureClonePathMounts( +async function ensureWorkspaceMount( + mount: MountClient, mountFn: NonNullable, workspaceId: string, - config: FactoryConfig, acceptableMountIds?: readonly string[], stderr: Pick = process.stderr, - debug = process.env.FACTORY_LOG_LEVEL?.toLowerCase() === 'debug', - reportSummary = true, -): Promise { +): Promise { const mountOpts = { acceptableWorkspaceIds: acceptableMountIds } - const daemonCwd = resolve(process.cwd()) - const clonePaths = [...new Set(Object.values(config.clonePaths ?? {}).map((clonePath) => resolve(clonePath)))] - .filter((clonePath) => clonePath !== daemonCwd) - const refreshedStaleMounts: RefreshedStaleMount[] = [] - let nextIndex = 0 - const mountNext = async (): Promise => { - while (nextIndex < clonePaths.length) { - const resolved = clonePaths[nextIndex++]! - const refreshed = await ensureMountPath(mountFn, workspaceId, resolved, mountOpts, stderr) - if (refreshed) refreshedStaleMounts.push(refreshed) - } - } - await Promise.all(Array.from( - { length: Math.min(CLONE_MOUNT_PREFLIGHT_CONCURRENCY, clonePaths.length) }, - mountNext, - )) - if (reportSummary) writeMountRefreshSummary(refreshedStaleMounts, stderr, debug) - return refreshedStaleMounts + const localDir = mount.getLocalMountRoot?.() + return ensureMountPath( + mountFn, + workspaceId, + localDir ? dirname(localDir) : process.cwd(), + mountOpts, + stderr, + localDir, + ) } type RefreshedStaleMount = { path: string; reason?: string } +type WorkspaceMountPreflight = { mounted: boolean; refreshed?: RefreshedStaleMount } async function ensureMountPath( mountFn: NonNullable, @@ -1010,27 +987,78 @@ async function ensureMountPath( path: string, mountOpts: { acceptableWorkspaceIds?: readonly string[] }, stderr: Pick, -): Promise { - const resolved = resolve(path) - const statePath = join(resolved, '.integrations', '.relay', 'state.json') + localDir = join(resolve(path), '.integrations'), +): Promise { + const statePath = join(localDir, '.relay', 'state.json') const staleBefore = checkMountStaleness(statePath, workspaceId, mountOpts.acceptableWorkspaceIds) try { - await mountFn(workspaceId, resolved, { + await mountFn(workspaceId, resolve(path), { ...mountOpts, ...(staleBefore.stale ? { suppressStaleRefreshLogs: true } : {}), }) if (staleBefore.stale && !checkMountStaleness(statePath, workspaceId, mountOpts.acceptableWorkspaceIds).stale) { - return { path: resolved, reason: staleBefore.reason } + return { mounted: true, refreshed: { path: localDir, reason: staleBefore.reason } } } + return { mounted: true } } catch (error) { // A scope shortfall is terminal and identical across every clone path; // propagate it so startup fails fast with one remediation instead of // logging the same unfixable warning per path. if (error instanceof MountAuthScopeError) throw error const message = error instanceof Error ? error.message : String(error) - stderr.write(`[factory] warning: could not start relayfile mount at ${resolved}: ${message}\n`) + stderr.write(`[factory] warning: could not start Relayfile workspace mirror at ${localDir}: ${message}\n`) + } + return { mounted: false } +} + +function resolveIntegrationsMountRoot(mount: MountClient): string { + return mount.getLocalMountRoot?.() ?? resolve(process.cwd(), '.integrations') +} + +async function factoryStatusWithMountHealth( + factory: Factory, + mount: MountClient, + heartbeatPath: string, + heartbeatStaleMs: number, +): Promise & { + localMountDegraded?: boolean + localMountDegradedReason?: string + localMountRoot?: string + /** Local mirror liveness, independently of whether the daemon is listening. */ + localMountEventFeed?: { + state: 'healthy' | 'degraded' + livenessSignal: '.integrations/.relay/state.json' + reason?: string + root?: string + } +}> { + const status = factory.status() + const heartbeat = await readFactoryLoopHeartbeat(heartbeatPath) + const liveness = checkFactoryLoopLiveness(heartbeat, { staleMs: heartbeatStaleMs }) + const eventListener = liveness.ok + ? heartbeat?.eventListener ?? { + state: 'unknown' as const, + reason: 'running daemon heartbeat does not report event listener state', + } + : { + state: 'not-listening' as const, + reason: liveness.reason, + } + const health = mount.getLocalMountHealth?.() + if (!health) return { ...status, eventListener } + return { + ...status, + eventListener, + localMountDegraded: health.degraded, + ...(health.reason ? { localMountDegradedReason: health.reason } : {}), + ...(health.localDir ? { localMountRoot: health.localDir } : {}), + localMountEventFeed: { + state: health.degraded ? 'degraded' : 'healthy', + livenessSignal: '.integrations/.relay/state.json', + ...(health.reason ? { reason: health.reason } : {}), + ...(health.localDir ? { root: health.localDir } : {}), + }, } - return undefined } function writeMountRefreshSummary( @@ -1539,6 +1567,7 @@ async function buildMount( let mount: MountClient mount = await (deps.cloudMountFromConfig ?? RelayfileCloudMountClient.fromConfig)({ workspaceId: loaded.config.workspaceId, + localMountRoot: loaded.config.localMountRoot, logger: observability.logger, onLocalMountHealth: observability.onLocalMountHealth, isAllowedDraft: (path, content, opts) => isAllowedFactoryDraft(path, content, opts, mount, loaded.config), diff --git a/src/config/schema.test.ts b/src/config/schema.test.ts index d99450a..ad99064 100644 --- a/src/config/schema.test.ts +++ b/src/config/schema.test.ts @@ -237,6 +237,21 @@ describe('FactoryConfigSchema', () => { expect(github.issueSource).toBe('github') }) + it('normalizes an explicit workspace mirror root and rejects unsafe values', () => { + const parsed = FactoryConfigSchema.parse({ + localMountRoot: ' /work/chief/.integrations ', + repos: { default: 'AgentWorkforce/factory' }, + }) + + expect(parsed.localMountRoot).toBe('/work/chief/.integrations') + for (const localMountRoot of ['', ' ', './.integrations', 'relative/mirror']) { + expect(() => FactoryConfigSchema.parse({ + localMountRoot, + repos: { default: 'AgentWorkforce/factory' }, + })).toThrow() + } + }) + it.each(['app', 'user', 'auto'] as const)('accepts github.identity %s', (identity) => { const parsed = FactoryConfigSchema.parse({ repos: { default: 'AgentWorkforce/factory' }, diff --git a/src/config/schema.ts b/src/config/schema.ts index 81c2dbf..aed87a5 100644 --- a/src/config/schema.ts +++ b/src/config/schema.ts @@ -1,5 +1,5 @@ import { homedir } from 'node:os' -import { join } from 'node:path' +import { isAbsolute, join } from 'node:path' import { z } from 'zod' @@ -234,6 +234,14 @@ const WorkspaceConfigObjectSchema = z.object({ // falling back to the SDK's built-in default. Set it only to pin a non-active // workspace. See resolveFactoryWorkspace() in relayfile-cloud-mount-client.ts. workspaceId: z.string().optional(), + // Optional exact root of this workspace's single Relayfile mirror. When + // omitted Factory reads Relayfile's existing registration. This is a + // workspace-scoped escape hatch, never a request to re-home per checkout. + localMountRoot: z.string() + .trim() + .min(1) + .refine(isAbsolute, 'localMountRoot must be an absolute path') + .optional(), subscription: subscriptionSchema, liveSubscription: liveSubscriptionSchema, dispatch: dispatchSchema, diff --git a/src/mount/local-mount-preflight.test.ts b/src/mount/local-mount-preflight.test.ts index 23e0b0f..54eed0c 100644 --- a/src/mount/local-mount-preflight.test.ts +++ b/src/mount/local-mount-preflight.test.ts @@ -48,6 +48,29 @@ describe('ensureLocalMount', () => { }) }) + it('uses the exact registered root when the mirror is not named .integrations', async () => { + await withTempDir(async (dir) => { + const localDir = join(dir, 'relayfile-mirror') + const startMount = vi.fn(async () => { + const stateDir = join(localDir, '.relay') + await mkdir(stateDir, { recursive: true }) + await writeFile(join(stateDir, 'state.json'), JSON.stringify({ + workspaceId: 'rw_test', + lastReconcileAt: new Date().toISOString(), + pid: process.pid, + })) + }) + + await expect(ensureLocalMount('rw_test', dir, { + localDir, + startMount, + stateWaitTimeoutMs: 100, + stateWaitPollMs: 1, + })).resolves.toBeUndefined() + expect(startMount).toHaveBeenCalledTimes(1) + }) + }) + it('accepts a fresh SDK mount state that intentionally omits a daemon pid', async () => { await withTempDir(async (dir) => { const startMount = vi.fn(async () => { @@ -99,6 +122,7 @@ describe('ensureLocalMount', () => { pid: process.pid, }) }) + const stderr = vi.spyOn(process.stderr, 'write').mockImplementation(() => true) await expect(ensureLocalMount('rw_test', dir, { startMount, @@ -106,6 +130,9 @@ describe('ensureLocalMount', () => { stateWaitPollMs: 1, })).resolves.toBeUndefined() expect(startMount).toHaveBeenCalledTimes(1) + expect(stderr).toHaveBeenCalledWith(expect.stringContaining( + `local mount at ${join(dir, '.integrations')} is stale`, + )) }) }) diff --git a/src/mount/local-mount-preflight.ts b/src/mount/local-mount-preflight.ts index 3235ece..7744425 100644 --- a/src/mount/local-mount-preflight.ts +++ b/src/mount/local-mount-preflight.ts @@ -10,8 +10,6 @@ import { readMountAuthErrorFromState, } from './mount-auth-error' -const STATE_FILE = '.integrations/.relay/state.json' - // How long to wait for a freshly-spawned mount to write a valid state.json // (workspace match + a fresh reconcile timestamp). A CLI mount over a large `.integrations` tree // can take well over 10s to complete its FIRST reconcile (early cycles hit @@ -22,6 +20,8 @@ const STATE_FILE = '.integrations/.relay/state.json' const DEFAULT_STATE_READY_TIMEOUT_MS = 60_000 export interface EnsureLocalMountOptions extends LocalMountOptions { + /** Exact registered mirror root, for mirrors not conventionally named `.integrations`. */ + localDir?: string /** * Starts an authenticated mount through the Relayfile SDK. Credential minting, * binary resolution, and launch details deliberately stay outside this @@ -35,7 +35,8 @@ export async function ensureLocalMount( startDir: string, options: EnsureLocalMountOptions, ): Promise { - const stateFilePath = join(startDir, STATE_FILE) + const localDir = options.localDir ?? join(startDir, '.integrations') + const stateFilePath = join(localDir, '.relay', 'state.json') if (!(await isMountStatePresent(stateFilePath))) { try { @@ -82,6 +83,10 @@ export async function ensureLocalMount( const suffix = staleness.reason !== undefined ? ` (${staleness.reason})` : '' const manualHint = 'Restart Factory after restoring the Agent Relay Cloud session' + // Include the registered target in an operator-facing refresh log. This is + // intentionally the state path's parent, not the current Factory checkout: + // a stale mirror must heal where Relayfile registered it. + const mountTarget = localDir // Stale AND under-scoped: refreshing cannot help. Fail fast and terminal so // the supervisor stops retrying and startup surfaces one actionable error. @@ -90,14 +95,14 @@ export async function ensureLocalMount( } if (options.refreshStaleMount === false) { - process.stderr.write(`[factory] local mount is stale${suffix}; writeback may not propagate. ${manualHint}\n`) + process.stderr.write(`[factory] local mount at ${mountTarget} is stale${suffix}; writeback may not propagate. ${manualHint}\n`) return } // Self-heal through the same SDK-authenticated launch path used for first // start, rather than silently shipping writebacks into a stale mirror. if (!options.suppressStaleRefreshLogs) { - process.stderr.write(`[factory] local mount is stale${suffix}; refreshing\n`) + process.stderr.write(`[factory] local mount at ${mountTarget} is stale${suffix}; refreshing\n`) } try { await options.startMount() @@ -109,7 +114,7 @@ export async function ensureLocalMount( options.acceptableWorkspaceIds, ) if (!options.suppressStaleRefreshLogs) { - process.stderr.write('[factory] local mount refreshed\n') + process.stderr.write(`[factory] local mount at ${mountTarget} refreshed\n`) } } catch (error) { if (error instanceof MountAuthScopeError) throw error @@ -124,7 +129,7 @@ export async function ensureLocalMount( cause: error, }) } - process.stderr.write(`[factory] local mount is stale${suffix} and auto-refresh failed (${reason}); writeback may not propagate. ${manualHint}\n`) + process.stderr.write(`[factory] local mount at ${mountTarget} is stale${suffix} and auto-refresh failed (${reason}); writeback may not propagate. ${manualHint}\n`) } } diff --git a/src/mount/relayfile-binary.test.ts b/src/mount/relayfile-binary.test.ts index fe72966..9e392d3 100644 --- a/src/mount/relayfile-binary.test.ts +++ b/src/mount/relayfile-binary.test.ts @@ -3,7 +3,13 @@ import { join } from 'node:path' import { tmpdir } from 'node:os' import { afterEach, describe, expect, it, vi } from 'vitest' -import { checkMountStaleness } from './relayfile-binary' +import { + checkMountStaleness, + RELAYFILE_SYNC_INTERVAL_MS, + STALE_RECONCILE_MS, + STALE_RECONCILE_INTERVALS, + staleReconcileMs, +} from './relayfile-binary' afterEach(() => { vi.restoreAllMocks() @@ -20,7 +26,13 @@ async function withTempDir(fn: (dir: string) => Promise): Promise { async function writeState( dir: string, - state: { workspaceId?: string; lastReconcileAt?: string; pid?: number; daemon?: { pid?: number } }, + state: { + workspaceId?: string + lastReconcileAt?: string + intervalMs?: number + pid?: number + daemon?: { pid?: number } + }, ): Promise { const statePath = join(dir, 'state.json') await writeFile(statePath, JSON.stringify(state), 'utf8') @@ -93,6 +105,53 @@ describe('checkMountStaleness', () => { }) }) + it('uses a multiple of the Relayfile poll interval as its stale threshold', async () => { + await withTempDir(async (dir) => { + const statePath = await writeState(dir, { + workspaceId: 'rw_test', + lastReconcileAt: new Date( + Date.now() - (RELAYFILE_SYNC_INTERVAL_MS * STALE_RECONCILE_INTERVALS) - 1, + ).toISOString(), + pid: process.pid, + }) + + expect(checkMountStaleness(statePath, 'rw_test')).toMatchObject({ + stale: true, + reason: expect.stringMatching(/^last reconcile \d+m ago$/u), + }) + }) + }) + + it('uses the registered non-default poll interval rather than falsely staling a healthy slow mirror', async () => { + await withTempDir(async (dir) => { + const intervalMs = 2 * 60 * 1000 + const statePath = await writeState(dir, { + workspaceId: 'rw_test', + intervalMs, + // Past the 90s default but still within three registered 2m intervals. + lastReconcileAt: new Date(Date.now() - STALE_RECONCILE_MS - 1).toISOString(), + pid: process.pid, + }) + + expect(staleReconcileMs(intervalMs)).toBe(intervalMs * STALE_RECONCILE_INTERVALS) + expect(checkMountStaleness(statePath, 'rw_test')).toEqual({ stale: false, pid: process.pid }) + }) + }) + + it('falls back to the default interval when state.json has an invalid cadence', async () => { + await withTempDir(async (dir) => { + const statePath = await writeState(dir, { + workspaceId: 'rw_test', + intervalMs: 0, + lastReconcileAt: new Date(Date.now() - STALE_RECONCILE_MS - 1).toISOString(), + pid: process.pid, + }) + + expect(staleReconcileMs(0)).toBe(STALE_RECONCILE_MS) + expect(checkMountStaleness(statePath, 'rw_test')).toMatchObject({ stale: true }) + }) + }) + it('marks a dead mount process stale', async () => { await withTempDir(async (dir) => { const statePath = await writeState(dir, { diff --git a/src/mount/relayfile-binary.ts b/src/mount/relayfile-binary.ts index 29333e6..16241eb 100644 --- a/src/mount/relayfile-binary.ts +++ b/src/mount/relayfile-binary.ts @@ -1,10 +1,29 @@ import { readFileSync } from 'node:fs' -const STALE_RECONCILE_MS = 15 * 60 * 1000 +// Relayfile's poll mirror reconciles every 30 seconds by default. Three +// intervals allow one missed poll and ordinary filesystem jitter, while still +// making a stalled projection visible within 90 seconds rather than hours. +export const RELAYFILE_SYNC_INTERVAL_MS = 30 * 1000 +export const STALE_RECONCILE_INTERVALS = 3 +export const STALE_RECONCILE_MS = RELAYFILE_SYNC_INTERVAL_MS * STALE_RECONCILE_INTERVALS + +/** Use the registered mirror cadence when available; fall back to Relayfile's default. */ +export function staleReconcileMs(intervalMs: unknown): number { + const registeredIntervalMs = typeof intervalMs === 'number' && + Number.isFinite(intervalMs) && + intervalMs >= 1_000 + ? Math.floor(intervalMs) + : RELAYFILE_SYNC_INTERVAL_MS + return registeredIntervalMs * STALE_RECONCILE_INTERVALS +} type MountState = { workspaceId?: unknown lastReconcileAt?: unknown + // Relayfile writes the active poll cadence in state.json. It is part of the + // mount's liveness contract, so the stale threshold must follow it rather + // than assuming the default cadence for every registered mirror. + intervalMs?: unknown // The mount process pid. Older mounts wrote a top-level `pid`; SDK-launched // mounts record it under `daemon.pid` instead. Either may be absent. pid?: unknown @@ -58,7 +77,7 @@ export function checkMountStaleness( } const ageMs = Date.now() - lastReconcileAt - if (ageMs > STALE_RECONCILE_MS) { + if (ageMs > staleReconcileMs(parsed.intervalMs)) { return { stale: true, reason: `last reconcile ${Math.floor(ageMs / 60000)}m ago`, diff --git a/src/mount/relayfile-cloud-mount-client.test.ts b/src/mount/relayfile-cloud-mount-client.test.ts index f8050f5..4dcfda8 100644 --- a/src/mount/relayfile-cloud-mount-client.test.ts +++ b/src/mount/relayfile-cloud-mount-client.test.ts @@ -389,6 +389,7 @@ describe('RelayfileCloudMountClient', () => { })).resolves.toBeUndefined() expect(localMountPreflight).toHaveBeenCalledWith('rw_test', '/work/repo', expect.objectContaining({ + localDir: join('/work/repo', '.integrations'), acceptableWorkspaceIds: ['cloud-workspace-uuid'], stateWaitTimeoutMs: 3210, startMount: expect.any(Function), @@ -410,6 +411,225 @@ describe('RelayfileCloudMountClient', () => { expect(stop).toHaveBeenCalledTimes(1) }) + it('passes an exact nonstandard registered mirror root to local preflight', async () => { + const localDir = '/work/chief/relayfile-mirror' + const fake = new FakeRelayFileClient() + const handle = { + workspaceId: 'cloud-workspace-uuid', + client: vi.fn(() => fake), + getToken: vi.fn(async () => 'delegated-relayfile-token'), + info: { relayfileUrl: 'https://relayfile.example' }, + } + const localMountPreflight = vi.fn(async ( + _workspaceId: string, + _startDir: string, + options: { startMount: () => Promise }, + ) => options.startMount()) + const mount = new RelayfileCloudMountClient({ + workspaceId: 'rw_test', + client: fake, + relayfileSetup: { joinWorkspace: vi.fn(), ensureMountedWorkspace: vi.fn(async () => ({ stop: async () => {} })) }, + relayfileWorkspace: handle, + localMountRoot: localDir, + localMountPreflight, + }) + + try { + await mount.ensureLocalMount('/work/unrelated-repository') + expect(localMountPreflight).toHaveBeenCalledWith('rw_test', '/work/chief', expect.objectContaining({ localDir })) + } finally { + await mount.dispose() + } + }) + + it('looks up a configured workspace mirror only once during fromConfig', async () => { + const fake = new FakeRelayFileClient() + const resolver = vi.fn(() => undefined) + const mount = await RelayfileCloudMountClient.fromConfig({ + workspaceId: 'rw_test', + cloudSessionProvider: vi.fn(async () => cloudSession(storedAuth())), + relayfileSetupFactory: vi.fn(() => ({ + joinWorkspace: vi.fn(async () => ({ + workspaceId: 'cloud-workspace-uuid', + client: () => fake, + getToken: async () => 'delegated-relayfile-token', + info: { relayfileUrl: 'https://relayfile.example' }, + })), + })), + workspaceMirrorResolver: resolver, + }) + + expect(resolver).toHaveBeenCalledTimes(1) + expect(resolver).toHaveBeenCalledWith(['rw_test', 'cloud-workspace-uuid']) + await mount.dispose() + }) + + it('uses one registered workspace mirror even when callers name different repository checkouts', async () => { + const root = await mkdtemp(join(tmpdir(), 'factory-shared-workspace-mirror-')) + const localDir = join(root, 'chief', '.integrations') + const fake = new FakeRelayFileClient() + const handle = { + workspaceId: 'cloud-workspace-uuid', + client: vi.fn(() => fake), + getToken: vi.fn(async () => 'delegated-relayfile-token'), + info: { relayfileUrl: 'https://relayfile.example' }, + } + const stop = vi.fn(async () => {}) + const ensureMountedWorkspace = vi.fn(async () => ({ stop })) + const localMountPreflight = vi.fn(async ( + _workspaceId: string, + _startDir: string, + options: { startMount: () => Promise }, + ) => options.startMount()) + const mount = new RelayfileCloudMountClient({ + workspaceId: 'rw_shared', + client: fake, + relayfileSetup: { joinWorkspace: vi.fn(), ensureMountedWorkspace }, + relayfileWorkspace: handle, + localMountRoot: localDir, + localMountPreflight, + }) + + try { + await Promise.all([ + mount.ensureLocalMount(join(root, 'repo-a')), + mount.ensureLocalMount(join(root, 'repo-b')), + ]) + + expect(mount.getLocalMountRoot()).toBe(localDir) + expect(ensureMountedWorkspace).toHaveBeenCalledTimes(1) + expect(ensureMountedWorkspace).toHaveBeenCalledWith(expect.objectContaining({ localDir })) + expect(localMountPreflight).toHaveBeenCalledWith('rw_shared', join(root, 'chief'), expect.any(Object)) + } finally { + await mount.dispose() + await rm(root, { recursive: true, force: true }) + } + }) + + it('resolves a direct client mirror through the cloud workspace identifier alias', async () => { + const fake = new FakeRelayFileClient() + const resolver = vi.fn((workspaceIds: readonly string[]) => + workspaceIds.includes('cloud-workspace-uuid') ? '/work/chief/.integrations' : undefined) + const mount = new RelayfileCloudMountClient({ + workspaceId: 'rw_shared', + client: fake, + relayfileSetup: { joinWorkspace: vi.fn(), ensureMountedWorkspace: vi.fn() }, + relayfileWorkspace: { + workspaceId: 'cloud-workspace-uuid', + client: () => fake, + getToken: async () => 'delegated-relayfile-token', + info: { relayfileUrl: 'https://relayfile.example' }, + }, + workspaceMirrorResolver: resolver, + }) + + expect(resolver).toHaveBeenCalledWith(['rw_shared', 'cloud-workspace-uuid']) + expect(mount.getLocalMountRoot()).toBe('/work/chief/.integrations') + await mount.dispose() + }) + + it('uses the registered root reported by Relayfile instead of re-homing an unresolved fallback', async () => { + const root = await mkdtemp(join(tmpdir(), 'factory-admission-registered-mirror-')) + const fallbackDir = join(root, 'first-checkout', '.integrations') + const registeredDir = join(root, 'chief', '.integrations') + const fake = new FakeRelayFileClient() + const stop = vi.fn(async () => {}) + const ensureMountedWorkspace = vi.fn(async ({ localDir }: { localDir: string }) => { + if (localDir === fallbackDir) { + throw new Error( + `workspace rw_shared is already mirrored at ${registeredDir}; refusing to silently re-home it to ${fallbackDir}`, + ) + } + return { stop } + }) + const localMountPreflight = vi.fn(async ( + _workspaceId: string, + _startDir: string, + options: { startMount: () => Promise }, + ) => options.startMount()) + const mount = new RelayfileCloudMountClient({ + workspaceId: 'rw_shared', + client: fake, + relayfileSetup: { joinWorkspace: vi.fn(), ensureMountedWorkspace }, + relayfileWorkspace: { + workspaceId: 'cloud-workspace-uuid', + client: () => fake, + getToken: async () => 'delegated-relayfile-token', + info: { relayfileUrl: 'https://relayfile.example' }, + }, + localMountPreflight, + }) + + try { + await Promise.all([ + mount.ensureLocalMount(join(root, 'first-checkout')), + mount.ensureLocalMount(join(root, 'second-checkout')), + ]) + + expect(ensureMountedWorkspace).toHaveBeenNthCalledWith(1, expect.objectContaining({ localDir: fallbackDir })) + expect(ensureMountedWorkspace).toHaveBeenNthCalledWith(2, expect.objectContaining({ localDir: registeredDir })) + expect(ensureMountedWorkspace).toHaveBeenCalledTimes(2) + expect(mount.getLocalMountRoot()).toBe(registeredDir) + } finally { + await mount.dispose() + await rm(root, { recursive: true, force: true }) + } + }) + + it('reports a stale registered mirror and refreshes that same directory', async () => { + const root = await mkdtemp(join(tmpdir(), 'factory-stale-registered-mirror-')) + const localDir = join(root, 'registered', '.integrations') + await mkdir(join(localDir, '.relay'), { recursive: true }) + await writeFile(join(localDir, '.relay', 'state.json'), JSON.stringify({ + workspaceId: 'cloud-workspace-uuid', + lastReconcileAt: new Date(Date.now() - 5 * 60_000).toISOString(), + pid: process.pid, + })) + const fake = new FakeRelayFileClient() + const ensureMountedWorkspace = vi.fn(async () => ({ stop: async () => {} })) + const localMountPreflight = vi.fn(async ( + _workspaceId: string, + _startDir: string, + options: { startMount: () => Promise }, + ) => { + await options.startMount() + await writeFile(join(localDir, '.relay', 'state.json'), JSON.stringify({ + workspaceId: 'cloud-workspace-uuid', + lastReconcileAt: new Date().toISOString(), + pid: process.pid, + })) + }) + const mount = new RelayfileCloudMountClient({ + workspaceId: 'rw_shared', + client: fake, + relayfileSetup: { joinWorkspace: vi.fn(), ensureMountedWorkspace }, + relayfileWorkspace: { + workspaceId: 'cloud-workspace-uuid', + client: () => fake, + getToken: async () => 'delegated-relayfile-token', + info: { relayfileUrl: 'https://relayfile.example' }, + }, + localMountRoot: localDir, + localMountPreflight, + }) + + try { + expect(mount.getLocalMountHealth()).toMatchObject({ + degraded: true, + reason: expect.stringMatching(/^last reconcile \d+m ago$/u), + localDir, + }) + + await mount.ensureLocalMount(join(root, 'unrelated-repository')) + + expect(ensureMountedWorkspace).toHaveBeenCalledWith(expect.objectContaining({ localDir })) + expect(mount.getLocalMountHealth()).toEqual({ degraded: false, localDir }) + } finally { + await mount.dispose() + await rm(root, { recursive: true, force: true }) + } + }) + it('reports and supervises an initial SDK mount failure with no prior state file', async () => { vi.useFakeTimers() const fake = new FakeRelayFileClient() @@ -537,7 +757,7 @@ describe('RelayfileCloudMountClient', () => { await mount.dispose() }) - it('bounds mount work across checkouts to prevent refresh storms', async () => { + it('coalesces routed checkouts onto one workspace-mirror mount operation', async () => { const fake = new FakeRelayFileClient() let active = 0 let maximumActive = 0 @@ -563,17 +783,16 @@ describe('RelayfileCloudMountClient', () => { info: { relayfileUrl: 'https://relayfile.example' }, }, localMountPreflight, - localMountMaxConcurrency: 2, }) const checks = Array.from({ length: 6 }, (_, index) => mount.ensureLocalMount(`/work/repo-${index}`)) - await vi.waitFor(() => expect(localMountPreflight).toHaveBeenCalledTimes(2)) - expect(maximumActive).toBe(2) + await vi.waitFor(() => expect(localMountPreflight).toHaveBeenCalledTimes(1)) + expect(maximumActive).toBe(1) releasePreflights() await Promise.all(checks) - expect(localMountPreflight).toHaveBeenCalledTimes(6) - expect(maximumActive).toBe(2) + expect(localMountPreflight).toHaveBeenCalledTimes(1) + expect(maximumActive).toBe(1) await mount.dispose() }) diff --git a/src/mount/relayfile-cloud-mount-client.ts b/src/mount/relayfile-cloud-mount-client.ts index fb45002..0d61125 100644 --- a/src/mount/relayfile-cloud-mount-client.ts +++ b/src/mount/relayfile-cloud-mount-client.ts @@ -22,7 +22,7 @@ import { } from '@relayfile/sdk' import { RelayfileSetup } from '@relayfile/sdk/cli' import { existsSync } from 'node:fs' -import { join, resolve } from 'node:path' +import { isAbsolute, join, resolve } from 'node:path' import type { EventPage, @@ -30,6 +30,7 @@ import type { FactoryIntegrationProvider, GithubConnectionWrite, LocalMountOptions, + LocalMountHealth, Logger, MountClient, ProviderSyncStatus, @@ -51,6 +52,7 @@ import { } from './local-mount-preflight' import { checkMountStaleness } from './relayfile-binary' import { MountAuthScopeError } from './mount-auth-error' +import { resolveRegisteredWorkspaceMirror } from './workspace-mirror' const DEFAULT_WORKSPACE_ID = 'rw_7ccfea89' const DEFAULT_AGENT_NAME = 'agent-relay-factory' @@ -210,6 +212,12 @@ export interface RelayfileCloudMountClientConfig { localMountHealthIntervalMs?: number /** Internal mount-work concurrency override for tests. */ localMountMaxConcurrency?: number + /** Explicit registered mirror root; never inferred from a routed checkout. */ + localMountRoot?: string + /** Read-only registration lookup override for tests and alternate runtimes. */ + workspaceMirrorResolver?: (workspaceIds: readonly string[]) => string | undefined + /** Internal: fromConfig already attempted the registration lookup, including no-match. */ + skipRegisteredMirrorLookup?: boolean isAllowedDraft?: (path: string, content: unknown, opts?: { guarded?: boolean }) => boolean | Promise isAllowedDelete?: (path: string, currentContent: unknown) => boolean | Promise } @@ -266,6 +274,7 @@ export class RelayfileCloudMountClient implements MountClient { readonly #localMountPreflight: LocalMountPreflight readonly #localMountAgentName: string readonly #localMountScopes: string[] + #localMountRoot?: string readonly #localMounts = new Map() readonly #localMountSupervisions = new Map { - const localDir = join(resolve(startDir), '.integrations') - return this.#runLocalMountOperation(localDir, () => this.#ensureLocalMount(startDir, localDir, options)) + // A Relayfile workspace has one registered local mirror. Do not derive a + // new one from every routed checkout: that asks Relayfile to re-home the + // workspace and makes all but the first route fail. The fallback is only + // invoked once per Factory command rather than once per repository. + const canDiscoverRegisteredMirror = this.#localMountRoot === undefined + const localDir = this.#localMountRoot ?? join(resolve(startDir), '.integrations') + this.#localMountRoot = localDir + return this.#runLocalMountOperation( + localDir, + () => this.#ensureLocalMountWithRegisteredFallback(localDir, options, canDiscoverRegisteredMirror), + ) + } + + getLocalMountRoot(): string | undefined { + return this.#localMountRoot } - async #ensureLocalMount(startDir: string, localDir: string, options: LocalMountOptions): Promise { + getLocalMountHealth(): LocalMountHealth { + const localDir = this.#localMountRoot + if (!localDir) { + return { degraded: true, reason: 'Relayfile workspace mirror is not registered' } + } + const statePath = join(localDir, '.relay', 'state.json') + if (!existsSync(statePath)) { + return { degraded: true, reason: `mount state is missing at ${statePath}`, localDir } + } + const staleness = checkMountStaleness(statePath, this.workspaceId, this.#acceptableWorkspaceIds()) + return { + degraded: staleness.stale, + ...(staleness.reason ? { reason: staleness.reason } : {}), + localDir, + } + } + + async #ensureLocalMount(localDir: string, options: LocalMountOptions): Promise { const setup = this.#relayfileSetup const workspace = this.#relayfileWorkspace if (!setup?.ensureMountedWorkspace || !workspace) { @@ -382,10 +431,7 @@ export class RelayfileCloudMountClient implements MountClient { } const ensureMountedWorkspace = setup.ensureMountedWorkspace.bind(setup) - const acceptableWorkspaceIds = new Set(options.acceptableWorkspaceIds ?? []) - if (workspace.workspaceId && workspace.workspaceId !== this.workspaceId) { - acceptableWorkspaceIds.add(workspace.workspaceId) - } + const acceptableWorkspaceIds = new Set(this.#acceptableWorkspaceIds(options.acceptableWorkspaceIds)) const launch = (): Promise => ensureMountedWorkspace({ workspace, localDir, @@ -404,7 +450,7 @@ export class RelayfileCloudMountClient implements MountClient { ...(options.stateWaitTimeoutMs === undefined ? {} : { readyTimeoutMs: options.stateWaitTimeoutMs }), }) this.#localMountSupervisions.set(localDir, { - startDir, + startDir: join(localDir, '..'), options: { ...options, acceptableWorkspaceIds: [...acceptableWorkspaceIds] }, launch, suggestedRefreshAtMs: this.#localMountSupervisions.get(localDir)?.suggestedRefreshAtMs, @@ -417,8 +463,9 @@ export class RelayfileCloudMountClient implements MountClient { if (staleBefore?.stale) this.#markLocalMountDegraded(localDir, 'mount_stale') try { - await this.#localMountPreflight(this.workspaceId, startDir, { + await this.#localMountPreflight(this.workspaceId, join(localDir, '..'), { ...options, + localDir, acceptableWorkspaceIds: [...acceptableWorkspaceIds], startMount: async () => { await this.#replaceLocalMount(localDir, launch) @@ -445,6 +492,42 @@ export class RelayfileCloudMountClient implements MountClient { this.#scheduleLocalMountHealthCheck(localDir) } + async #ensureLocalMountWithRegisteredFallback( + localDir: string, + options: LocalMountOptions, + canDiscoverRegisteredMirror: boolean, + ): Promise { + try { + await this.#ensureLocalMount(localDir, options) + return + } catch (error) { + // Older Relayfile installations do not persist the registration in a + // local file we can read. The mount admission response is nevertheless + // authoritative about the already-registered root. This retry remains + // inside the shared operation, so concurrent callers all await the same + // recovery rather than racing to re-home a checkout. + const registeredRoot = canDiscoverRegisteredMirror ? registeredMirrorFromMountError(error) : undefined + if (!registeredRoot || registeredRoot === localDir) throw error + this.#clearLocalMountHealthCheck(localDir) + this.#localMountSupervisions.delete(localDir) + this.#degradedLocalMounts.delete(localDir) + this.#authDegradedLocalMounts.delete(localDir) + await this.#ensureLocalMount(registeredRoot, options) + this.#localMountRoot = registeredRoot + } + } + + #acceptableWorkspaceIds(extra: readonly string[] = []): string[] { + const workspace = this.#relayfileWorkspace?.workspaceId + return [...new Set([...extra, ...(workspace && workspace !== this.workspaceId ? [workspace] : [])])] + } + + #clearLocalMountHealthCheck(localDir: string): void { + const timer = this.#localMountHealthTimers.get(localDir) + if (timer) clearTimeout(timer) + this.#localMountHealthTimers.delete(localDir) + } + #runLocalMountOperation(localDir: string, operation: () => Promise): Promise { const existing = this.#localMountOperations.get(localDir) if (existing) return existing @@ -921,6 +1004,19 @@ const serializeContent = (content: unknown): { content: string; contentType: str } } +/** + * Relayfile's single-mirror admission check names the registered directory in + * its refusal. Treat that directory as a read-only registration lookup: the + * retry asks for the already registered root and never supplies `--rehome`. + */ +function registeredMirrorFromMountError(error: unknown): string | undefined { + const message = error instanceof Error ? error.message : String(error) + const match = /\balready mirrored at\s+(.+?)(?:;|\n|$)/iu.exec(message) + if (!match?.[1]) return undefined + const localDir = match[1].trim().replace(/^["']|["']$/gu, '') + return isAbsolute(localDir) ? resolve(localDir) : undefined +} + const isHttpStatus = (error: unknown, status: number): boolean => { const record = error !== null && typeof error === 'object' ? error as Record : undefined return record?.status === status || record?.statusCode === status diff --git a/src/mount/workspace-mirror.test.ts b/src/mount/workspace-mirror.test.ts new file mode 100644 index 0000000..671270b --- /dev/null +++ b/src/mount/workspace-mirror.test.ts @@ -0,0 +1,109 @@ +import { mkdir, mkdtemp, rm, writeFile } from 'node:fs/promises' +import { tmpdir } from 'node:os' +import { join } from 'node:path' +import { describe, expect, it } from 'vitest' + +import { resolveRegisteredWorkspaceMirror } from './workspace-mirror' + +async function withTempHome(fn: (home: string) => Promise): Promise { + const home = await mkdtemp(join(tmpdir(), 'factory-workspace-mirror-')) + try { + return await fn(home) + } finally { + await rm(home, { recursive: true, force: true }) + } +} + +describe('resolveRegisteredWorkspaceMirror', () => { + it('uses the workspace registration rather than a caller checkout path', async () => { + await withTempHome(async (home) => { + const mirror = join(home, 'shared', '.integrations') + await mkdir(join(home, '.relayfile'), { recursive: true }) + await writeFile(join(home, '.relayfile', 'workspaces.json'), JSON.stringify({ + workspaces: [{ id: 'rw_shared', localDir: mirror }], + })) + + expect(resolveRegisteredWorkspaceMirror(['rw_shared'], home)).toEqual({ + localDir: mirror, + source: 'workspace-registry', + }) + }) + }) + + it('anchors a legacy relative workspace registry root to the Relayfile home', async () => { + await withTempHome(async (home) => { + await mkdir(join(home, '.relayfile'), { recursive: true }) + await writeFile(join(home, '.relayfile', 'workspaces.json'), JSON.stringify({ + workspaces: [{ id: 'rw_shared', localDir: '.integrations' }], + })) + + expect(resolveRegisteredWorkspaceMirror(['rw_shared'], home)).toEqual({ + localDir: join(home, '.integrations'), + source: 'workspace-registry', + }) + }) + }) + + it('falls back to the Relayfile private mount registration', async () => { + await withTempHome(async (home) => { + const mirror = join(home, 'registered', '.integrations') + const stateDir = join(home, '.relayfile-mount-state', 'mount-1') + await mkdir(stateDir, { recursive: true }) + await writeFile(join(stateDir, 'state.json'), JSON.stringify({ + workspaceId: 'cloud-workspace-id', + localRoot: mirror, + })) + + expect(resolveRegisteredWorkspaceMirror(['rw_handle', 'cloud-workspace-id'], home)).toEqual({ + localDir: mirror, + source: 'mount-state', + }) + }) + }) + + it('anchors a legacy relative mount-state root to the Relayfile home', async () => { + await withTempHome(async (home) => { + const stateDir = join(home, '.relayfile-mount-state', 'mount-1') + await mkdir(stateDir, { recursive: true }) + await writeFile(join(stateDir, 'state.json'), JSON.stringify({ + workspaceId: 'rw_shared', + localRoot: 'registered/.integrations', + })) + + expect(resolveRegisteredWorkspaceMirror(['rw_shared'], home)).toEqual({ + localDir: join(home, 'registered', '.integrations'), + source: 'mount-state', + }) + }) + }) + + it('refuses ambiguous workspace-registry roots across accepted aliases', async () => { + await withTempHome(async (home) => { + await mkdir(join(home, '.relayfile'), { recursive: true }) + await writeFile(join(home, '.relayfile', 'workspaces.json'), JSON.stringify({ + workspaces: [ + { id: 'rw_handle', localDir: join(home, 'first', '.integrations') }, + { id: 'cloud-workspace-id', localDir: join(home, 'second', '.integrations') }, + ], + })) + + expect(resolveRegisteredWorkspaceMirror(['rw_handle', 'cloud-workspace-id'], home)).toBeUndefined() + }) + }) + + it('refuses to guess when stale local state names more than one mirror', async () => { + await withTempHome(async (home) => { + const stateRoot = join(home, '.relayfile-mount-state') + await Promise.all(['one', 'two'].map(async (name) => { + const stateDir = join(stateRoot, name) + await mkdir(stateDir, { recursive: true }) + await writeFile(join(stateDir, 'state.json'), JSON.stringify({ + workspaceId: 'rw_shared', + localRoot: join(home, name, '.integrations'), + })) + })) + + expect(resolveRegisteredWorkspaceMirror(['rw_shared'], home)).toBeUndefined() + }) + }) +}) diff --git a/src/mount/workspace-mirror.ts b/src/mount/workspace-mirror.ts new file mode 100644 index 0000000..5d7fa20 --- /dev/null +++ b/src/mount/workspace-mirror.ts @@ -0,0 +1,105 @@ +import { readdirSync, readFileSync } from 'node:fs' +import { homedir } from 'node:os' +import { isAbsolute, join, resolve } from 'node:path' + +type RecordValue = Record + +export interface RegisteredWorkspaceMirror { + localDir: string + source: 'workspace-registry' | 'mount-state' +} + +/** + * Resolve the one local mirror that Relayfile has already associated with a + * workspace. Relayfile versions have stored this association in both the + * workspace registry and the private mount-state directory, so accept either + * format. This is deliberately read-only: a missing registration must never + * cause Factory to re-home a user's mirror. + */ +export function resolveRegisteredWorkspaceMirror( + workspaceIds: readonly string[], + homeDir = homedir(), +): RegisteredWorkspaceMirror | undefined { + const accepted = new Set(workspaceIds.filter((id) => id.trim().length > 0)) + if (accepted.size === 0) return undefined + + const registryMirror = readWorkspaceRegistry(join(homeDir, '.relayfile', 'workspaces.json'), accepted, homeDir) + if (registryMirror) return { localDir: registryMirror, source: 'workspace-registry' } + + return readMountStateDirectory(join(homeDir, '.relayfile-mount-state'), accepted, homeDir) +} + +function readWorkspaceRegistry(path: string, accepted: ReadonlySet, homeDir: string): string | undefined { + let payload: unknown + try { + payload = JSON.parse(readFileSync(path, 'utf8')) as unknown + } catch { + return undefined + } + + const mirrors = new Set() + for (const record of workspaceRecords(payload)) { + const workspaceId = stringField(record, 'id') ?? stringField(record, 'workspaceId') ?? stringField(record, 'workspace') + if (!workspaceId || !accepted.has(workspaceId)) continue + const localDir = stringField(record, 'localDir') ?? stringField(record, 'localRoot') ?? stringField(record, 'mirrorDir') + if (localDir) mirrors.add(resolveRegisteredLocalDir(homeDir, localDir)) + } + // Workspace aliases can appear as separate records. Like mount state, do + // not choose one by JSON ordering if they disagree about the mirror root. + return mirrors.size === 1 ? [...mirrors][0] : undefined +} + +function workspaceRecords(payload: unknown): RecordValue[] { + if (Array.isArray(payload)) return payload.filter(isRecord) + if (!isRecord(payload)) return [] + const workspaces = payload.workspaces + if (Array.isArray(workspaces)) return workspaces.filter(isRecord) + if (!isRecord(workspaces)) return [] + return Object.entries(workspaces) + .filter((entry): entry is [string, RecordValue] => isRecord(entry[1])) + .map(([id, record]) => ({ id, ...record })) +} + +function readMountStateDirectory( + stateRoot: string, + accepted: ReadonlySet, + homeDir: string, +): RegisteredWorkspaceMirror | undefined { + let entries: string[] + try { + entries = readdirSync(stateRoot) + } catch { + return undefined + } + + const mirrors = new Set() + for (const entry of entries) { + try { + const state = JSON.parse(readFileSync(join(stateRoot, entry, 'state.json'), 'utf8')) as unknown + if (!isRecord(state) || !accepted.has(stringField(state, 'workspaceId') ?? '')) continue + const localDir = stringField(state, 'localDir') ?? stringField(state, 'localRoot') + if (localDir) mirrors.add(resolveRegisteredLocalDir(homeDir, localDir)) + } catch { + // A partially-written or retired mount state is not a registration. + } + } + + // Relayfile admits one mirror per workspace. Treat contradictory local state + // as unavailable instead of guessing which directory to refresh. + if (mirrors.size !== 1) return undefined + return { localDir: [...mirrors][0]!, source: 'mount-state' } +} + +/** Registry values are stored under this user's Relayfile home; preserve that base for relative legacy entries. */ +function resolveRegisteredLocalDir(homeDir: string, localDir: string): string { + return isAbsolute(localDir) ? resolve(localDir) : resolve(homeDir, localDir) +} + +function stringField(record: RecordValue, key: string): string | undefined { + const value = record[key] + return typeof value === 'string' && value.trim().length > 0 ? value : undefined +} + +function isRecord(value: unknown): value is RecordValue { + return typeof value === 'object' && value !== null && !Array.isArray(value) +} diff --git a/src/orchestrator/factory.test.ts b/src/orchestrator/factory.test.ts index 83ead19..6ff9061 100644 --- a/src/orchestrator/factory.test.ts +++ b/src/orchestrator/factory.test.ts @@ -8609,6 +8609,7 @@ describe('FactoryLoop', () => { const factory = createFactory(config(), { mount, fleet, triage: new StaticTriage() }) await factory.start({ mode: 'live', liveSubscription: { transport: 'subscribe' } }) + expect(factory.status().eventListener).toEqual({ state: 'subscribed' }) mount.files.set(path, { content: realIssueFile(25) }) mount.emit(changeEvent(path, 'event-live-25')) await vi.waitFor(() => { @@ -9059,6 +9060,8 @@ describe('FactoryLoop', () => { await factory.start({ mode: 'live', liveSubscription: { transport: 'poll', pollIntervalMs: 10 } }) await vi.advanceTimersByTimeAsync(0) + expect(factory.status().eventListener).toEqual({ state: 'polling' }) + expect(fleet.spawns.map((spawn) => spawn.name)).toEqual(['ar-30-impl-pear', 'ar-30-review']) mount.files.set(newPath, { content: realIssueFile(31) }) diff --git a/src/orchestrator/factory.ts b/src/orchestrator/factory.ts index e4b3f15..927ec21 100644 --- a/src/orchestrator/factory.ts +++ b/src/orchestrator/factory.ts @@ -67,6 +67,7 @@ import { isResourceSubscriptionsUnavailable, type ResourceSubscription } from '. import type { DispatchResult, Factory, + FactoryEventListenerStatus, FactoryEventPayload, FactoryPorts, FactoryStatus, @@ -463,6 +464,10 @@ export class FactoryLoop implements Factory { #subscription?: Subscription #livePollTimer?: ReturnType #livePollInFlight = false + // The effective transport includes per-start overrides, which can differ + // from config.liveSubscription. Persist it so status/heartbeat describes + // the listener that was actually registered. + #liveTransport?: FactoryLiveSubscriptionOptions['transport'] #liveEventCursor?: string #liveEventHighWatermark?: string #liveConnectStartedAtMs = 0 @@ -832,6 +837,11 @@ export class FactoryLoop implements Factory { this.#schedulePreviewSweep() try { await this.#startLiveSubscription(issueSource, opts.liveSubscription) + // The initial live heartbeat is intentionally written before startup + // so crash reapers can see a daemon while it bootstraps. Write again + // once the listener is actually registered so external `factory + // status` can distinguish a quiet feed from no listener. + await this.#writeLiveHeartbeat('running') await this.#rearmSlackReplyWatchers() await this.#drainReadyClarificationWake() await this.#rearmGithubIssueCommentWatchers() @@ -1045,6 +1055,7 @@ export class FactoryLoop implements Factory { overrides: Partial = {}, ): Promise { const options = this.#liveOptions(overrides) + this.#liveTransport = options.transport this.#liveConnectStartedAtMs = this.#clock.now() this.#liveReplaySkewMarginMs = options.replaySkewMarginMs const highWatermark = await this.#currentEventHighWatermark() @@ -2782,9 +2793,29 @@ export class FactoryLoop implements Factory { counters: { ...this.#counters }, slackDegraded: this.#slackDegraded, slackDegradedReason: this.#slackDegradedReason, + eventListener: this.#eventListenerStatus(), } } + #eventListenerStatus(): FactoryEventListenerStatus { + if (this.#startMode !== 'live') { + return { + state: 'not-listening', + reason: this.#startMode ? `factory mode is ${this.#startMode}` : 'factory has not started', + } + } + if (!this.#liveHeartbeatActive) { + return { state: 'not-listening', reason: 'live daemon heartbeat is inactive' } + } + if (this.#subscription) { + return { state: 'subscribed' } + } + if (this.#liveTransport === 'poll') { + return { state: 'polling' } + } + return { state: 'starting' } + } + on(event: FactoryEvent, listener: Listener): () => void { let listeners = this.#listeners.get(event) if (!listeners) { @@ -5072,6 +5103,7 @@ export class FactoryLoop implements Factory { updatedAt: new Date(updatedAtMs).toISOString(), updatedAtMs, registryPath, + eventListener: this.#eventListenerStatus(), } await mkdir(dirname(path), { recursive: true }) await writeFile(path, `${JSON.stringify(heartbeat, null, 2)}\n`, 'utf8') @@ -12825,12 +12857,11 @@ export class FactoryLoop implements Factory { } } - // Absolute path to the local .integrations mount the daemon manages. The mount - // is created at the daemon's cwd (see ensureLocalMount), and spawned agents run - // in their repo clonePath, so writeback paths handed to agents must be absolute - // against this root rather than a bare relative `.integrations/...`. + // Absolute path to the one registered workspace mirror. Spawned agents run in + // repo clone paths, so writeback instructions must name the shared mirror, + // not a relative `.integrations` path or a per-repository re-home attempt. #integrationsMountRoot(): string { - return resolve(process.cwd(), '.integrations') + return this.#mount.getLocalMountRoot?.() ?? resolve(process.cwd(), '.integrations') } async #slackChannelDir(): Promise { diff --git a/src/ports/index.ts b/src/ports/index.ts index 971bbd8..90fb19c 100644 --- a/src/ports/index.ts +++ b/src/ports/index.ts @@ -8,6 +8,7 @@ export type { GithubConnectionWrite, GithubPublishPullRequestInput, GithubPublishPullRequestResult, + LocalMountHealth, LocalMountOptions, MountClient, ProviderSyncStatus, diff --git a/src/ports/mount.ts b/src/ports/mount.ts index f253201..ab142a7 100644 --- a/src/ports/mount.ts +++ b/src/ports/mount.ts @@ -43,6 +43,12 @@ export interface LocalMountOptions { stateWaitPollMs?: number } +export interface LocalMountHealth { + degraded: boolean + reason?: string + localDir?: string +} + export interface GithubPublishPullRequestInput { repo: string /** Local checkout fallback for internal/local dispatches. */ @@ -106,8 +112,12 @@ export interface MountClient { */ readonly resourceSubscriptions?: ResourceSubscriptionsClient readonly integrationConnections?: FactoryIntegrationConnections - /** Ensure the SDK-authenticated Relayfile mirror exists below a checkout. */ + /** Ensure the SDK-authenticated Relayfile workspace mirror is available. */ ensureLocalMount?(startDir: string, options?: LocalMountOptions): Promise + /** The registered workspace mirror root, when the mount can resolve one. */ + getLocalMountRoot?(): string | undefined + /** Read-only local mirror freshness for `factory status`. */ + getLocalMountHealth?(): LocalMountHealth /** * Whether a local mount is terminally degraded because the cloud session * lacks the filesystem scope the mount needs. When true, the mirror is diff --git a/src/types.ts b/src/types.ts index 1c3c05e..1821891 100644 --- a/src/types.ts +++ b/src/types.ts @@ -93,6 +93,16 @@ export interface FactoryLiveSubscriptionOptions { replaySkewMarginMs: number } +/** + * The daemon's own view of its primary Relayfile event listener. This is + * intentionally separate from the local mirror's reconcile health: a quiet + * event stream can be healthy, while a stopped daemon is not listening at all. + */ +export interface FactoryEventListenerStatus { + state: 'starting' | 'subscribed' | 'polling' | 'not-listening' | 'unknown' + reason?: string +} + export interface FactoryLoopRunOptions { dryRun?: boolean maxIterations?: number @@ -111,6 +121,7 @@ export interface FactoryLoopHeartbeat { updatedAt: string updatedAtMs: number registryPath?: string + eventListener?: FactoryEventListenerStatus } export interface FactoryInFlightRegistryAgent { @@ -205,6 +216,8 @@ export interface FactoryStatus { counters: Record slackDegraded?: boolean slackDegradedReason?: string + /** Primary Relayfile subscription/poll registration, not event activity. */ + eventListener?: FactoryEventListenerStatus } export type FactoryEventPayload =