Fix durable relay dispatch lifecycle

README.md

@@ -212,6 +212,23 @@ required capability (a named `node` target passes through), the node runs the
 agent in its mapped checkout, and the orchestrator detects exits by reconciling
 its tracked agents against the engine roster.
 
+Relay dispatch is lifecycle-owned, not fire-and-forget. A one-shot `factory
+dispatch --backend relay` keeps a small publisher runtime alive until the
+remote branch has produced a PR, terminal issue writeback is acknowledged, and
+remote agents are released. The lifecycle (including a per-run branch,
+placement results, PR receipt, and a fenced owner lease) is persisted beside
+the configured loop registry so `factory start` or a replacement dispatch
+process on the same control-plane host can take over after a crash. Execution
+nodes never need access to that state file, and remote PIDs are never signalled
+as local processes.
+
+The supported topology is one Factory control-plane host per workspace, with
+any number of relay execution nodes. Multiple Factory processes on that host
+are fenced through the shared `FileStateStore` lock/lease. Active/active Factory
+control planes on different hosts are intentionally unsupported: the current
+Relayfile and Relay messaging APIs do not expose a shared compare-and-set lease,
+so separate local state files cannot provide a truthful cross-host fence.
+
 Tokens involved — set only the first one on the orchestrator host:
 
 | Token | Prefix | Who holds it |

src/cli/fleet.test.ts

@@ -7,6 +7,7 @@ import type { CloseProbePrInput, Factory, FactoryPorts, createFactory } from '..
 import { stateResolutionFromIds } from '../index'
 import { FileStateStore } from '../state/file-state-store'
 import { FakeFleetClient, FakeMountClient } from '../testing'
+import type { GithubConnectionWrite, SpawnInput, SpawnResult } from '../ports'
 import { formatLogArgs, installFactoryStopSignalHandlers, parseFleetCommand, parseGlobalOptions, resolveBrokerConnectionPath, runFleetCli } from './fleet'
 
 const issuePath = '/linear/issues/AR-77__uuid-77.json'
@@ -48,6 +49,29 @@ const issueFile = {
   },
 }
 
