From 1ce4d21bd193247db4f91906397ef34c09c0c537 Mon Sep 17 00:00:00 2001 From: Miya Date: Fri, 7 Aug 2026 18:59:19 +0200 Subject: [PATCH 1/8] fix: use one registered relayfile workspace mirror --- src/cli/fleet.test.ts | 112 ++++++++----- src/cli/fleet.ts | 152 ++++++++---------- src/config/schema.ts | 4 + src/mount/local-mount-preflight.test.ts | 4 + src/mount/local-mount-preflight.ts | 12 +- src/mount/relayfile-binary.test.ts | 23 ++- src/mount/relayfile-binary.ts | 7 +- .../relayfile-cloud-mount-client.test.ts | 151 ++++++++++++++++- src/mount/relayfile-cloud-mount-client.ts | 97 +++++++++-- src/mount/workspace-mirror.test.ts | 65 ++++++++ src/mount/workspace-mirror.ts | 96 +++++++++++ src/orchestrator/factory.ts | 9 +- src/ports/index.ts | 1 + src/ports/mount.ts | 12 +- 14 files changed, 594 insertions(+), 151 deletions(-) create mode 100644 src/mount/workspace-mirror.test.ts create mode 100644 src/mount/workspace-mirror.ts diff --git a/src/cli/fleet.test.ts b/src/cli/fleet.test.ts index 17558e12..760d9126 100644 --- a/src/cli/fleet.test.ts +++ b/src/cli/fleet.test.ts @@ -1862,6 +1862,50 @@ describe('fleet CLI runtime', () => { } }) + it('surfaces a stale registered workspace mirror in factory status', async () => { + const root = await mkdtemp(join(tmpdir(), 'fleet-cli-stale-status-')) + try { + const configPath = await writeConfig(root) + 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 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({ + localMountDegraded: true, + localMountDegradedReason: 'last reconcile 5m ago', + localMountRoot: mirror, + }) + } finally { + await rm(root, { recursive: true, force: true }) + } + }) + it('drives the real RelayFleetClient when --backend relay is requested', async () => { const output = buffer() const errors = buffer() @@ -2008,8 +2052,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 +2353,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 +2403,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 +2425,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 +2488,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 +2531,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 +2618,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 +2630,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 +2644,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,32 +2659,30 @@ 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 }) diff --git a/src/cli/fleet.ts b/src/cli/fleet.ts index fd6fc128..04f2da20 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' } @@ -639,6 +637,7 @@ async function runFactoryCommand( // 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( + mount, mountFn, workspaceId, config, @@ -678,19 +677,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, factoryStatusWithMountHealth(factory, mount)) return 0 } if (command.action === 'loop-status') { @@ -707,20 +705,19 @@ 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: factoryStatusWithMountHealth(factory, mount) }) } finally { removeSignalHandlers() await factory.stop() @@ -762,6 +759,7 @@ async function runFactoryCommand( } async function warmStartPathMounts( + mount: MountClient, mountFn: NonNullable, workspaceId: string, config: FactoryConfig, @@ -769,23 +767,17 @@ async function warmStartPathMounts( 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], + 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`, ) } @@ -803,11 +795,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 +863,7 @@ async function runStandaloneBabysitCommand( maintainerCanModify: pr.maintainerCanModify, }, standaloneBabysitter: { specSource }, - integrationsMountRoot: resolve(process.cwd(), '.integrations'), + integrationsMountRoot: resolveIntegrationsMountRoot(mount), testGuidance, }) const receiptBase = { @@ -915,27 +903,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 +933,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 +965,47 @@ 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') +} + +function factoryStatusWithMountHealth(factory: Factory, mount: MountClient): ReturnType & { + localMountDegraded?: boolean + localMountDegradedReason?: string + localMountRoot?: string +} { + const health = mount.getLocalMountHealth?.() + if (!health) return factory.status() + return { + ...factory.status(), + localMountDegraded: health.degraded, + ...(health.reason ? { localMountDegradedReason: health.reason } : {}), + ...(health.localDir ? { localMountRoot: health.localDir } : {}), } - return undefined } function writeMountRefreshSummary( @@ -1539,6 +1514,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.ts b/src/config/schema.ts index 81c2dbf0..754253cd 100644 --- a/src/config/schema.ts +++ b/src/config/schema.ts @@ -234,6 +234,10 @@ 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().min(1).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 23e0b0f3..39bba53e 100644 --- a/src/mount/local-mount-preflight.test.ts +++ b/src/mount/local-mount-preflight.test.ts @@ -99,6 +99,7 @@ describe('ensureLocalMount', () => { pid: process.pid, }) }) + const stderr = vi.spyOn(process.stderr, 'write').mockImplementation(() => true) await expect(ensureLocalMount('rw_test', dir, { startMount, @@ -106,6 +107,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 3235eceb..d0b38ab8 100644 --- a/src/mount/local-mount-preflight.ts +++ b/src/mount/local-mount-preflight.ts @@ -82,6 +82,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 = join(startDir, '.integrations') // Stale AND under-scoped: refreshing cannot help. Fail fast and terminal so // the supervisor stops retrying and startup surfaces one actionable error. @@ -90,14 +94,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 +113,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 +128,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 fe729665..cd9a21d6 100644 --- a/src/mount/relayfile-binary.test.ts +++ b/src/mount/relayfile-binary.test.ts @@ -3,7 +3,11 @@ 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_INTERVALS, +} from './relayfile-binary' afterEach(() => { vi.restoreAllMocks() @@ -93,6 +97,23 @@ 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('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 29333e6a..8f6bb871 100644 --- a/src/mount/relayfile-binary.ts +++ b/src/mount/relayfile-binary.ts @@ -1,6 +1,11 @@ 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 type MountState = { workspaceId?: unknown diff --git a/src/mount/relayfile-cloud-mount-client.test.ts b/src/mount/relayfile-cloud-mount-client.test.ts index f8050f55..edb8a015 100644 --- a/src/mount/relayfile-cloud-mount-client.test.ts +++ b/src/mount/relayfile-cloud-mount-client.test.ts @@ -410,6 +410,146 @@ describe('RelayfileCloudMountClient', () => { expect(stop).toHaveBeenCalledTimes(1) }) + 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('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 mount.ensureLocalMount(join(root, 'first-checkout')) + + expect(ensureMountedWorkspace).toHaveBeenNthCalledWith(1, expect.objectContaining({ localDir: fallbackDir })) + expect(ensureMountedWorkspace).toHaveBeenNthCalledWith(2, expect.objectContaining({ localDir: registeredDir })) + 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 +677,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 +703,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 fb450021..508c69dd 100644 --- a/src/mount/relayfile-cloud-mount-client.ts +++ b/src/mount/relayfile-cloud-mount-client.ts @@ -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,10 @@ 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 isAllowedDraft?: (path: string, content: unknown, opts?: { guarded?: boolean }) => boolean | Promise isAllowedDelete?: (path: string, currentContent: unknown) => boolean | Promise } @@ -266,6 +272,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 hasResolvedMirror = this.#localMountRoot !== undefined + const localDir = this.#localMountRoot ?? join(resolve(startDir), '.integrations') + this.#localMountRoot = localDir + try { + await this.#runLocalMountOperation(localDir, () => 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. Retry exactly that + // root once; never ask Relayfile to re-home it and never override an + // explicit/configured mirror root. + const registeredRoot = hasResolvedMirror ? undefined : registeredMirrorFromMountError(error) + if (!registeredRoot || registeredRoot === localDir) throw error + this.#clearLocalMountHealthCheck(localDir) + this.#localMountSupervisions.delete(localDir) + this.#degradedLocalMounts.delete(localDir) + this.#authDegradedLocalMounts.delete(localDir) + this.#localMountRoot = registeredRoot + await this.#runLocalMountOperation(registeredRoot, () => this.#ensureLocalMount(registeredRoot, options)) + } + } + + getLocalMountRoot(): string | undefined { + return this.#localMountRoot + } + + 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(startDir: string, localDir: string, options: LocalMountOptions): Promise { + async #ensureLocalMount(localDir: string, options: LocalMountOptions): Promise { const setup = this.#relayfileSetup const workspace = this.#relayfileWorkspace if (!setup?.ensureMountedWorkspace || !workspace) { @@ -382,10 +440,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 +459,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,7 +472,7 @@ 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, acceptableWorkspaceIds: [...acceptableWorkspaceIds], startMount: async () => { @@ -445,6 +500,17 @@ export class RelayfileCloudMountClient implements MountClient { this.#scheduleLocalMountHealthCheck(localDir) } + #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 +987,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 localDir.startsWith('/') ? 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 00000000..9d42ac66 --- /dev/null +++ b/src/mount/workspace-mirror.test.ts @@ -0,0 +1,65 @@ +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('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('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 00000000..93cf0b15 --- /dev/null +++ b/src/mount/workspace-mirror.ts @@ -0,0 +1,96 @@ +import { readdirSync, readFileSync } from 'node:fs' +import { homedir } from 'node:os' +import { 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) + if (registryMirror) return { localDir: registryMirror, source: 'workspace-registry' } + + return readMountStateDirectory(join(homeDir, '.relayfile-mount-state'), accepted) +} + +function readWorkspaceRegistry(path: string, accepted: ReadonlySet): string | undefined { + let payload: unknown + try { + payload = JSON.parse(readFileSync(path, 'utf8')) as unknown + } catch { + return undefined + } + + 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) return resolve(localDir) + } + return 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, +): 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(resolve(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' } +} + +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.ts b/src/orchestrator/factory.ts index e4b3f151..758400a8 100644 --- a/src/orchestrator/factory.ts +++ b/src/orchestrator/factory.ts @@ -12825,12 +12825,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 971bbd87..90fb19c6 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 f2532010..ab142a79 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 From 5f5cb2b3a0bd111c668f49b1092dbe08ba27c0bf Mon Sep 17 00:00:00 2001 From: Miya Date: Fri, 7 Aug 2026 19:09:36 +0200 Subject: [PATCH 2/8] fix: harden registered mirror resolution --- src/config/schema.test.ts | 15 ++++++ src/config/schema.ts | 8 ++- .../relayfile-cloud-mount-client.test.ts | 28 +++++++++- src/mount/relayfile-cloud-mount-client.ts | 54 +++++++++++-------- src/mount/workspace-mirror.test.ts | 14 +++++ src/mount/workspace-mirror.ts | 7 ++- 6 files changed, 100 insertions(+), 26 deletions(-) diff --git a/src/config/schema.test.ts b/src/config/schema.test.ts index d99450ab..ad990648 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 754253cd..aed87a56 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' @@ -237,7 +237,11 @@ const WorkspaceConfigObjectSchema = z.object({ // 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().min(1).optional(), + 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/relayfile-cloud-mount-client.test.ts b/src/mount/relayfile-cloud-mount-client.test.ts index edb8a015..7881e518 100644 --- a/src/mount/relayfile-cloud-mount-client.test.ts +++ b/src/mount/relayfile-cloud-mount-client.test.ts @@ -452,6 +452,28 @@ describe('RelayfileCloudMountClient', () => { } }) + 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') @@ -485,10 +507,14 @@ describe('RelayfileCloudMountClient', () => { }) try { - await mount.ensureLocalMount(join(root, 'first-checkout')) + 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() diff --git a/src/mount/relayfile-cloud-mount-client.ts b/src/mount/relayfile-cloud-mount-client.ts index 508c69dd..60722144 100644 --- a/src/mount/relayfile-cloud-mount-client.ts +++ b/src/mount/relayfile-cloud-mount-client.ts @@ -328,9 +328,10 @@ export class RelayfileCloudMountClient implements MountClient { this.#localMountPreflight = config.localMountPreflight ?? runLocalMountPreflight this.#localMountAgentName = config.agentName ?? DEFAULT_AGENT_NAME this.#localMountScopes = config.scopes ?? [...FACTORY_RELAYFILE_SCOPES] + const workspaceIds = this.#acceptableWorkspaceIds([this.workspaceId]) this.#localMountRoot = config.localMountRoot ?? - config.workspaceMirrorResolver?.([this.workspaceId]) ?? - resolveRegisteredWorkspaceMirror([this.workspaceId])?.localDir + config.workspaceMirrorResolver?.(workspaceIds) ?? + resolveRegisteredWorkspaceMirror(workspaceIds)?.localDir this.#isAllowedDraft = config.isAllowedDraft this.#isAllowedDelete = config.isAllowedDelete this.githubWrite = new RelayfileGithubConnectionWrite({ mount: this }) @@ -388,27 +389,13 @@ export class RelayfileCloudMountClient implements MountClient { // 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 hasResolvedMirror = this.#localMountRoot !== undefined + const canDiscoverRegisteredMirror = this.#localMountRoot === undefined const localDir = this.#localMountRoot ?? join(resolve(startDir), '.integrations') this.#localMountRoot = localDir - try { - await this.#runLocalMountOperation(localDir, () => 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. Retry exactly that - // root once; never ask Relayfile to re-home it and never override an - // explicit/configured mirror root. - const registeredRoot = hasResolvedMirror ? undefined : registeredMirrorFromMountError(error) - if (!registeredRoot || registeredRoot === localDir) throw error - this.#clearLocalMountHealthCheck(localDir) - this.#localMountSupervisions.delete(localDir) - this.#degradedLocalMounts.delete(localDir) - this.#authDegradedLocalMounts.delete(localDir) - this.#localMountRoot = registeredRoot - await this.#runLocalMountOperation(registeredRoot, () => this.#ensureLocalMount(registeredRoot, options)) - } + return this.#runLocalMountOperation( + localDir, + () => this.#ensureLocalMountWithRegisteredFallback(localDir, options, canDiscoverRegisteredMirror), + ) } getLocalMountRoot(): string | undefined { @@ -500,6 +487,31 @@ 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] : [])])] diff --git a/src/mount/workspace-mirror.test.ts b/src/mount/workspace-mirror.test.ts index 9d42ac66..e5956da1 100644 --- a/src/mount/workspace-mirror.test.ts +++ b/src/mount/workspace-mirror.test.ts @@ -47,6 +47,20 @@ describe('resolveRegisteredWorkspaceMirror', () => { }) }) + 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') diff --git a/src/mount/workspace-mirror.ts b/src/mount/workspace-mirror.ts index 93cf0b15..e1ba29f8 100644 --- a/src/mount/workspace-mirror.ts +++ b/src/mount/workspace-mirror.ts @@ -37,13 +37,16 @@ function readWorkspaceRegistry(path: string, accepted: ReadonlySet): str 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) return resolve(localDir) + if (localDir) mirrors.add(resolve(localDir)) } - return undefined + // 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[] { From 718ef84b976aa80590831862f406791161549312 Mon Sep 17 00:00:00 2001 From: Miya Date: Fri, 7 Aug 2026 19:23:09 +0200 Subject: [PATCH 3/8] fix: surface relayfile event listener and feed health --- src/cli/fleet.test.ts | 84 +++++++++++++++++++++++++++++++- src/cli/fleet.ts | 46 ++++++++++++++--- src/orchestrator/factory.test.ts | 3 ++ src/orchestrator/factory.ts | 27 ++++++++++ src/types.ts | 13 +++++ 5 files changed, 165 insertions(+), 8 deletions(-) diff --git a/src/cli/fleet.test.ts b/src/cli/fleet.test.ts index 760d9126..f40c7a3c 100644 --- a/src/cli/fleet.test.ts +++ b/src/cli/fleet.test.ts @@ -1354,7 +1354,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 { @@ -1865,7 +1871,8 @@ describe('fleet CLI runtime', () => { it('surfaces a stale registered workspace mirror in factory status', async () => { const root = await mkdtemp(join(tmpdir(), 'fleet-cli-stale-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 mirror = join(root, 'chief', '.integrations') const factory = { @@ -1886,6 +1893,16 @@ describe('fleet CLI runtime', () => { 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(), @@ -1897,9 +1914,72 @@ describe('fleet CLI runtime', () => { 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 }) diff --git a/src/cli/fleet.ts b/src/cli/fleet.ts index 04f2da20..2bbbdc73 100644 --- a/src/cli/fleet.ts +++ b/src/cli/fleet.ts @@ -688,7 +688,7 @@ async function runFactoryCommand( return 0 } if (command.action === 'status') { - writeJson(out, factoryStatusWithMountHealth(factory, mount)) + writeJson(out, await factoryStatusWithMountHealth(factory, mount, config.loop.heartbeatPath, config.loop.heartbeatStaleMs)) return 0 } if (command.action === 'loop-status') { @@ -717,7 +717,10 @@ async function runFactoryCommand( }) try { const reports = await factory.runLoop({ dryRun: globals.dryRun }) - writeJson(out, { reports, status: factoryStatusWithMountHealth(factory, mount) }) + writeJson(out, { + reports, + status: await factoryStatusWithMountHealth(factory, mount, config.loop.heartbeatPath, config.loop.heartbeatStaleMs), + }) } finally { removeSignalHandlers() await factory.stop() @@ -993,18 +996,49 @@ function resolveIntegrationsMountRoot(mount: MountClient): string { return mount.getLocalMountRoot?.() ?? resolve(process.cwd(), '.integrations') } -function factoryStatusWithMountHealth(factory: Factory, mount: MountClient): ReturnType & { +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 factory.status() + if (!health) return { ...status, eventListener } return { - ...factory.status(), + ...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 } : {}), + }, } } diff --git a/src/orchestrator/factory.test.ts b/src/orchestrator/factory.test.ts index 83ead190..6ff90614 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 758400a8..1a93ce5f 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, @@ -832,6 +833,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() @@ -2782,9 +2788,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.#liveOptions({}).transport === 'poll' && this.#liveEventCursor !== undefined) { + return { state: 'polling' } + } + return { state: 'starting' } + } + on(event: FactoryEvent, listener: Listener): () => void { let listeners = this.#listeners.get(event) if (!listeners) { @@ -5072,6 +5098,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') diff --git a/src/types.ts b/src/types.ts index 1c3c05e3..18218915 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 = From f50dbec3870d362a7813eaa7eab3a540370518fd Mon Sep 17 00:00:00 2001 From: Miya Date: Fri, 7 Aug 2026 19:26:17 +0200 Subject: [PATCH 4/8] fix: derive mount staleness from registered cadence --- src/mount/relayfile-binary.test.ts | 40 +++++++++++++++++++++++++++++- src/mount/relayfile-binary.ts | 16 +++++++++++- 2 files changed, 54 insertions(+), 2 deletions(-) diff --git a/src/mount/relayfile-binary.test.ts b/src/mount/relayfile-binary.test.ts index cd9a21d6..9e392d3f 100644 --- a/src/mount/relayfile-binary.test.ts +++ b/src/mount/relayfile-binary.test.ts @@ -6,7 +6,9 @@ import { afterEach, describe, expect, it, vi } from 'vitest' import { checkMountStaleness, RELAYFILE_SYNC_INTERVAL_MS, + STALE_RECONCILE_MS, STALE_RECONCILE_INTERVALS, + staleReconcileMs, } from './relayfile-binary' afterEach(() => { @@ -24,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') @@ -114,6 +122,36 @@ describe('checkMountStaleness', () => { }) }) + 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 8f6bb871..16241ebd 100644 --- a/src/mount/relayfile-binary.ts +++ b/src/mount/relayfile-binary.ts @@ -7,9 +7,23 @@ 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 @@ -63,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`, From b6f054fa39d7052e0d0f4906139a2ea6b03ff737 Mon Sep 17 00:00:00 2001 From: Miya Date: Fri, 7 Aug 2026 19:28:19 +0200 Subject: [PATCH 5/8] test: isolate factory status heartbeat fixtures --- src/cli/fleet.test.ts | 13 +++++++++++-- 1 file changed, 11 insertions(+), 2 deletions(-) diff --git a/src/cli/fleet.test.ts b/src/cli/fleet.test.ts index f40c7a3c..6a768f43 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') @@ -1834,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 = { @@ -1862,7 +1865,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', + }, + }) } finally { await rm(root, { recursive: true, force: true }) } From f1790d4f9204f3957bc17973a74b97dacf32a028 Mon Sep 17 00:00:00 2001 From: Miya Date: Fri, 7 Aug 2026 19:35:03 +0200 Subject: [PATCH 6/8] fix: report effective live event transport --- src/orchestrator/factory.ts | 7 ++++++- 1 file changed, 6 insertions(+), 1 deletion(-) diff --git a/src/orchestrator/factory.ts b/src/orchestrator/factory.ts index 1a93ce5f..927ec21a 100644 --- a/src/orchestrator/factory.ts +++ b/src/orchestrator/factory.ts @@ -464,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 @@ -1051,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() @@ -2805,7 +2810,7 @@ export class FactoryLoop implements Factory { if (this.#subscription) { return { state: 'subscribed' } } - if (this.#liveOptions({}).transport === 'poll' && this.#liveEventCursor !== undefined) { + if (this.#liveTransport === 'poll') { return { state: 'polling' } } return { state: 'starting' } From 0162008be162e2a2153d72c6622fed265c6d280b Mon Sep 17 00:00:00 2001 From: Miya Date: Fri, 7 Aug 2026 19:44:42 +0200 Subject: [PATCH 7/8] fix: resolve workspace mirrors before live dispatch --- src/cli/fleet.test.ts | 78 +++++++++++++++++++ src/cli/fleet.ts | 46 +++++++---- src/mount/local-mount-preflight.test.ts | 23 ++++++ src/mount/local-mount-preflight.ts | 9 ++- .../relayfile-cloud-mount-client.test.ts | 54 +++++++++++++ src/mount/relayfile-cloud-mount-client.ts | 15 ++-- src/mount/workspace-mirror.test.ts | 30 +++++++ src/mount/workspace-mirror.ts | 18 +++-- 8 files changed, 243 insertions(+), 30 deletions(-) diff --git a/src/cli/fleet.test.ts b/src/cli/fleet.test.ts index 6a768f43..cc47853d 100644 --- a/src/cli/fleet.test.ts +++ b/src/cli/fleet.test.ts @@ -2778,6 +2778,84 @@ describe('fleet CLI runtime', () => { } }) + 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('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 2bbbdc73..ce1965f0 100644 --- a/src/cli/fleet.ts +++ b/src/cli/fleet.ts @@ -627,16 +627,17 @@ 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, @@ -645,16 +646,30 @@ 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 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`) + } + 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') + return 1 } - const message = error instanceof Error ? error.message : String(error) - mountStderr.write(`[factory] warning: background relayfile mount warmup failed: ${message}\n`) - }) + } catch (error) { + handleWarmMountError(error) + return 1 + } + } else { + void warmMount().catch(handleWarmMountError) + } const removeSignalHandlers = installFactoryStopSignalHandlers(factory, { exit: (code) => { stoppedBySignal = true @@ -769,7 +784,7 @@ async function warmStartPathMounts( acceptableMountIds?: readonly string[], stderr: Pick = process.stderr, debug = process.env.FACTORY_LOG_LEVEL?.toLowerCase() === 'debug', -): Promise { +): Promise { const result = await ensureWorkspaceMount( mount, mountFn, @@ -782,6 +797,7 @@ async function warmStartPathMounts( `[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( diff --git a/src/mount/local-mount-preflight.test.ts b/src/mount/local-mount-preflight.test.ts index 39bba53e..54eed0c1 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 () => { diff --git a/src/mount/local-mount-preflight.ts b/src/mount/local-mount-preflight.ts index d0b38ab8..7744425c 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 { @@ -85,7 +86,7 @@ export async function ensureLocalMount( // 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 = join(startDir, '.integrations') + 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. diff --git a/src/mount/relayfile-cloud-mount-client.test.ts b/src/mount/relayfile-cloud-mount-client.test.ts index 7881e518..4dcfda87 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,59 @@ 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') diff --git a/src/mount/relayfile-cloud-mount-client.ts b/src/mount/relayfile-cloud-mount-client.ts index 60722144..0d61125f 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, @@ -216,6 +216,8 @@ export interface RelayfileCloudMountClientConfig { 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 } @@ -329,9 +331,10 @@ export class RelayfileCloudMountClient implements MountClient { this.#localMountAgentName = config.agentName ?? DEFAULT_AGENT_NAME this.#localMountScopes = config.scopes ?? [...FACTORY_RELAYFILE_SCOPES] const workspaceIds = this.#acceptableWorkspaceIds([this.workspaceId]) - this.#localMountRoot = config.localMountRoot ?? - config.workspaceMirrorResolver?.(workspaceIds) ?? - resolveRegisteredWorkspaceMirror(workspaceIds)?.localDir + this.#localMountRoot = config.localMountRoot ?? (config.skipRegisteredMirrorLookup + ? undefined + : config.workspaceMirrorResolver?.(workspaceIds) ?? + resolveRegisteredWorkspaceMirror(workspaceIds)?.localDir) this.#isAllowedDraft = config.isAllowedDraft this.#isAllowedDelete = config.isAllowedDelete this.githubWrite = new RelayfileGithubConnectionWrite({ mount: this }) @@ -373,6 +376,7 @@ export class RelayfileCloudMountClient implements MountClient { ...config, workspaceId, ...(registeredMountRoot ? { localMountRoot: registeredMountRoot } : {}), + skipRegisteredMirrorLookup: true, client, // WorkspaceHandle.getToken() returns the token originally minted by // joinWorkspace. RelayFileClient.getToken() resolves the SDK's rotating @@ -461,6 +465,7 @@ export class RelayfileCloudMountClient implements MountClient { try { await this.#localMountPreflight(this.workspaceId, join(localDir, '..'), { ...options, + localDir, acceptableWorkspaceIds: [...acceptableWorkspaceIds], startMount: async () => { await this.#replaceLocalMount(localDir, launch) @@ -1009,7 +1014,7 @@ function registeredMirrorFromMountError(error: unknown): string | undefined { const match = /\balready mirrored at\s+(.+?)(?:;|\n|$)/iu.exec(message) if (!match?.[1]) return undefined const localDir = match[1].trim().replace(/^["']|["']$/gu, '') - return localDir.startsWith('/') ? resolve(localDir) : undefined + return isAbsolute(localDir) ? resolve(localDir) : undefined } const isHttpStatus = (error: unknown, status: number): boolean => { diff --git a/src/mount/workspace-mirror.test.ts b/src/mount/workspace-mirror.test.ts index e5956da1..671270b8 100644 --- a/src/mount/workspace-mirror.test.ts +++ b/src/mount/workspace-mirror.test.ts @@ -30,6 +30,20 @@ describe('resolveRegisteredWorkspaceMirror', () => { }) }) + 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') @@ -47,6 +61,22 @@ describe('resolveRegisteredWorkspaceMirror', () => { }) }) + 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 }) diff --git a/src/mount/workspace-mirror.ts b/src/mount/workspace-mirror.ts index e1ba29f8..5d7fa20f 100644 --- a/src/mount/workspace-mirror.ts +++ b/src/mount/workspace-mirror.ts @@ -1,6 +1,6 @@ import { readdirSync, readFileSync } from 'node:fs' import { homedir } from 'node:os' -import { join, resolve } from 'node:path' +import { isAbsolute, join, resolve } from 'node:path' type RecordValue = Record @@ -23,13 +23,13 @@ export function resolveRegisteredWorkspaceMirror( 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) + 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) + return readMountStateDirectory(join(homeDir, '.relayfile-mount-state'), accepted, homeDir) } -function readWorkspaceRegistry(path: string, accepted: ReadonlySet): string | undefined { +function readWorkspaceRegistry(path: string, accepted: ReadonlySet, homeDir: string): string | undefined { let payload: unknown try { payload = JSON.parse(readFileSync(path, 'utf8')) as unknown @@ -42,7 +42,7 @@ function readWorkspaceRegistry(path: string, accepted: ReadonlySet): str 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(resolve(localDir)) + 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. @@ -63,6 +63,7 @@ function workspaceRecords(payload: unknown): RecordValue[] { function readMountStateDirectory( stateRoot: string, accepted: ReadonlySet, + homeDir: string, ): RegisteredWorkspaceMirror | undefined { let entries: string[] try { @@ -77,7 +78,7 @@ function readMountStateDirectory( 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(resolve(localDir)) + if (localDir) mirrors.add(resolveRegisteredLocalDir(homeDir, localDir)) } catch { // A partially-written or retired mount state is not a registration. } @@ -89,6 +90,11 @@ function readMountStateDirectory( 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 From 9f39c85fbe8c98d0d0e7fe8f5e02301845b0b94f Mon Sep 17 00:00:00 2001 From: kjgbot Date: Fri, 7 Aug 2026 23:46:01 +0200 Subject: [PATCH 8/8] fix(cli): handle stop signals during mount preflight --- src/cli/fleet.test.ts | 67 +++++++++++++++++++++++++++++++++++++++++++ src/cli/fleet.ts | 31 +++++++++++--------- 2 files changed, 84 insertions(+), 14 deletions(-) diff --git a/src/cli/fleet.test.ts b/src/cli/fleet.test.ts index cc47853d..0525462e 100644 --- a/src/cli/fleet.test.ts +++ b/src/cli/fleet.test.ts @@ -2821,6 +2821,73 @@ describe('fleet CLI runtime', () => { } }) + 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 { diff --git a/src/cli/fleet.ts b/src/cli/fleet.ts index ce1965f0..63fd4809 100644 --- a/src/cli/fleet.ts +++ b/src/cli/fleet.ts @@ -656,20 +656,6 @@ async function runFactoryCommand( const message = error instanceof Error ? error.message : String(error) mountStderr.write(`[factory] warning: background relayfile mount warmup failed: ${message}\n`) } - 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') - return 1 - } - } catch (error) { - handleWarmMountError(error) - return 1 - } - } else { - void warmMount().catch(handleWarmMountError) - } const removeSignalHandlers = installFactoryStopSignalHandlers(factory, { exit: (code) => { stoppedBySignal = true @@ -681,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