+class CompletingRemoteFleetClient extends FakeFleetClient {
+  override readonly placementLocality = 'remote' as const
+  readonly lifecycleOrder: string[] = []
+
+  override async spawn(input: SpawnInput): Promise<SpawnResult> {
+    const result = await super.spawn(input)
+    if (input.name.includes('-impl-')) {
+      setTimeout(() => this.emitAgentExit(input.name, 'exited'), 0)
+    }
+    return { ...result, node: 'sf-mini', locality: 'remote' }
+  }
+
+  override async release(name: string, reason?: string): Promise<void> {
+    this.lifecycleOrder.push(`release:${name}`)
+    await super.release(name, reason)
+  }
+
+  override async dispose(): Promise<void> {
+    this.lifecycleOrder.push('dispose')
+    await super.dispose()
+  }
+}
+
 const githubIssueFile = (repo: string, number = 48) => ({
   provider: 'github',
   objectType: 'issue',
@@ -547,6 +571,69 @@ describe('fleet CLI runtime', () => {
     }
   })
 
+  it('keeps relay dispatch ownership until the remote PR is published and the issue is parked', async () => {
+    const root = await mkdtemp(join(tmpdir(), 'fleet-cli-relay-owner-'))
+    try {
+      const configPath = await writeConfig(root, {
+        loop: {
+          heartbeatPath: join(root, 'heartbeat.json'),
+          registryPath: join(root, 'registry.json'),
+          heartbeatStaleMs: 10_000,
+        },
+      })
+      const publishes: Parameters<GithubConnectionWrite['publishPullRequest']>[0][] = []
+      const githubWrite: GithubConnectionWrite = {
+        publishPullRequest: async (input) => {
+          publishes.push(input)
+          return {
+            repo: input.repo,
+            number: 77,
+            url: 'https://github.com/AgentWorkforce/pear/pull/77',
+            headRef: input.headRef ?? 'unexpected-local-head',
+          }
+        },
+        closePullRequest: async () => undefined,
+      }
+      const mount = new FakeMountClient({
+        [issuePath]: issueFile,
+        '/github/repos/AgentWorkforce/pear/meta.json': { default_branch: 'main' },
+      }, githubWrite)
+      const fleet = new CompletingRemoteFleetClient()
+      const output = buffer()
+      const errors = buffer()
+
+      const code = await runFleetCli([
+        'dispatch',
+        'AR-77',
+        '--backend',
+        'relay',
+        '--config',
+        configPath,
+      ], {
+        fleet,
+        mount,
+        stdout: output,
+        stderr: errors,
+        probePrGhRunner: async () => ({ stdout: '[]' }),
+      })
+
+      expect(code).toBe(0)
+      expect(JSON.parse(output.text())).toMatchObject({ issue: { key: 'AR-77' }, dryRun: false })
+      expect(publishes).toEqual([expect.objectContaining({
+        repo: 'AgentWorkforce/pear',
+        headRef: expect.stringMatching(/^factory\/ar-77-agentworkforce-pear-[0-9a-f]{8}$/u),
+      })])
+      expect(fleet.releases.map((release) => release.reason)).toEqual(['issue-human-review', 'issue-human-review'])
+      const firstDispose = fleet.lifecycleOrder.indexOf('dispose')
+      expect(firstDispose).toBeGreaterThanOrEqual(0)
+      expect(fleet.lifecycleOrder.slice(0, firstDispose).every((event) => event.startsWith('release:'))).toBe(true)
+      expect(fleet.lifecycleOrder.slice(firstDispose).every((event) => event === 'dispose')).toBe(true)
+      expect(errors.text()).not.toContain('[factory] error')
+    } finally {
+      await rm(root, { recursive: true, force: true })
+    }
+  })
+
   it('does not infer or preflight clone paths for a maintenance command', async () => {
     const root = await mkdtemp(join(tmpdir(), 'fleet-cli-maintenance-clone-'))
     try {

src/cli/fleet.ts

@@ -444,6 +444,21 @@ async function runFactoryCommand(
     return 0
   }
 
+  if (command.kind === 'factory-dispatch' && globals.backend === 'relay' && !globals.dryRun) {
+    // A relay dispatch is not placement-only. This process becomes the durable
+    // lifecycle publisher (or attaches to the current owner's durable row),
+    // and stays through takeover/publication/writeback/release to terminal.
+    await factory.start({ mode: 'dispatch-owner' })
+    try {
+      const result = await factory.dispatch(decision, { dryRun: false })
+      writeJson(out, result)
+      await factory.waitForDispatchTerminal(result.issue)
+      return 0
+    } finally {
+      await factory.stop()
+    }
+  }
+
   writeJson(out, await factory.dispatch(decision, { dryRun: globals.dryRun }))
   return 0
 }

src/dispatch/templates.ts

@@ -54,6 +54,8 @@ export interface RenderAgentTaskInput {
   }
   /** Pre-rendered writeback instructions for connected integrations. */
   integrationInstructions?: string
+  /** Exact branch Factory will publish after the implementer pushes it. */
+  branchName?: string
   /**
    * Absolute path to the .integrations mount root. The agent runs in its repo
    * clonePath, not the daemon cwd where .integrations lives, so every
@@ -84,7 +86,9 @@ export function renderAgentTask(input: RenderAgentTaskInput): string {
   const common = [
     ...header,
     '',
-    'Create a branch for this issue before editing.',
+    input.branchName
+      ? `Create a branch for this issue before editing. Create or reset the exact branch \`${input.branchName}\` from the repository default branch, then commit and push only this branch.`
+      : 'Create a branch for this issue before editing.',
     'Commit the implementation and tests.',
     'Push the branch to origin.',
     'When implementation is complete, finish your session normally; Factory will open the PR targeting the repository default branch through the connected GitHub workspace.',

src/fleet/internal-fleet-client.ts

@@ -95,6 +95,7 @@ const RELEASE_RETRY_MAX_ATTEMPTS = 3
 const RELEASE_RETRY_BACKOFF_MS = 250
 
 export class InternalFleetClient implements FleetClient {
+  readonly placementLocality = 'local' as const
   readonly #client: HarnessDriverClientLike
   #ownsBroker: boolean
   readonly #ownedBrokerAgentExitTimeoutMs: number
@@ -178,7 +179,14 @@ export class InternalFleetClient implements FleetClient {
     return spawnResultFrom(handle)
   }
 
-  async resume(input: { name?: string; sessionRef: string; node?: 'self' | string; capability?: Capability }): Promise<SpawnResult> {
+  async resume(input: {
+    name?: string
+    sessionRef: string
+    node?: 'self' | string
+    capability?: Capability
+    repo?: string
+    clonePath?: string
+  }): Promise<SpawnResult> {
     assertSelfNode(input.node)
     this.#ensureEventSubscription()
     this.#readyAgentNames.delete(input.name ?? input.sessionRef)

src/fleet/relay-fleet-client.test.ts

@@ -163,7 +163,7 @@ describe('RelayFleetClient', () => {
       cwd: '/checkout',
       invocationId: 'factory-inv-1',
       channel: 'wf-factory',
-    })).resolves.toEqual({ name: 'ar-1-impl', sessionRef: 'session-1', pid: 123 })
+    })).resolves.toEqual({ name: 'ar-1-impl', sessionRef: 'session-1', pid: 123, node: 'mac-mini', locality: 'remote' })
 
     expect(messaging.placements).toHaveLength(1)
     const placement = messaging.placements[0]!

src/fleet/relay-fleet-client.ts

@@ -77,6 +77,7 @@ const DEFAULT_NODE_OFFLINE_GRACE_MS = 90_000
 const DEFAULT_REGISTRATION_GRACE_MS = 60_000
 
 export class RelayFleetClient implements FleetClient {
+  readonly placementLocality = 'remote' as const
   readonly #options: RelayFleetClientOptions
   readonly #agentName: string
   readonly #spawnAckTimeoutMs: number
@@ -164,17 +165,26 @@ export class RelayFleetClient implements FleetClient {
       log: this.#log,
     })
     const invocation = await this.#awaitInvocation(ack.actionName || 'spawn', ack)
-    const result = spawnResultFromInvocation(input.name, input.sessionRef, invocation)
+    const result = spawnResultFromInvocation(input.name, input.sessionRef, invocation, ack)
     this.#track(result.name, ack)
     return result
   }
 
-  async resume(input: { name?: string; sessionRef: string; node?: 'self' | string; capability?: Capability }): Promise<SpawnResult> {
+  async resume(input: {
+    name?: string
+    sessionRef: string
+    node?: 'self' | string
+    capability?: Capability
+    repo?: string
+    clonePath?: string
+  }): Promise<SpawnResult> {
     const name = input.name ?? input.sessionRef
     return await this.spawn({
       name,
       capability: input.capability ?? 'spawn:codex',
       node: input.node,
+      repo: input.repo,
+      clonePath: input.clonePath,
       sessionRef: input.sessionRef,
     })
   }
@@ -204,7 +214,11 @@ export class RelayFleetClient implements FleetClient {
       messaging.nodes.list(),
     ])
     return {
-      agents: agents.map((agent) => ({ name: agent.name })),
+      agents: agents.map((agent) => {
+        const record = asRecord(agent)
+        const node = readString(record, 'node', 'node_id', 'nodeId')
+        return { name: agent.name, ...(node ? { node } : {}) }
+      }),
       nodes: nodes.map((node) => ({
         name: node.name,
         capabilities: normalizeCapabilities(node.capabilities),
@@ -515,7 +529,12 @@ function spawnActionInput(input: SpawnInput): Record<string, unknown> {
   })
 }
 
-function spawnResultFromInvocation(fallbackName: string, fallbackSessionRef: string | undefined, invocation: RelayActionInvocation): SpawnResult {
+function spawnResultFromInvocation(
+  fallbackName: string,
+  fallbackSessionRef: string | undefined,
+  invocation: RelayActionInvocation,
+  ack: RelayActionInvocationAck & { placement?: { node?: string } },
+): SpawnResult {
   const output = asRecord(invocation.output)
   const agent = asRecord(output?.agent)
   const name = readString(output, 'name', 'agent_name', 'agentName') ?? readString(agent, 'name') ?? fallbackName
@@ -524,11 +543,14 @@ function spawnResultFromInvocation(fallbackName: string, fallbackSessionRef: str
     ?? fallbackSessionRef
   const pid = readNumber(output, 'pid') ?? readNumber(agent, 'pid')
   const pids = readNumberArray(output, 'pids') ?? readNumberArray(agent, 'pids')
+  const node = readString(output, 'node') ?? readString(agent, 'node') ?? ack.placement?.node ?? ack.dispatchedNodeId ?? undefined
   return {
     name,
     ...(sessionRef ? { sessionRef } : {}),
     ...(pid !== undefined ? { pid } : {}),
     ...(pids ? { pids } : {}),
+    ...(node ? { node } : {}),
+    locality: 'remote',
   }
 }
 

src/mount/relayfile-github-connection-write.test.ts

@@ -16,6 +16,50 @@ const gitRunnerForBranch = (branch: string): GitCommandRunner => vi.fn(async (ar
 })
 
 describe('RelayfileGithubConnectionWrite', () => {
+  it('publishes an already-pushed remote branch without reading an orchestrator-local clone', async () => {
+    const pullRequestPath = '/github/repos/AgentWorkforce/factory/pull-requests/factory-factory-ar-85-agentworkforce-factory-pushed.json'
+    class ReceiptMount extends FakeMountClient {
+      override async writeFile(path: string, content: unknown, opts?: { guarded?: boolean }): Promise<void> {
+        await super.writeFile(path, content, opts)
+        if (path === pullRequestPath) {
+          this.files.set(path, { content: { created: 85, url: 'https://github.com/AgentWorkforce/factory/pull/85' } })
+        }
+      }
+    }
+    const mount = new ReceiptMount()
+    const git = vi.fn(async () => { throw new Error('remote publication must not inspect local git') })
+    const write = new RelayfileGithubConnectionWrite({ mount, gitRunner: git })
+
+    const input = {
+      repo: 'AgentWorkforce/factory',
+      headRef: 'factory/ar-85-agentworkforce-factory',
+      baseRef: 'main',
+      title: 'Issue 85',
+      body: 'Fixes #85',
+    }
+    await expect(write.publishPullRequest(input)).resolves.toEqual({
+      repo: 'AgentWorkforce/factory',
+      number: 85,
+      url: 'https://github.com/AgentWorkforce/factory/pull/85',
+      headRef: 'factory/ar-85-agentworkforce-factory',
+      headSha: undefined,
+    })
+    expect(git).not.toHaveBeenCalled()
+    expect(mount.writes).toEqual([{
+      path: pullRequestPath,
+      content: {
+        title: 'Issue 85',
+        head: 'factory/ar-85-agentworkforce-factory',
+        base: 'main',
+        body: 'Fixes #85',
+        author: 'app',
+      },
+    }])
+
+    await expect(write.publishPullRequest(input)).resolves.toMatchObject({ number: 85 })
+    expect(mount.writes[1]?.path).toBe(pullRequestPath)
+  })
+
   it('pushes the current ref before creating a pull request through Relayfile', async () => {
     const draft = 'factory-fix-issue-52-1234567890ab'
     const pullRequestPath = `/github/repos/AgentWorkforce/factory/pull-requests/${draft}.json`

src/mount/relayfile-github-connection-write.ts

@@ -42,8 +42,15 @@ export class RelayfileGithubConnectionWrite implements GithubConnectionWrite {
 
   async publishPullRequest(input: GithubPublishPullRequestInput): Promise<GithubPublishPullRequestResult> {
     const { owner, repo } = githubRepoParts(input.repo)
-    const headRef = await this.#gitValue(['-C', input.clonePath, 'symbolic-ref', '--short', 'HEAD'], 'current branch')
-    const headSha = await this.#gitValue(['-C', input.clonePath, 'rev-parse', 'HEAD'], 'HEAD commit')
+    const headRef = input.headRef ?? (input.clonePath
+      ? await this.#gitValue(['-C', input.clonePath, 'symbolic-ref', '--short', 'HEAD'], 'current branch')
+      : undefined)
+    const headSha = input.headSha ?? (input.clonePath && !input.headRef
+      ? await this.#gitValue(['-C', input.clonePath, 'rev-parse', 'HEAD'], 'HEAD commit')
+      : undefined)
+    if (!headRef) {
+      throw new Error('GitHub PR publication requires headRef or clonePath')
+    }
     if (headRef === input.baseRef) {
       throw new Error(`Refusing to publish GitHub PR with head equal to base branch: ${headRef}`)
     }
@@ -52,10 +59,14 @@ export class RelayfileGithubConnectionWrite implements GithubConnectionWrite {
     const fullHeadRef = `refs/heads/${headRef}`
     const refPath = `${repoRoot}/refs/${encodeURIComponent(fullHeadRef)}.json`
 
-    await this.#writeAndConfirm(refPath, {
-      ref: fullHeadRef,
-      sha: headSha,
-    })
+    // A remote implementer already pushed its branch. Only synthesize/update
+    // the ref for the legacy local-clone path where the publisher owns HEAD.
+    if (headSha) {
+      await this.#writeAndConfirm(refPath, {
+        ref: fullHeadRef,
+        sha: headSha,
+      })
+    }
 
     const pullRequestPath = `${repoRoot}/pull-requests/${draftName}.json`
     await this.#writeAndConfirm(pullRequestPath, {
@@ -139,9 +150,10 @@ const githubRepoParts = (value: string): { owner: string; repo: string } => {
   return { owner: match[1], repo: match[2] }
 }
 
-const githubDraftName = (headRef: string, headSha: string): string => {
+const githubDraftName = (headRef: string, headSha?: string): string => {
   const branch = headRef.toLowerCase().replace(/[^a-z0-9]+/gu, '-').replace(/^-|-$/gu, '') || 'branch'
-  return `factory-${branch.slice(0, 80)}-${headSha.slice(0, 12)}`
+  const identity = headSha?.slice(0, 12) ?? 'pushed'
+  return `factory-${branch.slice(0, 80)}-${identity}`
 }
 
 const record = (value: unknown): Record<string, unknown> =>