From e73a0829f752df1c91b0da0c70ca537de9e5d165 Mon Sep 17 00:00:00 2001 From: kjgbot Date: Fri, 7 Aug 2026 22:29:49 +0200 Subject: [PATCH 1/4] Widen babysitter intake to routed PRs --- .../factory-feature-guardian/agent.test.ts | 2 +- .agentworkforce/features/manifest.yaml | 38 ++- .agentworkforce/features/verify/procedures.md | 13 + README.md | 40 +++ src/cli/fleet.ts | 5 +- src/config/schema.test.ts | 8 +- src/config/schema.ts | 16 + src/dispatch/templates.test.ts | 10 +- src/dispatch/templates.ts | 16 +- src/github/index.ts | 6 + src/github/routed-pr-babysitter.test.ts | 91 ++++++ src/github/routed-pr-babysitter.ts | 232 +++++++++++++++ src/orchestrator/factory.test.ts | 81 +++++ src/orchestrator/factory.ts | 279 +++++++++++++++++- src/ports/state.ts | 32 ++ src/state/file-state-store.ts | 135 +++++++++ src/state/in-memory-state-store.ts | 83 ++++++ src/state/routed-pr-babysitter-claim.test.ts | 70 +++++ 18 files changed, 1146 insertions(+), 11 deletions(-) create mode 100644 src/github/routed-pr-babysitter.test.ts create mode 100644 src/github/routed-pr-babysitter.ts create mode 100644 src/state/routed-pr-babysitter-claim.test.ts diff --git a/.agentworkforce/agents/factory-feature-guardian/agent.test.ts b/.agentworkforce/agents/factory-feature-guardian/agent.test.ts index 1056a2d1..324c613e 100644 --- a/.agentworkforce/agents/factory-feature-guardian/agent.test.ts +++ b/.agentworkforce/agents/factory-feature-guardian/agent.test.ts @@ -683,7 +683,7 @@ describe('factory-feature-guardian runtime paths', () => { const catalog = await loadFactoryGuardianCatalog(production.ctx); - expect(catalog.features).toHaveLength(310); + expect(catalog.features).toHaveLength(314); expect(catalog.features.flatMap((feature) => feature.locations)).toEqual( expect.arrayContaining(['src/environments/', 'src/triage/', 'src/fleet/', 'src/writeback/']) ); diff --git a/.agentworkforce/features/manifest.yaml b/.agentworkforce/features/manifest.yaml index b8f4a42f..8fef5977 100644 --- a/.agentworkforce/features/manifest.yaml +++ b/.agentworkforce/features/manifest.yaml @@ -1,14 +1,14 @@ version: '1.1' -updated: '2026-08-05' +updated: '2026-08-07' catalog: category_count: 25 - feature_count: 310 + feature_count: 314 tier_counts: 1: 48 - 2: 124 + 2: 125 3: 11 - 4: 52 - 5: 60 + 4: 53 + 5: 62 6: 15 # Every user-facing feature in @agent-relay/factory, categorized and scored. @@ -821,6 +821,27 @@ categories: location: src/orchestrator/factory.ts, src/triage/heuristic.ts, src/dispatch/templates.ts verify_tier: 6 + - id: pr-routed-babysitter-discovery + name: Routed Repository PR Babysitter Discovery + api: babysitter.mode=routed-open-prs + description: Discover open non-draft same-repository PRs only from repos.names, fall back to PR title and body, and account for every excluded, deferred, unreadable, or failed candidate + location: src/github/routed-pr-babysitter.ts, src/orchestrator/factory.ts + verify_tier: 5 + + - id: pr-routed-babysitter-ownership + name: Routed PR Work-Unit Ownership + api: StateStore.claimRoutedPrBabysitter() + description: Atomically claim exact normalized repo and PR work units across routed and issue-created intake before spawn, with durable restart recovery and bounded admission + location: src/ports/state.ts, src/state/in-memory-state-store.ts, src/state/file-state-store.ts, src/orchestrator/factory.ts + verify_tier: 4 + + - id: pr-routed-babysitter-opt-out + name: Routed PR Pre-Write Opt-Out + api: babysitter.excludeLabels / babysitter.excludePullRequests + description: Reject configured PR identities or labeled PRs before spawn and require automatic workers to re-check labels before their first provider write + location: src/config/schema.ts, src/github/routed-pr-babysitter.ts, src/dispatch/templates.ts + verify_tier: 5 + - id: pr-standalone-babysitter-validation name: Standalone Babysitter PR Validation api: readStandalonePullRequest() @@ -2142,6 +2163,13 @@ categories: location: src/config/schema.ts, src/orchestrator/factory.ts verify_tier: 2 + - id: config-babysitter-routed-intake + name: Routed Babysitter Intake Controls + api: factory.config.json#babysitter.mode + description: Keep factory-created intake as the default while explicitly configuring routed discovery, exclusions, and notifications-off behavior + location: src/config/schema.ts, src/orchestrator/factory.ts + verify_tier: 2 + - id: config-merge-policy name: Merge Policy api: factory.config.json#mergePolicy diff --git a/.agentworkforce/features/verify/procedures.md b/.agentworkforce/features/verify/procedures.md index ea3cadc0..fab26dac 100644 --- a/.agentworkforce/features/verify/procedures.md +++ b/.agentworkforce/features/verify/procedures.md @@ -351,6 +351,19 @@ Verify the PR contains the issue key, commits, tests, and review activity; no dr Enable `babysitter.enabled`, open a non-draft PR through the issue-driven path, and confirm only one babysitter starts. Make CI fail or add a review request, confirm only the exact repo/PR owner wakes, then enter the documented destructive critical section and prove Factory ACKs only after the no-submit fence persists. Confirm coalesced activity is delivered once after exit, the babysitter fixes the current PR branch, then send/observe its readiness signal. A mismatched signal or draft/closed PR must not advance the issue. +For widened intake, additionally set `babysitter.mode` to +`routed-open-prs` in a disposable workspace. Provide one eligible PR, one PR +with `factory:skip-babysitter`, one configured `excludePullRequests` identity, +one draft, and one same-number PR in a repository absent from `repos.names`. +Run a dry/read-only sweep and verify the exact discovery counters account for +every candidate without spawning or writing. Then use a fake fleet/state test +to prove the eligible PR is atomically claimed before spawn, two owners cannot +claim it, the opt-outs spawn nothing, only one new routed PR is admitted per +sweep, capacity deferrals are logged, restart restores the exact repo/PR owner, +and the automatic prompt disables notifications and checks labels before its +first provider write. Do not run a provider-writing widened sweep until the +GitHub App identity path is independently validated. + Do not attempt a cross-host active/active control-plane test: the supported ownership topology is multiple Factory processes sharing one same-host `FileStateStore`. Remote relay execution nodes are supported and do not need that directory. ### Standalone babysitter diff --git a/README.md b/README.md index 5ebb9a5e..3ae24d43 100644 --- a/README.md +++ b/README.md @@ -303,6 +303,46 @@ existing PR branch, and always leaves the final review and merge to a human. The command prints a spawn receipt and returns; the PR-keyed task-exit worker continues on the relay broker and reports completion or access blockers there. +### Automatic routed-PR babysitting + +`babysitter.enabled` continues to cover Factory-created PRs by default. To +admit existing PRs independently of issue dispatch, set: + +```json +{ + "babysitter": { + "enabled": true, + "mode": "routed-open-prs", + "excludeLabels": ["factory:skip-babysitter"], + "excludePullRequests": [], + "notifyHumans": false + } +} +``` + +The widened mode scans only repositories named by `repos.names`; other routing +fallbacks do not silently expand the sweep. It admits open, non-draft, +same-repository PRs and uses their title/body as the standalone definition of +done when no Factory issue lifecycle exists. Such PRs never gain an issue state +transition or automatic merge. The exact normalized `owner/repo#number` work +unit is claimed durably before spawn, so issue-driven and routed discovery +cannot start two babysitters for one PR. + +Opt-out labels and `excludePullRequests` are checked before spawn. Automatic +workers re-check the labels before their first and every later provider write. Human-facing +status comments, mentions, and escalation stay disabled unless +`notifyHumans` is explicitly enabled. Admission uses `batchSize` as its active +ceiling and starts no more than one new routed PR per sweep. Every sweep logs +exact scanned, eligible, excluded, incomplete, already-owned, admitted, +capacity-deferred, unchanged, and failed counts; deferred work is retried, not +silently truncated. This bounded intake is deliberate: issue #222 recorded a +GitHub API measurement of 302 open PRs, 248 ready for review, across 41 routed +repositories on 2026-08-07. + +Do not enable `routed-open-prs` until the GitHub App identity path from #221 is +deployed and validated. The code can land first because the default mode remains +`factory-created`. + ### Scheduled sync-fidelity canary `factory canary` is the regression detector for upstream sync drift: if a synced diff --git a/src/cli/fleet.ts b/src/cli/fleet.ts index 63fd4809..5d95f024 100644 --- a/src/cli/fleet.ts +++ b/src/cli/fleet.ts @@ -884,7 +884,10 @@ async function runStandaloneBabysitCommand( crossRepository: pr.crossRepository, maintainerCanModify: pr.maintainerCanModify, }, - standaloneBabysitter: { specSource }, + // The explicit CLI command is itself the human opt-in and preserves its + // existing conversational behavior. Automatic routed sweeps pass the + // safer notification flag from config instead. + standaloneBabysitter: { specSource, notifyHumans: true }, integrationsMountRoot: resolveIntegrationsMountRoot(mount), testGuidance, }) diff --git a/src/config/schema.test.ts b/src/config/schema.test.ts index ad990648..0cb2a748 100644 --- a/src/config/schema.test.ts +++ b/src/config/schema.test.ts @@ -36,7 +36,13 @@ describe('FactoryConfigSchema', () => { reviewer: 'spawn:claude', babysitter: 'spawn:claude', }) - expect(parsed.babysitter).toEqual({ enabled: false }) + expect(parsed.babysitter).toEqual({ + enabled: false, + mode: 'factory-created', + excludeLabels: ['factory:skip-babysitter'], + excludePullRequests: [], + notifyHumans: false, + }) expect(parsed.terminalState).toBe('human-review') expect(parsed.stateIds.humanReview).toBeUndefined() expect(parsed.loop.registryPath).toBe('/tmp/factory-run/factory-loop-registry.json') diff --git a/src/config/schema.ts b/src/config/schema.ts index aed87a56..7af368d6 100644 --- a/src/config/schema.ts +++ b/src/config/schema.ts @@ -121,6 +121,22 @@ const slackSchema = z.object({ const babysitterSchema = z.object({ enabled: z.boolean().default(false), + // Preserve the historical issue-created intake unless an operator + // deliberately opts into the wider routed-repository sweep. Keeping the + // rollout switch separate from `enabled` lets the identity fix in #221 land + // before automated writes are attributed across existing human PRs. + mode: z.enum(['factory-created', 'routed-open-prs']).default('factory-created'), + // This label is an author-controlled hard stop. Discovery reads it before + // spawning an agent, and routed babysitters are instructed to re-check it + // before their first provider write. + excludeLabels: z.array(z.string().trim().min(1)).default(['factory:skip-babysitter']), + excludePullRequests: z.array(z.string().regex( + /^[A-Za-z0-9](?:[A-Za-z0-9_.-]{0,99})\/[A-Za-z0-9_.-]{1,100}#[1-9]\d*$/u, + 'expected owner/repo#number', + )).default([]), + // Routed intake stays quiet until an operator explicitly approves the + // notification shape observed in a read-only sweep. + notifyHumans: z.boolean().default(false), }).default({}) const reportingSchema = z.object({ diff --git a/src/dispatch/templates.test.ts b/src/dispatch/templates.test.ts index 7aa9fdac..f877c7f3 100644 --- a/src/dispatch/templates.test.ts +++ b/src/dispatch/templates.test.ts @@ -186,7 +186,11 @@ describe('renderAgentTask', () => { baseRef: 'main', headRepo: 'AgentWorkforce/hoopsheet', }, - standaloneBabysitter: { specSource: 'pull-request' }, + standaloneBabysitter: { + specSource: 'pull-request', + excludeLabels: ['factory:skip-babysitter'], + notifyHumans: false, + }, integrationsMountRoot: '/workspace/.integrations', }) @@ -210,6 +214,10 @@ describe('renderAgentTask', () => { expect(task).toContain('push the same PR head') expect(task).toContain('re-read the live merge state and fresh checks') expect(task).toContain('never merge it yourself') + expect(task).toContain('Before your first provider write') + expect(task).toContain('factory:skip-babysitter') + expect(task).toContain('Do not post status comments, mention humans, send notifications, or escalate') + expect(task).not.toContain('proactively offer to discuss') expect(task).toContain('Never search for, read, or substitute credentials or tokens') expect(task).toContain('output `/exit` on its own line') expect(task).not.toContain('DM `broker`') diff --git a/src/dispatch/templates.ts b/src/dispatch/templates.ts index fee09e7f..19ee2fda 100644 --- a/src/dispatch/templates.ts +++ b/src/dispatch/templates.ts @@ -47,6 +47,10 @@ export interface RenderAgentTaskInput { */ standaloneBabysitter?: { specSource: 'pull-request' | 'linked-issue' + /** Labels which abort this run before its first provider write. */ + excludeLabels?: string[] + /** Human-facing comments, mentions, and escalation are opt-in for sweeps. */ + notifyHumans?: boolean } slackDispatchThread?: { channel: string @@ -222,6 +226,12 @@ export function renderAgentTask(input: RenderAgentTaskInput): string { const standaloneMergePolicy = input.config.mergePolicy === 'on-green-with-review' ? 'Merge policy: on-green-with-review. This standalone run has no guarded merge executor, so never merge the PR yourself; leave the final merge to a human.' : 'Merge policy: never - leave the PR open for human review and approval; never merge it yourself.' + const optOutLine = input.standaloneBabysitter.excludeLabels?.length + ? `Before your first provider write and again before every later provider write, re-read the live PR labels. If any label in JSON ${JSON.stringify(input.standaloneBabysitter.excludeLabels)} is present, make no further provider writes, report the opt-out, and exit.` + : undefined + const notificationLine = input.standaloneBabysitter.notifyHumans + ? 'You may notify or mention humans when a concrete decision is required.' + : 'Do not post status comments, mention humans, send notifications, or escalate this run. Only write the code, commits, pushes, and direct review-thread replies required to shepherd the PR.' return [ `GitHub repo: ${repo}`, cloneInstruction, @@ -229,6 +239,8 @@ export function renderAgentTask(input: RenderAgentTaskInput): string { '', `You are the standalone PR babysitter for ${prRef}.`, 'Your job: drive this PR to genuinely green and correct against the definition of done above, then hand it to a human. Do NOT merge it yourself.', + ...(optOutLine ? [optOutLine] : []), + notificationLine, 'Fix things directly and aggressively: inspect the existing implementation, make substantive corrections, and keep the PR scope anchored to the definition of done.', ...(branchLine ? [branchLine] : []), checkoutLine, @@ -242,7 +254,9 @@ export function renderAgentTask(input: RenderAgentTaskInput): string { 'After every push, wait for the checks on the newly pushed head commit. Never reuse green results from an older commit when declaring the PR ready.', 'Commit and push fixes only to the existing PR head branch. Use a normal push when possible; if rebasing requires rewriting the PR head, use `--force-with-lease`, never an unconditional force push.', 'If the push is denied, stop and report the access blocker. Never search for, read, or substitute credentials or tokens, and never modify Git/GitHub authentication configuration.', - 'If a human can be reached, proactively offer to discuss the PR status, trade-offs, and open questions.', + ...(input.standaloneBabysitter.notifyHumans + ? ['If a human can be reached, proactively offer to discuss the PR status, trade-offs, and open questions.'] + : []), 'When the PR is green — no failing CI, no merge conflicts, and every review comment addressed — report a concise completion summary and output `/exit` on its own line so the Agent Relay task-exit lifecycle closes cleanly.', standaloneFinishLine, standaloneMergePolicy, diff --git a/src/github/index.ts b/src/github/index.ts index 72b6b025..5b4e5268 100644 --- a/src/github/index.ts +++ b/src/github/index.ts @@ -13,6 +13,11 @@ export { readStandalonePullRequest, standaloneBabysitterAgentName, } from './standalone-babysitter' +export { + discoverRoutedPullRequests, + routedPrIdentity, + routedPrRepos, +} from './routed-pr-babysitter' export type { GhRunner, GhRunResult, @@ -30,3 +35,4 @@ export type { StandaloneBabysitTarget, StandalonePullRequest, } from './standalone-babysitter' +export type { RoutedPrCandidate, RoutedPrDiscoveryReport } from './routed-pr-babysitter' diff --git a/src/github/routed-pr-babysitter.test.ts b/src/github/routed-pr-babysitter.test.ts new file mode 100644 index 00000000..84194e46 --- /dev/null +++ b/src/github/routed-pr-babysitter.test.ts @@ -0,0 +1,91 @@ +import { describe, expect, it } from 'vitest' + +import { FactoryConfigSchema } from '../config/schema' +import type { MountClient } from '../ports' +import { discoverRoutedPullRequests, routedPrRepos } from './routed-pr-babysitter' + +const config = (babysitter: Record = {}) => FactoryConfigSchema.parse({ + repos: { + org: 'AgentWorkforce', + names: ['pear'], + byLabel: { legacy: 'AgentWorkforce/legacy' }, + }, + babysitter: { enabled: true, mode: 'routed-open-prs', ...babysitter }, +}) + +const pull = (overrides: Record = {}) => ({ + number: 7, + title: 'Repair routed babysitting', + body: 'Use the PR as the definition of done.', + state: 'OPEN', + isDraft: false, + merged: false, + url: 'https://github.com/AgentWorkforce/pear/pull/7', + headRefName: 'fix/routed-babysitting', + headRefOid: 'abc123', + baseRefName: 'main', + headRepository: { nameWithOwner: 'AgentWorkforce/pear' }, + isCrossRepository: false, + labels: [], + ...overrides, +}) + +const mount = (files: Record): MountClient => ({ + listTree: async (prefix: string) => Object.keys(files).filter((path) => path.startsWith(prefix)), + readFile: async (path: string) => ({ content: files[path] }), +} as unknown as MountClient) + +describe('routed PR babysitter discovery', () => { + it('bounds widened intake strictly to repos.names', () => { + expect(routedPrRepos(config())).toEqual(['AgentWorkforce/pear']) + }) + + it('deduplicates mount aliases and returns complete same-repository PRs', async () => { + const files = { + '/github/repos/AgentWorkforce/pear/pulls/7/meta.json': pull(), + '/github/repos/AgentWorkforce__pear/pulls/by-id/7.json': pull(), + '/github/repos/AgentWorkforce__legacy/pulls/by-id/8.json': pull({ number: 8 }), + } + const report = await discoverRoutedPullRequests(mount(files), config()) + expect(report.scanned).toBe(2) + expect(report.duplicates).toBe(1) + expect(report.candidates.map(({ repo, number }) => `${repo}#${number}`)).toEqual([ + 'AgentWorkforce/pear#7', + ]) + }) + + it('honours label and config opt-outs before admission', async () => { + const files = { + '/github/repos/AgentWorkforce__pear/pulls/by-id/7.json': pull({ + labels: [{ name: 'factory:skip-babysitter' }], + }), + } + const byLabel = await discoverRoutedPullRequests(mount(files), config()) + expect(byLabel.excluded).toBe(1) + expect(byLabel.candidates).toEqual([]) + + const byIdentity = await discoverRoutedPullRequests( + mount({ ...files, '/github/repos/AgentWorkforce__pear/pulls/by-id/7.json': pull() }), + config({ excludePullRequests: ['AgentWorkforce/pear#7'] }), + ) + expect(byIdentity.excluded).toBe(1) + expect(byIdentity.candidates).toEqual([]) + }) + + it('fails closed on incomplete, draft, and cross-repository metadata', async () => { + const files = { + '/github/repos/AgentWorkforce__pear/pulls/by-id/7.json': pull({ headRefOid: undefined }), + '/github/repos/AgentWorkforce__pear/pulls/by-id/8.json': pull({ number: 8, isDraft: true }), + '/github/repos/AgentWorkforce__pear/pulls/by-id/9.json': pull({ + number: 9, + headRepository: { nameWithOwner: 'contributor/pear' }, + isCrossRepository: true, + }), + } + const report = await discoverRoutedPullRequests(mount(files), config()) + expect(report.incomplete).toBe(1) + expect(report.terminal).toBe(1) + expect(report.crossRepository).toBe(1) + expect(report.candidates).toEqual([]) + }) +}) diff --git a/src/github/routed-pr-babysitter.ts b/src/github/routed-pr-babysitter.ts new file mode 100644 index 00000000..3e9dd11a --- /dev/null +++ b/src/github/routed-pr-babysitter.ts @@ -0,0 +1,232 @@ +import type { FactoryConfig } from '../config/schema' +import type { MountClient } from '../ports' +import { asRecord, stableHash, wrappedPayload } from '../writeback/shared' + +export type RoutedPrCandidate = { + repo: string + number: number + title: string + body: string + state: string + draft: boolean + merged: boolean + url: string + headRef: string + headSha: string + baseRef: string + headRepo: string + crossRepository: boolean + maintainerCanModify?: boolean + labels: string[] + filesChanged?: string[] + path: string + revision: string +} + +export type RoutedPrDiscoveryReport = { + scanned: number + eligible: number + excluded: number + incomplete: number + terminal: number + crossRepository: number + duplicates: number + failures: Array<{ repo: string; prNumber?: number; reason: string }> + candidates: RoutedPrCandidate[] +} + +export const routedPrIdentity = (repo: string, prNumber: number): string => + `${repo.toLowerCase()}#${prNumber}` + +export const routedPrRepos = (config: FactoryConfig): string[] => { + const repos = new Map() + for (const name of config.repos.names ?? []) { + const configured = config.repos.byLabel[name] ?? name + const repo = configured.includes('/') + ? configured + : config.repos.org + ? `${config.repos.org}/${configured}` + : undefined + if (!repo || !/^[^/]+\/[^/]+$/u.test(repo)) continue + repos.set(repo.toLowerCase(), repo) + } + return [...repos.values()].sort((left, right) => left.localeCompare(right)) +} + +export async function discoverRoutedPullRequests( + mount: MountClient, + config: FactoryConfig, +): Promise { + const report: RoutedPrDiscoveryReport = { + scanned: 0, + eligible: 0, + excluded: 0, + incomplete: 0, + terminal: 0, + crossRepository: 0, + duplicates: 0, + failures: [], + candidates: [], + } + const seen = new Set() + const excludedLabels = new Set(config.babysitter.excludeLabels.map((label) => label.trim().toLowerCase())) + const excludedPrs = new Set(config.babysitter.excludePullRequests.map((identity) => identity.toLowerCase())) + + for (const repo of routedPrRepos(config)) { + const [owner, name] = repo.split('/') + const roots = [ + `/github/repos/${owner}/${name}/pulls/`, + `/github/repos/${owner}__${name}/pulls/by-id/`, + ] + const paths = new Set() + try { + for (const root of roots) { + for (const path of await mount.listTree(root)) paths.add(path) + } + } catch (error) { + report.failures.push({ repo, reason: error instanceof Error ? error.message : String(error) }) + continue + } + for (const path of [...paths].sort()) { + const pathNumber = pullNumberFromPath(path, owner!, name!) + if (!pathNumber) continue + report.scanned += 1 + const identity = routedPrIdentity(repo, pathNumber) + if (seen.has(identity)) { + report.duplicates += 1 + continue + } + seen.add(identity) + let candidate: RoutedPrCandidate | undefined + try { + candidate = parseRoutedPrCandidate((await mount.readFile(path)).content, repo, pathNumber, path) + } catch (error) { + report.failures.push({ repo, prNumber: pathNumber, reason: error instanceof Error ? error.message : String(error) }) + continue + } + if (!candidate) { + report.incomplete += 1 + report.failures.push({ repo, prNumber: pathNumber, reason: 'incomplete authoritative PR metadata' }) + continue + } + if (candidate.merged || candidate.state !== 'OPEN' || candidate.draft) { + report.terminal += 1 + continue + } + if (candidate.crossRepository || candidate.headRepo.toLowerCase() !== repo.toLowerCase()) { + report.crossRepository += 1 + continue + } + if (excludedPrs.has(identity) || candidate.labels.some((label) => excludedLabels.has(label.toLowerCase()))) { + report.excluded += 1 + continue + } + report.eligible += 1 + report.candidates.push(candidate) + } + } + report.candidates.sort((left, right) => + left.repo.localeCompare(right.repo) || left.number - right.number + ) + return report +} + +const pullNumberFromPath = (path: string, owner: string, repo: string): number | undefined => { + const match = path.match( + /^\/github\/repos\/(?:([^/]+)\/([^/]+)|([^/]+)__([^/]+))\/pulls\/(?:by-id\/)?(\d+)(?:__[^/]*)?(?:\/(?:meta|metadata)\.json|\.json)$/u, + ) + const actualOwner = match?.[1] ?? match?.[3] + const actualRepo = match?.[2] ?? match?.[4] + const number = Number(match?.[5]) + return actualOwner?.toLowerCase() === owner.toLowerCase() && + actualRepo?.toLowerCase() === repo.toLowerCase() && + Number.isSafeInteger(number) && number > 0 + ? number + : undefined +} + +const parseRoutedPrCandidate = ( + content: unknown, + repo: string, + number: number, + path: string, +): RoutedPrCandidate | undefined => { + const payload = wrappedPayload(content) + const explicitNumber = numberValue(payload.number) + if (explicitNumber !== undefined && explicitNumber !== number) return undefined + const head = asRecord(payload.head) + const base = asRecord(payload.base) + const headRepository = asRecord(payload.headRepository) + const headOwner = asRecord(payload.headRepositoryOwner) + const headRepo = stringValue(asRecord(head?.repo)?.full_name) ?? + stringValue(headRepository?.nameWithOwner) ?? + (() => { + const name = stringValue(headRepository?.name) + const owner = stringValue(headOwner?.login) ?? stringValue(headOwner?.name) + return name && owner ? `${owner}/${name}` : undefined + })() + const title = stringValue(payload.title) + const state = stringValue(payload.state)?.toUpperCase() + const draft = booleanValue(payload.isDraft) ?? booleanValue(payload.draft) + const merged = booleanValue(payload.merged) ?? state === 'MERGED' + const headRef = stringValue(payload.headRefName) ?? stringValue(head?.ref) ?? stringValue(payload.head_ref) + const headSha = stringValue(payload.headRefOid) ?? stringValue(head?.sha) + const baseRef = stringValue(payload.baseRefName) ?? stringValue(base?.ref) + const crossRepository = booleanValue(payload.isCrossRepository) ?? + booleanValue(payload.crossRepository) ?? + (headRepo ? headRepo.toLowerCase() !== repo.toLowerCase() : undefined) + if (!title || !state || draft === undefined || merged === undefined || !headRef || !headSha || !baseRef || !headRepo || crossRepository === undefined || !Array.isArray(payload.labels)) { + return undefined + } + const labels = labelNames(payload.labels) + const body = stringValue(payload.body) ?? '' + const candidate = { + repo, + number, + title, + body, + state, + draft, + merged, + url: stringValue(payload.url) ?? stringValue(payload.html_url) ?? `https://github.com/${repo}/pull/${number}`, + headRef, + headSha, + baseRef, + headRepo, + crossRepository, + maintainerCanModify: booleanValue(payload.maintainerCanModify) ?? booleanValue(payload.maintainer_can_modify), + labels, + filesChanged: changedFiles(payload.filesChanged ?? payload.files), + path, + } + return { + ...candidate, + revision: stableHash(JSON.stringify({ + headSha: candidate.headSha, + labels: [...candidate.labels].sort(), + state: candidate.state, + draft: candidate.draft, + body: candidate.body, + title: candidate.title, + updatedAt: stringValue(payload.updatedAt) ?? stringValue(payload.updated_at), + reviewDecision: stringValue(payload.reviewDecision) ?? stringValue(payload.review_decision), + statusCheckRollup: payload.statusCheckRollup ?? payload.status_check_rollup, + })), + } +} + +const labelNames = (value: unknown): string[] => Array.isArray(value) + ? value.map((entry) => typeof entry === 'string' ? entry : stringValue(asRecord(entry)?.name)) + .filter((entry): entry is string => Boolean(entry)) + : [] + +const changedFiles = (value: unknown): string[] | undefined => { + if (!Array.isArray(value)) return undefined + const paths = value.map((entry) => typeof entry === 'string' ? entry : stringValue(asRecord(entry)?.path)) + .filter((entry): entry is string => Boolean(entry)) + return paths.length > 0 ? paths : undefined +} + +const stringValue = (value: unknown): string | undefined => typeof value === 'string' && value.trim() ? value : undefined +const booleanValue = (value: unknown): boolean | undefined => typeof value === 'boolean' ? value : undefined +const numberValue = (value: unknown): number | undefined => typeof value === 'number' && Number.isSafeInteger(value) ? value : undefined diff --git a/src/orchestrator/factory.test.ts b/src/orchestrator/factory.test.ts index 6ff90614..a9ace169 100644 --- a/src/orchestrator/factory.test.ts +++ b/src/orchestrator/factory.test.ts @@ -16241,6 +16241,87 @@ describe('FactoryLoop PR babysitter', () => { const humanReviewStateId = 'state-human-review' const inPlanning = '3de351f2-90e6-4731-aa6b-4a55b77f481e' + it('discovers one opted-in routed PR without issue dispatch and keeps notifications off', async () => { + const mount = new FakeMountClient({ + '/github/repos/AgentWorkforce__pear/pulls/by-id/701.json': { + number: 701, + title: 'Repair a human-authored PR', + body: 'Use this PR body as the definition of done.', + state: 'OPEN', + isDraft: false, + merged: false, + url: 'https://github.com/AgentWorkforce/pear/pull/701', + headRefName: 'human/repair-pr', + headRefOid: 'head-701', + baseRefName: 'main', + headRepository: { nameWithOwner: 'AgentWorkforce/pear' }, + isCrossRepository: false, + labels: [], + }, + }) + const fleet = new FakeFleetClient() + const factory = createFactory(config({ + issueSource: 'github', + repos: { + org: 'AgentWorkforce', + names: ['pear'], + clonePaths: { 'AgentWorkforce/pear': '/work/pear' }, + default: 'AgentWorkforce/pear', + }, + babysitter: { + enabled: true, + mode: 'routed-open-prs', + excludeLabels: ['factory:skip-babysitter'], + excludePullRequests: [], + notifyHumans: false, + }, + }), { mount, fleet, triage: new StaticTriage() }) + + await factory.runOnce() + + expect(fleet.spawns).toHaveLength(1) + expect(fleet.spawns[0]?.invocationId).toBe('factory-babysit:agentworkforce/pear#701') + expect(fleet.spawns[0]?.task).toContain('PR title JSON (definition of done): "Repair a human-authored PR"') + expect(fleet.spawns[0]?.task).toContain('Before your first provider write') + expect(fleet.spawns[0]?.task).toContain('Do not post status comments, mention humans, send notifications, or escalate') + expect(fleet.spawns[0]?.task).not.toContain('proactively offer to discuss') + }) + + it('checks routed PR opt-out before spawning', async () => { + const mount = new FakeMountClient({ + '/github/repos/AgentWorkforce__pear/pulls/by-id/702.json': { + number: 702, + title: 'Leave this PR alone', + body: '', + state: 'OPEN', + isDraft: false, + merged: false, + headRefName: 'human/no-babysitter', + headRefOid: 'head-702', + baseRefName: 'main', + headRepository: { nameWithOwner: 'AgentWorkforce/pear' }, + isCrossRepository: false, + labels: [{ name: 'factory:skip-babysitter' }], + }, + }) + const fleet = new FakeFleetClient() + const factory = createFactory(config({ + issueSource: 'github', + repos: { org: 'AgentWorkforce', names: ['pear'], clonePaths: { 'AgentWorkforce/pear': '/work/pear' } }, + babysitter: { + enabled: true, + mode: 'routed-open-prs', + excludeLabels: ['factory:skip-babysitter'], + excludePullRequests: [], + notifyHumans: false, + }, + }), { mount, fleet, triage: new StaticTriage() }) + + await factory.runOnce() + + expect(fleet.spawns).toEqual([]) + }) + const babysitterConfig = (overrides: FactoryConfigOverrides = {}): FactoryConfig => config({ babysitter: { enabled: true }, terminalState: 'human-review', diff --git a/src/orchestrator/factory.ts b/src/orchestrator/factory.ts index 927ec21a..8bfbd346 100644 --- a/src/orchestrator/factory.ts +++ b/src/orchestrator/factory.ts @@ -5,7 +5,16 @@ import { dirname, isAbsolute, resolve } from 'node:path' import { FactoryConfigSchema, type FactoryConfig } from '../config/schema' import { linearByStatePath, linearByIdPath, linearByUuidPath } from '../constants/linear' import { stateResolutionFromIds, type FactoryStateResolution } from '../linear/state-resolver' -import { GithubMergeGate, closeProbePr, type GhRunner, type GithubMergeGate as GithubMergeGatePort } from '../github' +import { + GithubMergeGate, + closeProbePr, + discoverRoutedPullRequests, + routedPrIdentity, + standaloneBabysitterAgentName, + type GhRunner, + type GithubMergeGate as GithubMergeGatePort, + type RoutedPrCandidate, +} from '../github' import { VerificationPipeline, type VerificationGate } from '../environments/verification-pipeline' import type { AgentMessage, @@ -282,6 +291,7 @@ const INJECTION_RETRY_DELAY_MS = 1_000 const INJECTION_RETRY_ATTEMPT_TIMEOUT_MS = 15_000 const INJECTION_MAX_ATTEMPTS = 6 const BABYSITTER_EVENT_COALESCE_MS = 750 +const ROUTED_PR_BABYSITTER_CLAIM_LEASE_MS = 15 * 60_000 const BABYSITTER_EVENT_RETRY_MS = 1_000 const BABYSITTER_SUBSCRIPTION_TTL_SECONDS = 60 * 60 // Relayfile receives provider-native GitHub events, not the materialized file @@ -441,6 +451,9 @@ export class FactoryLoop implements Factory { readonly #clarificationWakeRetryTimers = new Map>() readonly #clarificationWakeOwner = `${process.pid}:${randomUUID()}` readonly #dispatchLifecycleOwner = `${process.pid}:${randomUUID()}` + readonly #routedPrBabysitterOwner = `${process.pid}:${randomUUID()}` + readonly #routedPrBabysitterAgents = new Map() + #routedPrBabysitterSweep?: Promise readonly #dispatchLifecycleEpochs = new Map() readonly #dispatchTerminalWaiters = new Map void>>() readonly #dispatchLifecycleRetryTimers = new Map>() @@ -804,6 +817,7 @@ export class FactoryLoop implements Factory { try { this.#wireFleetEvents() await this.#adoptInFlightAgents(legacyRegistry) + await this.#restoreRoutedPrBabysitterClaims() this.#startupAgentAdoptionActive = false if (opts.mode !== 'dispatch-owner') await this.#reapOrphanedWorktreesOnStartup(legacyRegistry) if (this.#config.babysitter.enabled) { @@ -927,6 +941,7 @@ export class FactoryLoop implements Factory { this.#clarificationIntents.clear() await this.#drainBabysitterWakesForStop() + await this.#routedPrBabysitterSweep await this.#drainAgentExitsInFlight() // Durable relay placements must survive an owner restart so a successor @@ -946,6 +961,7 @@ export class FactoryLoop implements Factory { this.#babysitterSubscriptionOwners.clear() this.#babysitterReady.clear() this.#babysitterCriticalAgents.clear() + this.#routedPrBabysitterAgents.clear() const subscription = this.#subscription this.#subscription = undefined await this.#boundedStopTeardown('factory subscription unsubscribe', () => subscription?.unsubscribe()) @@ -1246,6 +1262,7 @@ export class FactoryLoop implements Factory { if (!page.nextCursor || page.nextCursor === cursor) break cursor = page.nextCursor } + await this.#sweepRoutedPrBabysitters() } catch (error) { this.#logger.warn?.('[factory] live subscription poll failed', error) } finally { @@ -1535,6 +1552,7 @@ export class FactoryLoop implements Factory { // while the event loop was busy (for example during a large startup pull). // Keep reconciliation active even when babysitters own PR completion. await this.#fleet.reconcileTrackedAgents?.() + await this.#sweepRoutedPrBabysitters() // When the babysitter owns PR-open, completion is driven by PR webhooks + // the babysitter's readiness signal (see #handlePrChange / #handleAgentExit), // not this polling sweep. Disabling it here is what makes the babysitter path @@ -1669,6 +1687,199 @@ export class FactoryLoop implements Factory { return undefined } + async #restoreRoutedPrBabysitterClaims(): Promise { + if (this.#config.babysitter.mode !== 'routed-open-prs') return + const roster = await this.#fleet.roster() + const online = new Set(roster.agents.map((agent) => agent.name)) + for (const [identity, claim] of await this.#state.listRoutedPrBabysitterClaims(this.#workspaceId)) { + if (claim.status === 'running' && claim.agentName && online.has(claim.agentName)) { + this.#routedPrBabysitterAgents.set(claim.agentName, identity) + this.#increment('routedPrBabysitterClaimsRestored') + } else if (claim.status !== 'complete' && claim.leaseUntilMs > this.#clock.now()) { + this.#logger.warn?.('[factory] routed PR babysitter claim awaits lease expiry after missing agent', { + repo: claim.repo, + prNumber: claim.prNumber, + leaseUntilMs: claim.leaseUntilMs, + }) + } + } + } + + async #sweepRoutedPrBabysitters(dryRun = this.#config.dryRun): Promise { + if (!this.#config.babysitter.enabled || this.#config.babysitter.mode !== 'routed-open-prs' || this.#stopping) return + if (this.#routedPrBabysitterSweep) return this.#routedPrBabysitterSweep + const sweep = this.#runRoutedPrBabysitterSweep(dryRun).finally(() => { + if (this.#routedPrBabysitterSweep === sweep) this.#routedPrBabysitterSweep = undefined + }) + this.#routedPrBabysitterSweep = sweep + return sweep + } + + async #runRoutedPrBabysitterSweep(dryRun: boolean): Promise { + const discovery = await discoverRoutedPullRequests(this.#mount, this.#config) + const priorClaims = await this.#state.listRoutedPrBabysitterClaims(this.#workspaceId) + for (const [identity, claim] of priorClaims) { + if ( + claim.owner === this.#routedPrBabysitterOwner && + claim.status === 'running' && + claim.agentName && + this.#routedPrBabysitterAgents.get(claim.agentName) === identity + ) { + await this.#state.claimRoutedPrBabysitter( + this.#workspaceId, + identity, + { + repo: claim.repo, + prNumber: claim.prNumber, + revision: claim.revision, + source: claim.source, + }, + this.#routedPrBabysitterOwner, + this.#clock.now(), + ROUTED_PR_BABYSITTER_CLAIM_LEASE_MS, + this.#config.batchSize, + ) + } + } + const stats = { + scanned: discovery.scanned, + eligible: discovery.eligible, + excluded: discovery.excluded, + incomplete: discovery.incomplete, + terminal: discovery.terminal, + crossRepository: discovery.crossRepository, + duplicates: discovery.duplicates, + failures: discovery.failures.length, + alreadyOwned: 0, + admitted: 0, + capacityDeferred: 0, + unchanged: 0, + } + for (const failure of discovery.failures) { + this.#logger.warn?.('[factory] routed PR babysitter discovery dropped candidate', failure) + } + const issueSessions = await this.#state.listBabysitterSessions(this.#workspaceId) + const issueOwned = new Set(issueSessions.map(([, session]) => routedPrIdentity(session.repo, session.prNumber))) + for (const candidate of discovery.candidates) { + const identity = routedPrIdentity(candidate.repo, candidate.number) + if (issueOwned.has(identity)) { + stats.alreadyOwned += 1 + continue + } + if (dryRun) continue + const claim = await this.#state.claimRoutedPrBabysitter( + this.#workspaceId, + identity, + { + repo: candidate.repo, + prNumber: candidate.number, + revision: candidate.revision, + source: 'routed-open-prs', + }, + this.#routedPrBabysitterOwner, + this.#clock.now(), + ROUTED_PR_BABYSITTER_CLAIM_LEASE_MS, + this.#config.batchSize, + ) + if (claim.outcome === 'capacity') { + stats.capacityDeferred += 1 + continue + } + if (claim.outcome === 'already-running' || claim.outcome === 'owned') { + stats.alreadyOwned += 1 + continue + } + if (claim.outcome === 'unchanged') { + stats.unchanged += 1 + continue + } + try { + await this.#spawnRoutedPrBabysitter(candidate, identity) + stats.admitted += 1 + } catch (error) { + await this.#state.releaseRoutedPrBabysitterClaim( + this.#workspaceId, + identity, + this.#routedPrBabysitterOwner, + ) + this.#increment('routedPrBabysitterSpawnFailures') + this.#logger.warn?.('[factory] routed PR babysitter spawn failed', { + repo: candidate.repo, + prNumber: candidate.number, + error: describeError(error).errorMessage, + }) + } + // Pacing is intentionally one new admission per sweep. Every remaining + // eligible candidate is explicitly accounted as deferred, never dropped. + stats.capacityDeferred += discovery.candidates.length - + stats.alreadyOwned - stats.unchanged - stats.admitted - stats.capacityDeferred + break + } + this.#logger.info?.('[factory] routed PR babysitter sweep completed', stats) + } + + async #spawnRoutedPrBabysitter(candidate: RoutedPrCandidate, identity: string): Promise { + const clonePath = this.#config.repos.clonePaths[candidate.repo] + const issue = { + key: `${candidate.repo}#${candidate.number}`, + title: candidate.title, + description: candidate.body || '(No PR description was provided.)', + } + const testGuidance = await resolveTestGuidance({ + repoPath: clonePath, + issue, + changedFiles: candidate.filesChanged, + }) + const name = standaloneBabysitterAgentName(candidate.repo, candidate.number) + const task = renderAgentTask({ + issue, + route: { repo: candidate.repo, clonePath }, + role: 'babysitter', + config: { mergePolicy: 'never', terminalState: 'human-review' }, + reviewerName: '', + pr: { + number: candidate.number, + url: candidate.url, + headRef: candidate.headRef, + headSha: candidate.headSha, + baseRef: candidate.baseRef, + headRepo: candidate.headRepo, + crossRepository: candidate.crossRepository, + maintainerCanModify: candidate.maintainerCanModify, + }, + standaloneBabysitter: { + specSource: 'pull-request', + excludeLabels: this.#config.babysitter.excludeLabels, + notifyHumans: this.#config.babysitter.notifyHumans, + }, + integrationsMountRoot: this.#integrationsMountRoot(), + testGuidance, + }) + const result = await this.#fleet.spawn({ + name, + capability: this.#config.agentCapabilities.babysitter, + node: 'self', + repo: candidate.repo, + clonePath, + task, + model: this.#config.models.babysitter, + cwd: clonePath, + invocationId: `factory-babysit:${identity}`, + }) + if (!await this.#state.markRoutedPrBabysitterRunning( + this.#workspaceId, + identity, + this.#routedPrBabysitterOwner, + result.name, + this.#clock.now(), + )) { + await this.#fleet.release(result.name, 'routed-pr-claim-lost') + throw new Error(`Routed PR babysitter claim lost after spawning ${identity}`) + } + this.#routedPrBabysitterAgents.set(result.name, identity) + this.#increment('routedPrBabysittersSpawned') + } + async runOnce(opts: { dryRun?: boolean } = {}): Promise { const dryRun = opts.dryRun ?? this.#config.dryRun const startedAtMs = this.#clock.now() @@ -1678,6 +1889,7 @@ export class FactoryLoop implements Factory { this.#logger.info?.('[factory] run-once started', { dryRun }) let report: IterationReport | undefined try { + await this.#sweepRoutedPrBabysitters(dryRun) this.#dependencyIssues.clear() // Terminal observations are only a live-cycle cache. Rebuild them from // current provider snapshots (or merged PR metadata) so a reopened issue @@ -5693,6 +5905,20 @@ export class FactoryLoop implements Factory { return } + const routedIdentity = this.#routedPrBabysitterAgents.get(name) + if (routedIdentity) { + this.#routedPrBabysitterAgents.delete(name) + await this.#state.completeRoutedPrBabysitter( + this.#workspaceId, + routedIdentity, + name, + this.#clock.now(), + ) + this.#increment('routedPrBabysittersCompleted') + await this.#sweepRoutedPrBabysitters() + return + } + // Agent messages and exits are separate fleet callbacks. A needs-input DM // can therefore be followed by the instructed session exit before the // first durable state await completes. The message handler installs this @@ -5748,6 +5974,14 @@ export class FactoryLoop implements Factory { if (tracingReconciledExit) this.#logger.info?.('[factory] reconciled agent exit question replay completed', { issue: record.issue.key, name }) const exiting = record.agents.get(name) + if (isCompletionReason(reason) && exiting?.spec.role === 'babysitter' && exiting.spec.ownedPullRequest) { + await this.#state.completeRoutedPrBabysitter( + this.#workspaceId, + routedPrIdentity(exiting.spec.ownedPullRequest.repo, exiting.spec.ownedPullRequest.number), + name, + this.#clock.now(), + ) + } if (exiting) await this.#reportAgent(record, exiting, 'agent.exited', { releaseReason: reason }) if (tracingReconciledExit) this.#logger.info?.('[factory] reconciled agent exit telemetry completed', { issue: record.issue.key, name }) @@ -10041,6 +10275,11 @@ export class FactoryLoop implements Factory { return } + // The routed sweep performs its own authoritative metadata and opt-out + // checks. Trigger it from PR activity, but keep the issue-created router + // below unchanged. + void this.#sweepRoutedPrBabysitters() + const repo = `${parts.owner}/${parts.repo}` // Once ownership exists, it is authoritative even if the PR title or head // branch is renamed. Branch/title/body matching is spawn-time discovery @@ -10311,6 +10550,30 @@ export class FactoryLoop implements Factory { this.#increment('babysitterLifecycleOwnershipRejected') return } + const prWorkIdentity = routedPrIdentity(prRef.repo, prRef.prNumber) + const prWorkClaim = await this.#state.claimRoutedPrBabysitter( + this.#workspaceId, + prWorkIdentity, + { + repo: prRef.repo, + prNumber: prRef.prNumber, + revision: `issue:${issueKey(record.issue)}`, + source: 'issue-created', + }, + this.#dispatchLifecycleOwner, + this.#clock.now(), + ROUTED_PR_BABYSITTER_CLAIM_LEASE_MS, + Number.MAX_SAFE_INTEGER, + ) + if (prWorkClaim.outcome === 'already-running' && prWorkClaim.claim?.source === 'routed-open-prs') { + this.#increment('babysitterOwnershipConflictsSuppressed') + this.#logger.warn?.('[factory] issue-created babysitter suppressed by existing PR work-unit claim', { + issue: record.issue.key, + repo: prRef.repo, + prNumber: prRef.prNumber, + }) + return + } const replacedSuperseded = prRef.authoritative ? await this.#retireSupersededBabysitters(record, prRef) : false @@ -10363,6 +10626,13 @@ export class FactoryLoop implements Factory { this.#babysitterSpawned.add(babysitterKey) const ref = this.#babysitterPr.get(babysitterKey)! await this.#persistBabysitterSession(record.issue, ref, tracked) + await this.#state.markRoutedPrBabysitterRunning( + this.#workspaceId, + prWorkIdentity, + this.#dispatchLifecycleOwner, + tracked.result?.name ?? trackedName, + this.#clock.now(), + ) await this.#ensureBabysitterResourceSubscription(record.issue, ref, tracked) await this.#retargetSlackConversationToBabysitter(record) return @@ -10466,6 +10736,13 @@ export class FactoryLoop implements Factory { }) const ref = this.#babysitterPr.get(babysitterKey)! await this.#persistBabysitterSession(record.issue, ref, tracked) + await this.#state.markRoutedPrBabysitterRunning( + this.#workspaceId, + prWorkIdentity, + this.#dispatchLifecycleOwner, + tracked?.result?.name ?? spawned.name, + this.#clock.now(), + ) await this.#ensureBabysitterResourceSubscription(record.issue, ref, tracked) await this.#retargetSlackConversationToBabysitter(record) await this.#writeInFlightRegistry() diff --git a/src/ports/state.ts b/src/ports/state.ts index 78dbf98c..6d12df54 100644 --- a/src/ports/state.ts +++ b/src/ports/state.ts @@ -91,6 +91,24 @@ export type BabysitterSessionState = { pendingDeliveryClaims?: Array<{ deliveryId: string; claimToken: string }> } +export type RoutedPrBabysitterClaim = { + repo: string + prNumber: number + /** Stable hash of the mounted PR snapshot which admitted this run. */ + revision: string + source: 'issue-created' | 'routed-open-prs' + status: 'claimed' | 'running' | 'complete' + owner: string + leaseUntilMs: number + claimedAtMs: number + updatedAtMs: number + agentName?: string +} + +export type RoutedPrBabysitterClaimResult = + | { outcome: 'claimed'; claim: RoutedPrBabysitterClaim } + | { outcome: 'owned' | 'already-running' | 'unchanged' | 'capacity'; claim?: RoutedPrBabysitterClaim } + export type ConversationMessage = { id: string text: string @@ -380,6 +398,20 @@ export interface StateStore { listBabysitterSessions(workspaceId: string): Promise> clearBabysitterSession(workspaceId: string, issueKey: string): Promise + claimRoutedPrBabysitter( + workspaceId: string, + identity: string, + seed: Omit, + owner: string, + nowMs: number, + leaseMs: number, + maxActive: number, + ): Promise + markRoutedPrBabysitterRunning(workspaceId: string, identity: string, owner: string, agentName: string, nowMs: number): Promise + completeRoutedPrBabysitter(workspaceId: string, identity: string, agentName: string, nowMs: number): Promise + releaseRoutedPrBabysitterClaim(workspaceId: string, identity: string, owner: string): Promise + listRoutedPrBabysitterClaims(workspaceId: string): Promise> + recordCanonicalState(workspaceId: string, key: string, stateId: string): Promise getCanonicalState(workspaceId: string, key: string): Promise } diff --git a/src/state/file-state-store.ts b/src/state/file-state-store.ts index f7705217..3885c792 100644 --- a/src/state/file-state-store.ts +++ b/src/state/file-state-store.ts @@ -13,6 +13,8 @@ import type { GithubIssueCommentWatchState, ConversationMessage, ConversationSessionState, + RoutedPrBabysitterClaim, + RoutedPrBabysitterClaimResult, WaitingClarification, } from '../ports/state' import { InMemoryStateStore, type InMemoryStateStoreOptions } from './in-memory-state-store' @@ -22,6 +24,7 @@ type PersistedWorkspaceState = { githubIssueCommentWatches: Record waitingClarifications: Record babysitterSessions: Record + routedPrBabysitterClaims: Record conversationSessions: Record dispatchLifecycles: Record } @@ -648,6 +651,107 @@ export class FileStateStore extends InMemoryStateStore { }) } + override async claimRoutedPrBabysitter( + workspaceId: string, + identity: string, + seed: Omit, + owner: string, + nowMs: number, + leaseMs: number, + maxActive: number, + ): Promise { + return await this.#exclusive(async () => this.#withMutationLock(async () => { + const document = await this.#loadFromDisk() + const workspace = document.workspaces[workspaceId] ??= emptyWorkspaceState() + const existing = workspace.routedPrBabysitterClaims[identity] + if (existing?.status !== 'complete' && existing?.owner === owner) { + existing.leaseUntilMs = nowMs + leaseMs + existing.updatedAtMs = nowMs + await this.#persist(document) + return { outcome: 'owned', claim: structuredClone(existing) } + } + if (existing?.status !== 'complete' && existing && existing.leaseUntilMs > nowMs) { + return { outcome: 'already-running', claim: structuredClone(existing) } + } + if (existing?.status === 'complete' && (existing.source === 'issue-created' || existing.revision === seed.revision)) { + return { outcome: 'unchanged', claim: structuredClone(existing) } + } + const active = Object.entries(workspace.routedPrBabysitterClaims).filter(([key, claim]) => + key !== identity && claim.status !== 'complete' && claim.leaseUntilMs > nowMs + ).length + if (active >= Math.max(1, Math.trunc(maxActive))) return { outcome: 'capacity' } + const claim: RoutedPrBabysitterClaim = { + ...seed, + status: 'claimed', + owner, + leaseUntilMs: nowMs + leaseMs, + claimedAtMs: nowMs, + updatedAtMs: nowMs, + } + workspace.routedPrBabysitterClaims[identity] = claim + await this.#persist(document) + return { outcome: 'claimed', claim: structuredClone(claim) } + })) + } + + override async markRoutedPrBabysitterRunning( + workspaceId: string, + identity: string, + owner: string, + agentName: string, + nowMs: number, + ): Promise { + return await this.#exclusive(async () => this.#withMutationLock(async () => { + const document = await this.#loadFromDisk() + const claim = document.workspaces[workspaceId]?.routedPrBabysitterClaims[identity] + if (!claim || claim.owner !== owner || claim.status === 'complete') return false + claim.status = 'running' + claim.agentName = agentName + claim.updatedAtMs = nowMs + await this.#persist(document) + return true + })) + } + + override async completeRoutedPrBabysitter( + workspaceId: string, + identity: string, + agentName: string, + nowMs: number, + ): Promise { + return await this.#exclusive(async () => this.#withMutationLock(async () => { + const document = await this.#loadFromDisk() + const claim = document.workspaces[workspaceId]?.routedPrBabysitterClaims[identity] + if (!claim || claim.agentName !== agentName || claim.status === 'complete') return false + claim.status = 'complete' + claim.leaseUntilMs = nowMs + claim.updatedAtMs = nowMs + await this.#persist(document) + return true + })) + } + + override async releaseRoutedPrBabysitterClaim(workspaceId: string, identity: string, owner: string): Promise { + return await this.#exclusive(async () => this.#withMutationLock(async () => { + const document = await this.#loadFromDisk() + const workspace = document.workspaces[workspaceId] + const claim = workspace?.routedPrBabysitterClaims[identity] + if (!workspace || !claim || claim.owner !== owner || claim.status === 'running') return false + delete workspace.routedPrBabysitterClaims[identity] + if (workspaceIsEmpty(workspace)) delete document.workspaces[workspaceId] + await this.#persist(document) + return true + })) + } + + override async listRoutedPrBabysitterClaims(workspaceId: string): Promise> { + return await this.#exclusive(async () => { + const document = await this.#loadFromDisk() + return Object.entries(document.workspaces[workspaceId]?.routedPrBabysitterClaims ?? {}) + .map(([identity, claim]) => [identity, structuredClone(claim)]) + }) + } + override async reserveConversationSession( workspaceId: string, conversationId: string, @@ -883,12 +987,14 @@ const parseDocument = (value: unknown): WatchStateDocument => { const watches = rawWorkspace.githubIssueCommentWatches const clarifications = rawWorkspace.waitingClarifications const babysitters = rawWorkspace.babysitterSessions + const routedClaims = rawWorkspace.routedPrBabysitterClaims const conversations = rawWorkspace.conversationSessions const lifecycles = rawWorkspace.dispatchLifecycles if ( !isRecord(watches) || !isRecord(clarifications) || (babysitters !== undefined && !isRecord(babysitters)) || + (routedClaims !== undefined && !isRecord(routedClaims)) || (conversations !== undefined && !isRecord(conversations)) || (lifecycles !== undefined && !isRecord(lifecycles)) ) { @@ -898,6 +1004,7 @@ const parseDocument = (value: unknown): WatchStateDocument => { githubIssueCommentWatches: watches as Record, waitingClarifications: clarifications as Record, babysitterSessions: parseBabysitterSessions(babysitters ?? {}), + routedPrBabysitterClaims: parseRoutedPrBabysitterClaims(routedClaims ?? {}), conversationSessions: parseConversationSessions(conversations ?? {}), dispatchLifecycles: (lifecycles ?? {}) as Record, } @@ -918,6 +1025,7 @@ const parseDocument = (value: unknown): WatchStateDocument => { githubIssueCommentWatches: watches as Record, waitingClarifications: clarifications as Record, babysitterSessions: parseBabysitterSessions(babysitters ?? {}), + routedPrBabysitterClaims: {}, conversationSessions: {}, dispatchLifecycles: {}, } @@ -934,6 +1042,7 @@ const parseDocument = (value: unknown): WatchStateDocument => { githubIssueCommentWatches: watches as Record, waitingClarifications: {}, babysitterSessions: {}, + routedPrBabysitterClaims: {}, conversationSessions: {}, dispatchLifecycles: {}, } @@ -952,6 +1061,30 @@ const cloneClarification = (record: WaitingClarification): WaitingClarification const cloneBabysitterSession = (session: BabysitterSessionState): BabysitterSessionState => structuredClone(session) +const parseRoutedPrBabysitterClaims = (value: Record): Record => { + const claims: Record = {} + for (const [identity, candidate] of Object.entries(value)) { + if ( + !isRecord(candidate) || + typeof candidate.repo !== 'string' || + !Number.isSafeInteger(candidate.prNumber) || + (candidate.prNumber as number) < 1 || + typeof candidate.revision !== 'string' || + !['issue-created', 'routed-open-prs'].includes(String(candidate.source)) || + !['claimed', 'running', 'complete'].includes(String(candidate.status)) || + typeof candidate.owner !== 'string' || + typeof candidate.leaseUntilMs !== 'number' || + typeof candidate.claimedAtMs !== 'number' || + typeof candidate.updatedAtMs !== 'number' || + (candidate.agentName !== undefined && typeof candidate.agentName !== 'string') + ) { + throw new Error('Factory GitHub watch state file is invalid') + } + claims[identity] = structuredClone(candidate) as RoutedPrBabysitterClaim + } + return claims +} + const cloneConversationSession = (session: ConversationSessionState): ConversationSessionState => structuredClone(session) @@ -1134,6 +1267,7 @@ const emptyWorkspaceState = (): PersistedWorkspaceState => ({ githubIssueCommentWatches: {}, waitingClarifications: {}, babysitterSessions: {}, + routedPrBabysitterClaims: {}, conversationSessions: {}, dispatchLifecycles: {}, }) @@ -1142,6 +1276,7 @@ const workspaceIsEmpty = (workspace: PersistedWorkspaceState): boolean => Object.keys(workspace.githubIssueCommentWatches).length === 0 && Object.keys(workspace.waitingClarifications).length === 0 && Object.keys(workspace.babysitterSessions).length === 0 && + Object.keys(workspace.routedPrBabysitterClaims).length === 0 && Object.keys(workspace.conversationSessions).length === 0 && Object.keys(workspace.dispatchLifecycles).length === 0 diff --git a/src/state/in-memory-state-store.ts b/src/state/in-memory-state-store.ts index 024cde28..5e580ea2 100644 --- a/src/state/in-memory-state-store.ts +++ b/src/state/in-memory-state-store.ts @@ -9,6 +9,8 @@ import type { DispatchAttemptState, GithubIssueCommentWatchState, RegistryHandoffAgent, + RoutedPrBabysitterClaim, + RoutedPrBabysitterClaimResult, ConversationMessage, ConversationSessionState, StateStore, @@ -31,6 +33,7 @@ type WorkspaceState = { canonicalIssueStates: Map dispatchFailureReaperHandoffs: Map babysitterSessions: Map + routedPrBabysitterClaims: Map dispatchLifecycles: Map } @@ -633,6 +636,85 @@ export class InMemoryStateStore implements StateStore { this.#workspace(workspaceId).babysitterSessions.delete(issueKey) } + async claimRoutedPrBabysitter( + workspaceId: string, + identity: string, + seed: Omit, + owner: string, + nowMs: number, + leaseMs: number, + maxActive: number, + ): Promise { + const claims = this.#workspace(workspaceId).routedPrBabysitterClaims + const existing = claims.get(identity) + if (existing?.status !== 'complete' && existing?.owner === owner) { + existing.leaseUntilMs = nowMs + leaseMs + existing.updatedAtMs = nowMs + return { outcome: 'owned', claim: structuredClone(existing) } + } + if (existing?.status !== 'complete' && existing && existing.leaseUntilMs > nowMs) { + return { outcome: 'already-running', claim: structuredClone(existing) } + } + if (existing?.status === 'complete' && (existing.source === 'issue-created' || existing.revision === seed.revision)) { + return { outcome: 'unchanged', claim: structuredClone(existing) } + } + const active = [...claims.entries()].filter(([key, claim]) => + key !== identity && claim.status !== 'complete' && claim.leaseUntilMs > nowMs + ).length + if (active >= Math.max(1, Math.trunc(maxActive))) return { outcome: 'capacity' } + const claim: RoutedPrBabysitterClaim = { + ...seed, + status: 'claimed', + owner, + leaseUntilMs: nowMs + leaseMs, + claimedAtMs: nowMs, + updatedAtMs: nowMs, + } + claims.set(identity, claim) + return { outcome: 'claimed', claim: structuredClone(claim) } + } + + async markRoutedPrBabysitterRunning( + workspaceId: string, + identity: string, + owner: string, + agentName: string, + nowMs: number, + ): Promise { + const claim = this.#workspace(workspaceId).routedPrBabysitterClaims.get(identity) + if (!claim || claim.owner !== owner || claim.status === 'complete') return false + claim.status = 'running' + claim.agentName = agentName + claim.updatedAtMs = nowMs + return true + } + + async completeRoutedPrBabysitter( + workspaceId: string, + identity: string, + agentName: string, + nowMs: number, + ): Promise { + const claim = this.#workspace(workspaceId).routedPrBabysitterClaims.get(identity) + if (!claim || claim.agentName !== agentName || claim.status === 'complete') return false + claim.status = 'complete' + claim.leaseUntilMs = nowMs + claim.updatedAtMs = nowMs + return true + } + + async releaseRoutedPrBabysitterClaim(workspaceId: string, identity: string, owner: string): Promise { + const claims = this.#workspace(workspaceId).routedPrBabysitterClaims + const claim = claims.get(identity) + if (!claim || claim.owner !== owner || claim.status === 'running') return false + return claims.delete(identity) + } + + async listRoutedPrBabysitterClaims(workspaceId: string): Promise> { + return [...this.#workspace(workspaceId).routedPrBabysitterClaims] + .map(([identity, claim]) => [identity, structuredClone(claim)]) + } + async recordCanonicalState(workspaceId: string, key: string, stateId: string): Promise { this.#workspace(workspaceId).canonicalIssueStates.set(key, stateId) } @@ -658,6 +740,7 @@ export class InMemoryStateStore implements StateStore { canonicalIssueStates: new Map(), dispatchFailureReaperHandoffs: new Map(), babysitterSessions: new Map(), + routedPrBabysitterClaims: new Map(), dispatchLifecycles: new Map(), } this.#workspaces.set(workspaceId, state) diff --git a/src/state/routed-pr-babysitter-claim.test.ts b/src/state/routed-pr-babysitter-claim.test.ts new file mode 100644 index 00000000..448d34e5 --- /dev/null +++ b/src/state/routed-pr-babysitter-claim.test.ts @@ -0,0 +1,70 @@ +import { mkdtemp } from 'node:fs/promises' +import { tmpdir } from 'node:os' +import { join } from 'node:path' + +import { describe, expect, it } from 'vitest' + +import type { StateStore } from '../ports/state' +import { FileStateStore } from './file-state-store' +import { InMemoryStateStore } from './in-memory-state-store' + +const seed = { + repo: 'AgentWorkforce/pear', + prNumber: 7, + revision: 'rev-1', + source: 'routed-open-prs' as const, +} + +const verifyClaims = (label: string, create: () => Promise | StateStore) => { + it(`${label} atomically gives one owner the PR work unit`, async () => { + const store = await create() + const results = await Promise.all([ + store.claimRoutedPrBabysitter('workspace', 'agentworkforce/pear#7', seed, 'owner-a', 1_000, 60_000, 5), + store.claimRoutedPrBabysitter('workspace', 'agentworkforce/pear#7', seed, 'owner-b', 1_000, 60_000, 5), + ]) + expect(results.map((result) => result.outcome).sort()).toEqual(['already-running', 'claimed']) + }) + + it(`${label} enforces global capacity and re-admits only changed completed work`, async () => { + const store = await create() + const first = await store.claimRoutedPrBabysitter('workspace', 'agentworkforce/pear#7', seed, 'owner-a', 1_000, 60_000, 1) + expect(first.outcome).toBe('claimed') + expect(await store.markRoutedPrBabysitterRunning('workspace', 'agentworkforce/pear#7', 'owner-a', 'agent-a', 1_001)).toBe(true) + expect((await store.claimRoutedPrBabysitter( + 'workspace', + 'agentworkforce/pear#8', + { ...seed, prNumber: 8 }, + 'owner-a', + 1_002, + 60_000, + 1, + )).outcome).toBe('capacity') + expect(await store.completeRoutedPrBabysitter('workspace', 'agentworkforce/pear#7', 'agent-a', 1_003)).toBe(true) + expect((await store.claimRoutedPrBabysitter( + 'workspace', + 'agentworkforce/pear#7', + seed, + 'owner-b', + 1_004, + 60_000, + 1, + )).outcome).toBe('unchanged') + expect((await store.claimRoutedPrBabysitter( + 'workspace', + 'agentworkforce/pear#7', + { ...seed, revision: 'rev-2' }, + 'owner-b', + 1_005, + 60_000, + 1, + )).outcome).toBe('claimed') + }) +} + +describe('routed PR babysitter claims', () => { + verifyClaims('memory state', () => new InMemoryStateStore({ batchSize: 5 })) + verifyClaims('file state', async () => { + const root = await mkdtemp(join(tmpdir(), 'factory-routed-pr-claim-')) + return new FileStateStore({ batchSize: 5, watchStatePath: join(root, 'state.json') }) + }) +}) From 80557485bcfad6144eff9eb3dbba4ff287f81b3d Mon Sep 17 00:00:00 2001 From: kjgbot Date: Sat, 8 Aug 2026 22:29:24 +0200 Subject: [PATCH 2/4] fix: address routed babysitter review findings --- src/config/schema.test.ts | 12 ++ src/config/schema.ts | 21 ++- src/github/routed-pr-babysitter.test.ts | 39 ++++++ src/github/routed-pr-babysitter.ts | 25 ++-- src/orchestrator/factory.test.ts | 80 +++++++++++ src/orchestrator/factory.ts | 80 +++++++---- src/ports/state.ts | 10 ++ src/state/file-state-store.ts | 37 ++++- src/state/in-memory-state-store.ts | 33 ++++- src/state/routed-pr-babysitter-claim.test.ts | 135 ++++++++++++++++++- 10 files changed, 434 insertions(+), 38 deletions(-) diff --git a/src/config/schema.test.ts b/src/config/schema.test.ts index 0cb2a748..3fb180fe 100644 --- a/src/config/schema.test.ts +++ b/src/config/schema.test.ts @@ -320,6 +320,18 @@ describe('FactoryConfigSchema', () => { expect(parsed.repos.names).toEqual(['pear', 'cloud', 'agentswarm']) }) + it('requires explicit routed repositories for routed PR babysitting', () => { + expect(() => FactoryConfigSchema.parse({ + babysitter: { enabled: true, mode: 'routed-open-prs' }, + repos: {}, + })).toThrow(/repos\.names must contain at least one repository/u) + + expect(FactoryConfigSchema.parse({ + babysitter: { enabled: true, mode: 'factory-created' }, + repos: {}, + }).babysitter.mode).toBe('factory-created') + }) + it('lets explicit byLabel/clonePaths/labels override the derived ones', () => { const parsed = FactoryConfigSchema.parse({ subscription: { labels: ['pear'] }, diff --git a/src/config/schema.ts b/src/config/schema.ts index 7af368d6..30a99314 100644 --- a/src/config/schema.ts +++ b/src/config/schema.ts @@ -323,9 +323,26 @@ const NodeConfigObjectSchema = z.object({ const FactoryConfigObjectSchema = WorkspaceConfigObjectSchema.merge(NodeConfigObjectSchema) -export const WorkspaceConfigSchema = WorkspaceConfigObjectSchema.transform((cfg) => normalizeWorkspaceConfig(cfg)) +const requireRoutedBabysitterRepos = ( + cfg: z.infer, + ctx: z.RefinementCtx, +): void => { + if (cfg.babysitter.mode === 'routed-open-prs' && (cfg.repos.names?.length ?? 0) === 0) { + ctx.addIssue({ + code: z.ZodIssueCode.custom, + path: ['repos', 'names'], + message: 'repos.names must contain at least one repository when babysitter.mode is routed-open-prs', + }) + } +} + +export const WorkspaceConfigSchema = WorkspaceConfigObjectSchema + .superRefine(requireRoutedBabysitterRepos) + .transform((cfg) => normalizeWorkspaceConfig(cfg)) export const NodeConfigSchema = NodeConfigObjectSchema.transform((cfg) => normalizeNodeConfig(cfg)) -export const FactoryConfigSchema = FactoryConfigObjectSchema.transform((cfg) => normalizeFactoryConfig(cfg)) +export const FactoryConfigSchema = FactoryConfigObjectSchema + .superRefine(requireRoutedBabysitterRepos) + .transform((cfg) => normalizeFactoryConfig(cfg)) function normalizeWorkspaceConfig(cfg: z.infer) { const resolved = resolveRepos(cfg.repos, cfg.repos.cloneRoot) diff --git a/src/github/routed-pr-babysitter.test.ts b/src/github/routed-pr-babysitter.test.ts index 84194e46..c31147be 100644 --- a/src/github/routed-pr-babysitter.test.ts +++ b/src/github/routed-pr-babysitter.test.ts @@ -54,6 +54,45 @@ describe('routed PR babysitter discovery', () => { ]) }) + it('continues through an unreadable alias root and an unreadable first PR alias', async () => { + const nested = '/github/repos/AgentWorkforce/pear/pulls/7/meta.json' + const byID = '/github/repos/AgentWorkforce__pear/pulls/by-id/7.json' + const client = { + listTree: async (prefix: string) => { + if (prefix.includes('/AgentWorkforce/pear/')) throw new Error('nested alias unavailable') + return [nested, byID] + }, + readFile: async (path: string) => { + if (path === nested) throw new Error('stale alias unreadable') + return { content: pull() } + }, + } as unknown as MountClient + + const report = await discoverRoutedPullRequests(client, config()) + + expect(report.candidates.map(({ number }) => number)).toEqual([7]) + expect(report.failures.map(({ reason }) => reason)).toEqual([ + expect.stringContaining('nested alias unavailable'), + 'stale alias unreadable', + ]) + }) + + it('reads REST and established changed-file aliases', async () => { + const rest = await discoverRoutedPullRequests(mount({ + '/github/repos/AgentWorkforce__pear/pulls/by-id/7.json': pull({ + files: [{ filename: 'src/rest.ts' }], + }), + }), config()) + expect(rest.candidates[0]?.filesChanged).toEqual(['src/rest.ts']) + + const established = await discoverRoutedPullRequests(mount({ + '/github/repos/AgentWorkforce__pear/pulls/by-id/7.json': pull({ + files_changed: [{ path: 'src/established.ts' }], + }), + }), config()) + expect(established.candidates[0]?.filesChanged).toEqual(['src/established.ts']) + }) + it('honours label and config opt-outs before admission', async () => { const files = { '/github/repos/AgentWorkforce__pear/pulls/by-id/7.json': pull({ diff --git a/src/github/routed-pr-babysitter.ts b/src/github/routed-pr-babysitter.ts index 3e9dd11a..e1d13297 100644 --- a/src/github/routed-pr-babysitter.ts +++ b/src/github/routed-pr-babysitter.ts @@ -79,14 +79,19 @@ export async function discoverRoutedPullRequests( `/github/repos/${owner}__${name}/pulls/by-id/`, ] const paths = new Set() - try { - for (const root of roots) { + let readableRoot = false + for (const root of roots) { + try { for (const path of await mount.listTree(root)) paths.add(path) + readableRoot = true + } catch (error) { + report.failures.push({ + repo, + reason: `${root}: ${error instanceof Error ? error.message : String(error)}`, + }) } - } catch (error) { - report.failures.push({ repo, reason: error instanceof Error ? error.message : String(error) }) - continue } + if (!readableRoot) continue for (const path of [...paths].sort()) { const pathNumber = pullNumberFromPath(path, owner!, name!) if (!pathNumber) continue @@ -96,7 +101,6 @@ export async function discoverRoutedPullRequests( report.duplicates += 1 continue } - seen.add(identity) let candidate: RoutedPrCandidate | undefined try { candidate = parseRoutedPrCandidate((await mount.readFile(path)).content, repo, pathNumber, path) @@ -109,6 +113,7 @@ export async function discoverRoutedPullRequests( report.failures.push({ repo, prNumber: pathNumber, reason: 'incomplete authoritative PR metadata' }) continue } + seen.add(identity) if (candidate.merged || candidate.state !== 'OPEN' || candidate.draft) { report.terminal += 1 continue @@ -196,7 +201,7 @@ const parseRoutedPrCandidate = ( crossRepository, maintainerCanModify: booleanValue(payload.maintainerCanModify) ?? booleanValue(payload.maintainer_can_modify), labels, - filesChanged: changedFiles(payload.filesChanged ?? payload.files), + filesChanged: changedFiles(payload.filesChanged ?? payload.files_changed ?? payload.files), path, } return { @@ -222,7 +227,11 @@ const labelNames = (value: unknown): string[] => Array.isArray(value) const changedFiles = (value: unknown): string[] | undefined => { if (!Array.isArray(value)) return undefined - const paths = value.map((entry) => typeof entry === 'string' ? entry : stringValue(asRecord(entry)?.path)) + const paths = value.map((entry) => { + if (typeof entry === 'string') return entry + const record = asRecord(entry) + return stringValue(record?.path) ?? stringValue(record?.filename) + }) .filter((entry): entry is string => Boolean(entry)) return paths.length > 0 ? paths : undefined } diff --git a/src/orchestrator/factory.test.ts b/src/orchestrator/factory.test.ts index a9ace169..dc0b7918 100644 --- a/src/orchestrator/factory.test.ts +++ b/src/orchestrator/factory.test.ts @@ -16322,6 +16322,86 @@ describe('FactoryLoop PR babysitter', () => { expect(fleet.spawns).toEqual([]) }) + it('retries abnormal routed exits and completes only successful exits', async () => { + const mount = new FakeMountClient({ + '/github/repos/AgentWorkforce__pear/pulls/by-id/703.json': { + number: 703, + title: 'Retry an interrupted routed babysitter', + body: 'Keep the work unit retryable after a crash.', + state: 'OPEN', + isDraft: false, + merged: false, + headRefName: 'human/retry-babysitter', + headRefOid: 'head-703', + baseRefName: 'main', + headRepository: { nameWithOwner: 'AgentWorkforce/pear' }, + isCrossRepository: false, + labels: [], + }, + }) + const fleet = new FakeFleetClient() + const stateStore = new InMemoryStateStore({ batchSize: 2 }) + const factory = createFactory(config({ + issueSource: 'github', + repos: { + org: 'AgentWorkforce', + names: ['pear'], + clonePaths: { 'AgentWorkforce/pear': '/work/pear' }, + }, + babysitter: { + enabled: true, + mode: 'routed-open-prs', + excludeLabels: ['factory:skip-babysitter'], + excludePullRequests: [], + notifyHumans: false, + }, + }), { mount, fleet, stateStore, triage: new StaticTriage() }) + + await factory.runOnce() + const agentName = fleet.spawns[0]!.name + fleet.emitAgentExit(agentName, 'crash') + await vi.waitFor(() => expect(fleet.spawns).toHaveLength(2)) + + fleet.emitAgentExit(agentName, 'completed') + await vi.waitFor(async () => { + expect((await stateStore.listRoutedPrBabysitterClaims('factory-test'))[0]?.[1].status) + .toBe('complete') + }) + await factory.runOnce() + expect(fleet.spawns).toHaveLength(2) + }) + + it('contains routed sweep state failures at the optional feature boundary', async () => { + class FailingRoutedClaimsStore extends InMemoryStateStore { + override async listRoutedPrBabysitterClaims(): Promise { + throw new Error('routed claim store unavailable') + } + } + const factory = createFactory(config({ + issueSource: 'github', + repos: { + org: 'AgentWorkforce', + names: ['pear'], + clonePaths: { 'AgentWorkforce/pear': '/work/pear' }, + }, + babysitter: { + enabled: true, + mode: 'routed-open-prs', + excludeLabels: [], + excludePullRequests: [], + notifyHumans: false, + }, + }), { + mount: new FakeMountClient(), + fleet: new FakeFleetClient(), + stateStore: new FailingRoutedClaimsStore({ batchSize: 2 }), + triage: new StaticTriage(), + }) + + await expect(factory.runOnce()).resolves.toBeDefined() + expect(factory.status().counters.routedPrBabysitterSweepErrors).toBe(1) + }) + const babysitterConfig = (overrides: FactoryConfigOverrides = {}): FactoryConfig => config({ babysitter: { enabled: true }, terminalState: 'human-review', diff --git a/src/orchestrator/factory.ts b/src/orchestrator/factory.ts index 8bfbd346..904f32c2 100644 --- a/src/orchestrator/factory.ts +++ b/src/orchestrator/factory.ts @@ -1689,28 +1689,51 @@ export class FactoryLoop implements Factory { async #restoreRoutedPrBabysitterClaims(): Promise { if (this.#config.babysitter.mode !== 'routed-open-prs') return - const roster = await this.#fleet.roster() - const online = new Set(roster.agents.map((agent) => agent.name)) - for (const [identity, claim] of await this.#state.listRoutedPrBabysitterClaims(this.#workspaceId)) { - if (claim.status === 'running' && claim.agentName && online.has(claim.agentName)) { - this.#routedPrBabysitterAgents.set(claim.agentName, identity) - this.#increment('routedPrBabysitterClaimsRestored') - } else if (claim.status !== 'complete' && claim.leaseUntilMs > this.#clock.now()) { - this.#logger.warn?.('[factory] routed PR babysitter claim awaits lease expiry after missing agent', { - repo: claim.repo, - prNumber: claim.prNumber, - leaseUntilMs: claim.leaseUntilMs, - }) + try { + const roster = await this.#fleet.roster() + const online = new Set(roster.agents.map((agent) => agent.name)) + for (const [identity, claim] of await this.#state.listRoutedPrBabysitterClaims(this.#workspaceId)) { + if (claim.status === 'running' && claim.agentName && online.has(claim.agentName)) { + const adopted = await this.#state.adoptRoutedPrBabysitterClaim( + this.#workspaceId, + identity, + claim.agentName, + this.#routedPrBabysitterOwner, + this.#clock.now(), + ROUTED_PR_BABYSITTER_CLAIM_LEASE_MS, + ) + if (adopted) { + this.#routedPrBabysitterAgents.set(claim.agentName, identity) + this.#increment('routedPrBabysitterClaimsRestored') + } + } else if (claim.status !== 'complete' && claim.leaseUntilMs > this.#clock.now()) { + this.#logger.warn?.('[factory] routed PR babysitter claim awaits lease expiry after missing agent', { + repo: claim.repo, + prNumber: claim.prNumber, + leaseUntilMs: claim.leaseUntilMs, + }) + } } + } catch (error) { + this.#logger.warn?.('[factory] failed to restore routed PR babysitter claims', { + error: describeError(error).errorMessage, + }) } } async #sweepRoutedPrBabysitters(dryRun = this.#config.dryRun): Promise { if (!this.#config.babysitter.enabled || this.#config.babysitter.mode !== 'routed-open-prs' || this.#stopping) return if (this.#routedPrBabysitterSweep) return this.#routedPrBabysitterSweep - const sweep = this.#runRoutedPrBabysitterSweep(dryRun).finally(() => { - if (this.#routedPrBabysitterSweep === sweep) this.#routedPrBabysitterSweep = undefined - }) + const sweep = this.#runRoutedPrBabysitterSweep(dryRun) + .catch((error) => { + this.#increment('routedPrBabysitterSweepErrors') + this.#logger.warn?.('[factory] routed PR babysitter sweep failed', { + error: describeError(error).errorMessage, + }) + }) + .finally(() => { + if (this.#routedPrBabysitterSweep === sweep) this.#routedPrBabysitterSweep = undefined + }) this.#routedPrBabysitterSweep = sweep return sweep } @@ -1754,6 +1777,7 @@ export class FactoryLoop implements Factory { admitted: 0, capacityDeferred: 0, unchanged: 0, + spawnFailures: 0, } for (const failure of discovery.failures) { this.#logger.warn?.('[factory] routed PR babysitter discovery dropped candidate', failure) @@ -1803,6 +1827,7 @@ export class FactoryLoop implements Factory { this.#routedPrBabysitterOwner, ) this.#increment('routedPrBabysitterSpawnFailures') + stats.spawnFailures += 1 this.#logger.warn?.('[factory] routed PR babysitter spawn failed', { repo: candidate.repo, prNumber: candidate.number, @@ -1812,7 +1837,7 @@ export class FactoryLoop implements Factory { // Pacing is intentionally one new admission per sweep. Every remaining // eligible candidate is explicitly accounted as deferred, never dropped. stats.capacityDeferred += discovery.candidates.length - - stats.alreadyOwned - stats.unchanged - stats.admitted - stats.capacityDeferred + stats.alreadyOwned - stats.unchanged - stats.admitted - stats.capacityDeferred - stats.spawnFailures break } this.#logger.info?.('[factory] routed PR babysitter sweep completed', stats) @@ -5908,13 +5933,22 @@ export class FactoryLoop implements Factory { const routedIdentity = this.#routedPrBabysitterAgents.get(name) if (routedIdentity) { this.#routedPrBabysitterAgents.delete(name) - await this.#state.completeRoutedPrBabysitter( - this.#workspaceId, - routedIdentity, - name, - this.#clock.now(), - ) - this.#increment('routedPrBabysittersCompleted') + if (isCompletionReason(reason)) { + await this.#state.completeRoutedPrBabysitter( + this.#workspaceId, + routedIdentity, + name, + this.#clock.now(), + ) + this.#increment('routedPrBabysittersCompleted') + } else { + await this.#state.releaseRoutedPrBabysitterClaim( + this.#workspaceId, + routedIdentity, + this.#routedPrBabysitterOwner, + ) + this.#increment('routedPrBabysitterAbnormalExits') + } await this.#sweepRoutedPrBabysitters() return } diff --git a/src/ports/state.ts b/src/ports/state.ts index 6d12df54..e2198d53 100644 --- a/src/ports/state.ts +++ b/src/ports/state.ts @@ -105,6 +105,8 @@ export type RoutedPrBabysitterClaim = { agentName?: string } +export const ROUTED_PR_BABYSITTER_COMPLETED_RETENTION_MS = 30 * 24 * 60 * 60_000 + export type RoutedPrBabysitterClaimResult = | { outcome: 'claimed'; claim: RoutedPrBabysitterClaim } | { outcome: 'owned' | 'already-running' | 'unchanged' | 'capacity'; claim?: RoutedPrBabysitterClaim } @@ -408,6 +410,14 @@ export interface StateStore { maxActive: number, ): Promise markRoutedPrBabysitterRunning(workspaceId: string, identity: string, owner: string, agentName: string, nowMs: number): Promise + adoptRoutedPrBabysitterClaim( + workspaceId: string, + identity: string, + agentName: string, + owner: string, + nowMs: number, + leaseMs: number, + ): Promise completeRoutedPrBabysitter(workspaceId: string, identity: string, agentName: string, nowMs: number): Promise releaseRoutedPrBabysitterClaim(workspaceId: string, identity: string, owner: string): Promise listRoutedPrBabysitterClaims(workspaceId: string): Promise> diff --git a/src/state/file-state-store.ts b/src/state/file-state-store.ts index 3885c792..96629b55 100644 --- a/src/state/file-state-store.ts +++ b/src/state/file-state-store.ts @@ -5,6 +5,7 @@ import { dirname, join } from 'node:path' import lockfile from 'proper-lockfile' import { githubRepositoriesMatch } from '../github/repo-identity' +import { ROUTED_PR_BABYSITTER_COMPLETED_RETENTION_MS } from '../ports/state' import type { BabysitterSessionState, ClarificationReply, @@ -663,8 +664,20 @@ export class FileStateStore extends InMemoryStateStore { return await this.#exclusive(async () => this.#withMutationLock(async () => { const document = await this.#loadFromDisk() const workspace = document.workspaces[workspaceId] ??= emptyWorkspaceState() + for (const [key, claim] of Object.entries(workspace.routedPrBabysitterClaims)) { + if ( + claim.status === 'complete' && + claim.updatedAtMs <= nowMs - ROUTED_PR_BABYSITTER_COMPLETED_RETENTION_MS + ) delete workspace.routedPrBabysitterClaims[key] + } const existing = workspace.routedPrBabysitterClaims[identity] - if (existing?.status !== 'complete' && existing?.owner === owner) { + if ( + existing?.status !== 'complete' && + existing?.owner === owner && + existing.leaseUntilMs > nowMs && + (existing.source === 'routed-open-prs' || + (existing.source === seed.source && existing.revision === seed.revision)) + ) { existing.leaseUntilMs = nowMs + leaseMs existing.updatedAtMs = nowMs await this.#persist(document) @@ -713,6 +726,26 @@ export class FileStateStore extends InMemoryStateStore { })) } + override async adoptRoutedPrBabysitterClaim( + workspaceId: string, + identity: string, + agentName: string, + owner: string, + nowMs: number, + leaseMs: number, + ): Promise { + return await this.#exclusive(async () => this.#withMutationLock(async () => { + const document = await this.#loadFromDisk() + const claim = document.workspaces[workspaceId]?.routedPrBabysitterClaims[identity] + if (!claim || claim.status !== 'running' || claim.agentName !== agentName) return false + claim.owner = owner + claim.leaseUntilMs = nowMs + leaseMs + claim.updatedAtMs = nowMs + await this.#persist(document) + return true + })) + } + override async completeRoutedPrBabysitter( workspaceId: string, identity: string, @@ -736,7 +769,7 @@ export class FileStateStore extends InMemoryStateStore { const document = await this.#loadFromDisk() const workspace = document.workspaces[workspaceId] const claim = workspace?.routedPrBabysitterClaims[identity] - if (!workspace || !claim || claim.owner !== owner || claim.status === 'running') return false + if (!workspace || !claim || claim.owner !== owner) return false delete workspace.routedPrBabysitterClaims[identity] if (workspaceIsEmpty(workspace)) delete document.workspaces[workspaceId] await this.#persist(document) diff --git a/src/state/in-memory-state-store.ts b/src/state/in-memory-state-store.ts index 5e580ea2..9d4c5990 100644 --- a/src/state/in-memory-state-store.ts +++ b/src/state/in-memory-state-store.ts @@ -1,5 +1,6 @@ import { BatchTracker } from '../orchestrator/batch-tracker' import { githubRepositoriesMatch } from '../github/repo-identity' +import { ROUTED_PR_BABYSITTER_COMPLETED_RETENTION_MS } from '../ports/state' import type { BatchSnapshot, BabysitterSessionState, @@ -646,8 +647,20 @@ export class InMemoryStateStore implements StateStore { maxActive: number, ): Promise { const claims = this.#workspace(workspaceId).routedPrBabysitterClaims + for (const [key, claim] of claims) { + if ( + claim.status === 'complete' && + claim.updatedAtMs <= nowMs - ROUTED_PR_BABYSITTER_COMPLETED_RETENTION_MS + ) claims.delete(key) + } const existing = claims.get(identity) - if (existing?.status !== 'complete' && existing?.owner === owner) { + if ( + existing?.status !== 'complete' && + existing?.owner === owner && + existing.leaseUntilMs > nowMs && + (existing.source === 'routed-open-prs' || + (existing.source === seed.source && existing.revision === seed.revision)) + ) { existing.leaseUntilMs = nowMs + leaseMs existing.updatedAtMs = nowMs return { outcome: 'owned', claim: structuredClone(existing) } @@ -689,6 +702,22 @@ export class InMemoryStateStore implements StateStore { return true } + async adoptRoutedPrBabysitterClaim( + workspaceId: string, + identity: string, + agentName: string, + owner: string, + nowMs: number, + leaseMs: number, + ): Promise { + const claim = this.#workspace(workspaceId).routedPrBabysitterClaims.get(identity) + if (!claim || claim.status !== 'running' || claim.agentName !== agentName) return false + claim.owner = owner + claim.leaseUntilMs = nowMs + leaseMs + claim.updatedAtMs = nowMs + return true + } + async completeRoutedPrBabysitter( workspaceId: string, identity: string, @@ -706,7 +735,7 @@ export class InMemoryStateStore implements StateStore { async releaseRoutedPrBabysitterClaim(workspaceId: string, identity: string, owner: string): Promise { const claims = this.#workspace(workspaceId).routedPrBabysitterClaims const claim = claims.get(identity) - if (!claim || claim.owner !== owner || claim.status === 'running') return false + if (!claim || claim.owner !== owner) return false return claims.delete(identity) } diff --git a/src/state/routed-pr-babysitter-claim.test.ts b/src/state/routed-pr-babysitter-claim.test.ts index 448d34e5..2496c8e9 100644 --- a/src/state/routed-pr-babysitter-claim.test.ts +++ b/src/state/routed-pr-babysitter-claim.test.ts @@ -4,7 +4,10 @@ import { join } from 'node:path' import { describe, expect, it } from 'vitest' -import type { StateStore } from '../ports/state' +import { + ROUTED_PR_BABYSITTER_COMPLETED_RETENTION_MS, + type StateStore, +} from '../ports/state' import { FileStateStore } from './file-state-store' import { InMemoryStateStore } from './in-memory-state-store' @@ -59,6 +62,136 @@ const verifyClaims = (label: string, create: () => Promise | StateSt 1, )).outcome).toBe('claimed') }) + + it(`${label} treats an expired same-owner claim as a fresh admission`, async () => { + const store = await create() + await store.claimRoutedPrBabysitter( + 'workspace', + 'agentworkforce/pear#7', + seed, + 'owner-a', + 1_000, + 1_000, + 1, + ) + await store.claimRoutedPrBabysitter( + 'workspace', + 'agentworkforce/pear#8', + { ...seed, prNumber: 8 }, + 'owner-b', + 2_001, + 60_000, + 1, + ) + expect((await store.claimRoutedPrBabysitter( + 'workspace', + 'agentworkforce/pear#7', + seed, + 'owner-a', + 2_002, + 60_000, + 1, + )).outcome).toBe('capacity') + }) + + it(`${label} does not merge distinct issue-created work units owned by one process`, async () => { + const store = await create() + const issueSeed = { ...seed, source: 'issue-created' as const } + expect((await store.claimRoutedPrBabysitter( + 'workspace', + 'agentworkforce/pear#7', + issueSeed, + 'dispatch-owner', + 1_000, + 60_000, + 5, + )).outcome).toBe('claimed') + expect((await store.claimRoutedPrBabysitter( + 'workspace', + 'agentworkforce/pear#7', + { ...issueSeed, revision: 'different-issue-lifecycle' }, + 'dispatch-owner', + 1_001, + 60_000, + 5, + )).outcome).toBe('already-running') + }) + + it(`${label} atomically adopts a restored running claim`, async () => { + const store = await create() + await store.claimRoutedPrBabysitter( + 'workspace', + 'agentworkforce/pear#7', + seed, + 'old-owner', + 1_000, + 60_000, + 5, + ) + await store.markRoutedPrBabysitterRunning( + 'workspace', + 'agentworkforce/pear#7', + 'old-owner', + 'agent-a', + 1_001, + ) + expect(await store.adoptRoutedPrBabysitterClaim( + 'workspace', + 'agentworkforce/pear#7', + 'agent-a', + 'new-owner', + 2_000, + 60_000, + )).toBe(true) + expect((await store.listRoutedPrBabysitterClaims('workspace'))[0]?.[1]).toMatchObject({ + owner: 'new-owner', + leaseUntilMs: 62_000, + agentName: 'agent-a', + status: 'running', + }) + expect(await store.releaseRoutedPrBabysitterClaim( + 'workspace', + 'agentworkforce/pear#7', + 'old-owner', + )).toBe(false) + }) + + it(`${label} prunes completed claims after bounded retention`, async () => { + const store = await create() + await store.claimRoutedPrBabysitter( + 'workspace', + 'agentworkforce/pear#7', + seed, + 'owner-a', + 1_000, + 60_000, + 5, + ) + await store.markRoutedPrBabysitterRunning( + 'workspace', + 'agentworkforce/pear#7', + 'owner-a', + 'agent-a', + 1_001, + ) + await store.completeRoutedPrBabysitter( + 'workspace', + 'agentworkforce/pear#7', + 'agent-a', + 1_002, + ) + await store.claimRoutedPrBabysitter( + 'workspace', + 'agentworkforce/pear#8', + { ...seed, prNumber: 8 }, + 'owner-b', + 1_002 + ROUTED_PR_BABYSITTER_COMPLETED_RETENTION_MS + 1, + 60_000, + 5, + ) + expect((await store.listRoutedPrBabysitterClaims('workspace')) + .map(([identity]) => identity)).toEqual(['agentworkforce/pear#8']) + }) } describe('routed PR babysitter claims', () => { From 6ad300ef6824b6625de7d8e47836a181576d8153 Mon Sep 17 00:00:00 2001 From: kjgbot Date: Sat, 8 Aug 2026 22:36:15 +0200 Subject: [PATCH 3/4] fix: guard routed claim adoption by lease --- src/state/file-state-store.ts | 7 +- src/state/in-memory-state-store.ts | 7 +- src/state/routed-pr-babysitter-claim.test.ts | 72 +++++++++++++++++++- 3 files changed, 81 insertions(+), 5 deletions(-) diff --git a/src/state/file-state-store.ts b/src/state/file-state-store.ts index 96629b55..3c002e66 100644 --- a/src/state/file-state-store.ts +++ b/src/state/file-state-store.ts @@ -737,7 +737,12 @@ export class FileStateStore extends InMemoryStateStore { return await this.#exclusive(async () => this.#withMutationLock(async () => { const document = await this.#loadFromDisk() const claim = document.workspaces[workspaceId]?.routedPrBabysitterClaims[identity] - if (!claim || claim.status !== 'running' || claim.agentName !== agentName) return false + if ( + !claim || + claim.status !== 'running' || + claim.agentName !== agentName || + (claim.owner !== owner && claim.leaseUntilMs > nowMs) + ) return false claim.owner = owner claim.leaseUntilMs = nowMs + leaseMs claim.updatedAtMs = nowMs diff --git a/src/state/in-memory-state-store.ts b/src/state/in-memory-state-store.ts index 9d4c5990..f2160005 100644 --- a/src/state/in-memory-state-store.ts +++ b/src/state/in-memory-state-store.ts @@ -711,7 +711,12 @@ export class InMemoryStateStore implements StateStore { leaseMs: number, ): Promise { const claim = this.#workspace(workspaceId).routedPrBabysitterClaims.get(identity) - if (!claim || claim.status !== 'running' || claim.agentName !== agentName) return false + if ( + !claim || + claim.status !== 'running' || + claim.agentName !== agentName || + (claim.owner !== owner && claim.leaseUntilMs > nowMs) + ) return false claim.owner = owner claim.leaseUntilMs = nowMs + leaseMs claim.updatedAtMs = nowMs diff --git a/src/state/routed-pr-babysitter-claim.test.ts b/src/state/routed-pr-babysitter-claim.test.ts index 2496c8e9..0481b640 100644 --- a/src/state/routed-pr-babysitter-claim.test.ts +++ b/src/state/routed-pr-babysitter-claim.test.ts @@ -117,7 +117,7 @@ const verifyClaims = (label: string, create: () => Promise | StateSt )).outcome).toBe('already-running') }) - it(`${label} atomically adopts a restored running claim`, async () => { + it(`${label} atomically adopts an expired restored running claim`, async () => { const store = await create() await store.claimRoutedPrBabysitter( 'workspace', @@ -140,12 +140,12 @@ const verifyClaims = (label: string, create: () => Promise | StateSt 'agentworkforce/pear#7', 'agent-a', 'new-owner', - 2_000, + 61_001, 60_000, )).toBe(true) expect((await store.listRoutedPrBabysitterClaims('workspace'))[0]?.[1]).toMatchObject({ owner: 'new-owner', - leaseUntilMs: 62_000, + leaseUntilMs: 121_001, agentName: 'agent-a', status: 'running', }) @@ -156,6 +156,72 @@ const verifyClaims = (label: string, create: () => Promise | StateSt )).toBe(false) }) + it(`${label} refuses to adopt another owner's live running lease`, async () => { + const store = await create() + await store.claimRoutedPrBabysitter( + 'workspace', + 'agentworkforce/pear#7', + seed, + 'old-owner', + 1_000, + 60_000, + 5, + ) + await store.markRoutedPrBabysitterRunning( + 'workspace', + 'agentworkforce/pear#7', + 'old-owner', + 'agent-a', + 1_001, + ) + + expect(await store.adoptRoutedPrBabysitterClaim( + 'workspace', + 'agentworkforce/pear#7', + 'agent-a', + 'new-owner', + 2_000, + 60_000, + )).toBe(false) + expect((await store.listRoutedPrBabysitterClaims('workspace'))[0]?.[1]).toMatchObject({ + owner: 'old-owner', + leaseUntilMs: 61_000, + }) + }) + + it(`${label} lets the current owner renew a live running lease`, async () => { + const store = await create() + await store.claimRoutedPrBabysitter( + 'workspace', + 'agentworkforce/pear#7', + seed, + 'owner-a', + 1_000, + 60_000, + 5, + ) + await store.markRoutedPrBabysitterRunning( + 'workspace', + 'agentworkforce/pear#7', + 'owner-a', + 'agent-a', + 1_001, + ) + + expect(await store.adoptRoutedPrBabysitterClaim( + 'workspace', + 'agentworkforce/pear#7', + 'agent-a', + 'owner-a', + 2_000, + 60_000, + )).toBe(true) + expect((await store.listRoutedPrBabysitterClaims('workspace'))[0]?.[1]).toMatchObject({ + owner: 'owner-a', + leaseUntilMs: 62_000, + }) + }) + it(`${label} prunes completed claims after bounded retention`, async () => { const store = await create() await store.claimRoutedPrBabysitter( From fd3769d428bc55672403fe15e00c31eba398d41f Mon Sep 17 00:00:00 2001 From: kjgbot Date: Sat, 8 Aug 2026 23:04:48 +0200 Subject: [PATCH 4/4] fix: fence routed babysitter generations --- src/config/schema.test.ts | 17 ++ src/config/schema.ts | 16 +- src/github/routed-pr-babysitter.test.ts | 19 +++ src/github/routed-pr-babysitter.ts | 2 +- src/orchestrator/factory.test.ts | 48 ++++++ src/orchestrator/factory.ts | 161 ++++++++++++++----- src/ports/state.ts | 30 +++- src/state/file-state-store.ts | 80 ++++++--- src/state/in-memory-state-store.ts | 55 +++++-- src/state/routed-pr-babysitter-claim.test.ts | 91 +++++++++-- 10 files changed, 430 insertions(+), 89 deletions(-) diff --git a/src/config/schema.test.ts b/src/config/schema.test.ts index 3fb180fe..a0bf1944 100644 --- a/src/config/schema.test.ts +++ b/src/config/schema.test.ts @@ -330,6 +330,23 @@ describe('FactoryConfigSchema', () => { babysitter: { enabled: true, mode: 'factory-created' }, repos: {}, }).babysitter.mode).toBe('factory-created') + + expect(() => FactoryConfigSchema.parse({ + babysitter: { enabled: true, mode: 'routed-open-prs' }, + repos: { names: ['pear'] }, + })).toThrow(/must resolve at least one owner\/repository route/u) + }) + + it('accepts one-character repository opt-out identities', () => { + const parsed = FactoryConfigSchema.parse({ + babysitter: { + enabled: true, + excludePullRequests: ['owner/r#1'], + }, + repos: {}, + }) + + expect(parsed.babysitter.excludePullRequests).toEqual(['owner/r#1']) }) it('lets explicit byLabel/clonePaths/labels override the derived ones', () => { diff --git a/src/config/schema.ts b/src/config/schema.ts index 30a99314..a2dfd5f4 100644 --- a/src/config/schema.ts +++ b/src/config/schema.ts @@ -131,7 +131,7 @@ const babysitterSchema = z.object({ // before their first provider write. excludeLabels: z.array(z.string().trim().min(1)).default(['factory:skip-babysitter']), excludePullRequests: z.array(z.string().regex( - /^[A-Za-z0-9](?:[A-Za-z0-9_.-]{0,99})\/[A-Za-z0-9_.-]{1,100}#[1-9]\d*$/u, + /^[A-Za-z0-9](?:[A-Za-z0-9_.-]{0,99})\/[A-Za-z0-9_.-]{0,99}#[1-9]\d*$/u, 'expected owner/repo#number', )).default([]), // Routed intake stays quiet until an operator explicitly approves the @@ -333,6 +333,20 @@ const requireRoutedBabysitterRepos = ( path: ['repos', 'names'], message: 'repos.names must contain at least one repository when babysitter.mode is routed-open-prs', }) + return + } + if (cfg.babysitter.mode === 'routed-open-prs') { + const routedRepos = (cfg.repos.names ?? []).map((name) => + cfg.repos.byLabel[name] ?? cfg.repos.overrides[name] ?? + (cfg.repos.org ? `${cfg.repos.org}/${name}` : name) + ) + if (!routedRepos.some((repo) => /^[^/]+\/[^/]+$/u.test(repo))) { + ctx.addIssue({ + code: z.ZodIssueCode.custom, + path: ['repos', 'names'], + message: 'repos.names must resolve at least one owner/repository route when babysitter.mode is routed-open-prs', + }) + } } } diff --git a/src/github/routed-pr-babysitter.test.ts b/src/github/routed-pr-babysitter.test.ts index c31147be..e0f5bb74 100644 --- a/src/github/routed-pr-babysitter.test.ts +++ b/src/github/routed-pr-babysitter.test.ts @@ -54,6 +54,25 @@ describe('routed PR babysitter discovery', () => { ]) }) + it('parses compact aliases for repository names containing double underscores', async () => { + const cfg = FactoryConfigSchema.parse({ + repos: { + names: ['pear__ui'], + byLabel: { pear__ui: 'AgentWorkforce/pear__ui' }, + }, + babysitter: { enabled: true, mode: 'routed-open-prs' }, + }) + const report = await discoverRoutedPullRequests(mount({ + '/github/repos/AgentWorkforce__pear__ui/pulls/by-id/7.json': pull({ + headRepository: { nameWithOwner: 'AgentWorkforce/pear__ui' }, + }), + }), cfg) + + expect(report.candidates.map(({ repo, number }) => `${repo}#${number}`)).toEqual([ + 'AgentWorkforce/pear__ui#7', + ]) + }) + it('continues through an unreadable alias root and an unreadable first PR alias', async () => { const nested = '/github/repos/AgentWorkforce/pear/pulls/7/meta.json' const byID = '/github/repos/AgentWorkforce__pear/pulls/by-id/7.json' diff --git a/src/github/routed-pr-babysitter.ts b/src/github/routed-pr-babysitter.ts index e1d13297..bd036f01 100644 --- a/src/github/routed-pr-babysitter.ts +++ b/src/github/routed-pr-babysitter.ts @@ -138,7 +138,7 @@ export async function discoverRoutedPullRequests( const pullNumberFromPath = (path: string, owner: string, repo: string): number | undefined => { const match = path.match( - /^\/github\/repos\/(?:([^/]+)\/([^/]+)|([^/]+)__([^/]+))\/pulls\/(?:by-id\/)?(\d+)(?:__[^/]*)?(?:\/(?:meta|metadata)\.json|\.json)$/u, + /^\/github\/repos\/(?:([^/]+)\/([^/]+)|([A-Za-z0-9-]+)__([^/]+))\/pulls\/(?:by-id\/)?(\d+)(?:__[^/]*)?(?:\/(?:meta|metadata)\.json|\.json)$/u, ) const actualOwner = match?.[1] ?? match?.[3] const actualRepo = match?.[2] ?? match?.[4] diff --git a/src/orchestrator/factory.test.ts b/src/orchestrator/factory.test.ts index dc0b7918..7ce95dc0 100644 --- a/src/orchestrator/factory.test.ts +++ b/src/orchestrator/factory.test.ts @@ -16281,12 +16281,60 @@ describe('FactoryLoop PR babysitter', () => { expect(fleet.spawns).toHaveLength(1) expect(fleet.spawns[0]?.invocationId).toBe('factory-babysit:agentworkforce/pear#701') + expect(fleet.spawns[0]?.restartPolicy).toEqual({ max_restarts: 0 }) expect(fleet.spawns[0]?.task).toContain('PR title JSON (definition of done): "Repair a human-authored PR"') expect(fleet.spawns[0]?.task).toContain('Before your first provider write') expect(fleet.spawns[0]?.task).toContain('Do not post status comments, mention humans, send notifications, or escalate') expect(fleet.spawns[0]?.task).not.toContain('proactively offer to discuss') }) + it('does not renew durable routed claims during a dry-run sweep', async () => { + const mount = new FakeMountClient({ + '/github/repos/AgentWorkforce__pear/pulls/by-id/704.json': { + number: 704, + title: 'Keep dry-run read-only', + body: 'Do not mutate the existing lease.', + state: 'OPEN', + isDraft: false, + merged: false, + headRefName: 'human/dry-run', + headRefOid: 'head-704', + baseRefName: 'main', + headRepository: { nameWithOwner: 'AgentWorkforce/pear' }, + isCrossRepository: false, + labels: [], + }, + }) + const stateStore = new InMemoryStateStore({ batchSize: 2 }) + const factory = createFactory(config({ + issueSource: 'github', + repos: { + org: 'AgentWorkforce', + names: ['pear'], + clonePaths: { 'AgentWorkforce/pear': '/work/pear' }, + }, + babysitter: { + enabled: true, + mode: 'routed-open-prs', + excludeLabels: [], + excludePullRequests: [], + notifyHumans: false, + }, + }), { + mount, + fleet: new FakeFleetClient(), + stateStore, + triage: new StaticTriage(), + }) + + await factory.runOnce() + const before = (await stateStore.listRoutedPrBabysitterClaims('factory-test'))[0]?.[1] + await factory.runOnce({ dryRun: true }) + const after = (await stateStore.listRoutedPrBabysitterClaims('factory-test'))[0]?.[1] + + expect(after).toEqual(before) + }) + it('checks routed PR opt-out before spawning', async () => { const mount = new FakeMountClient({ '/github/repos/AgentWorkforce__pear/pulls/by-id/702.json': { diff --git a/src/orchestrator/factory.ts b/src/orchestrator/factory.ts index 904f32c2..10f2f8b0 100644 --- a/src/orchestrator/factory.ts +++ b/src/orchestrator/factory.ts @@ -452,7 +452,8 @@ export class FactoryLoop implements Factory { readonly #clarificationWakeOwner = `${process.pid}:${randomUUID()}` readonly #dispatchLifecycleOwner = `${process.pid}:${randomUUID()}` readonly #routedPrBabysitterOwner = `${process.pid}:${randomUUID()}` - readonly #routedPrBabysitterAgents = new Map() + readonly #routedPrBabysitterAgents = new Map() + readonly #issueBabysitterClaims = new Map() #routedPrBabysitterSweep?: Promise readonly #dispatchLifecycleEpochs = new Map() readonly #dispatchTerminalWaiters = new Map void>>() @@ -962,6 +963,7 @@ export class FactoryLoop implements Factory { this.#babysitterReady.clear() this.#babysitterCriticalAgents.clear() this.#routedPrBabysitterAgents.clear() + this.#issueBabysitterClaims.clear() const subscription = this.#subscription this.#subscription = undefined await this.#boundedStopTeardown('factory subscription unsubscribe', () => subscription?.unsubscribe()) @@ -1688,7 +1690,7 @@ export class FactoryLoop implements Factory { } async #restoreRoutedPrBabysitterClaims(): Promise { - if (this.#config.babysitter.mode !== 'routed-open-prs') return + if (this.#config.babysitter.mode !== 'routed-open-prs' || this.#config.dryRun) return try { const roster = await this.#fleet.roster() const online = new Set(roster.agents.map((agent) => agent.name)) @@ -1703,7 +1705,10 @@ export class FactoryLoop implements Factory { ROUTED_PR_BABYSITTER_CLAIM_LEASE_MS, ) if (adopted) { - this.#routedPrBabysitterAgents.set(claim.agentName, identity) + this.#routedPrBabysitterAgents.set(claim.agentName, { + identity, + claimId: adopted.claimId, + }) this.#increment('routedPrBabysitterClaimsRestored') } } else if (claim.status !== 'complete' && claim.leaseUntilMs > this.#clock.now()) { @@ -1740,28 +1745,36 @@ export class FactoryLoop implements Factory { async #runRoutedPrBabysitterSweep(dryRun: boolean): Promise { const discovery = await discoverRoutedPullRequests(this.#mount, this.#config) - const priorClaims = await this.#state.listRoutedPrBabysitterClaims(this.#workspaceId) - for (const [identity, claim] of priorClaims) { - if ( - claim.owner === this.#routedPrBabysitterOwner && - claim.status === 'running' && - claim.agentName && - this.#routedPrBabysitterAgents.get(claim.agentName) === identity - ) { - await this.#state.claimRoutedPrBabysitter( - this.#workspaceId, - identity, - { - repo: claim.repo, - prNumber: claim.prNumber, - revision: claim.revision, - source: claim.source, - }, - this.#routedPrBabysitterOwner, - this.#clock.now(), - ROUTED_PR_BABYSITTER_CLAIM_LEASE_MS, - this.#config.batchSize, - ) + if (!dryRun) { + const priorClaims = await this.#state.listRoutedPrBabysitterClaims(this.#workspaceId) + for (const [identity, claim] of priorClaims) { + if ( + claim.owner === this.#routedPrBabysitterOwner && + claim.status === 'running' && + claim.agentName && + this.#routedPrBabysitterAgents.get(claim.agentName)?.identity === identity + ) { + const renewed = await this.#state.claimRoutedPrBabysitter( + this.#workspaceId, + identity, + { + repo: claim.repo, + prNumber: claim.prNumber, + revision: claim.revision, + source: claim.source, + }, + this.#routedPrBabysitterOwner, + this.#clock.now(), + ROUTED_PR_BABYSITTER_CLAIM_LEASE_MS, + this.#config.batchSize, + ) + if (renewed.claim) { + this.#routedPrBabysitterAgents.set(claim.agentName, { + identity, + claimId: renewed.claim.claimId, + }) + } + } } } const stats = { @@ -1817,14 +1830,16 @@ export class FactoryLoop implements Factory { stats.unchanged += 1 continue } + if (claim.outcome !== 'claimed') continue try { - await this.#spawnRoutedPrBabysitter(candidate, identity) + await this.#spawnRoutedPrBabysitter(candidate, identity, claim.claim.claimId) stats.admitted += 1 } catch (error) { await this.#state.releaseRoutedPrBabysitterClaim( this.#workspaceId, identity, this.#routedPrBabysitterOwner, + claim.claim.claimId, ) this.#increment('routedPrBabysitterSpawnFailures') stats.spawnFailures += 1 @@ -1843,7 +1858,11 @@ export class FactoryLoop implements Factory { this.#logger.info?.('[factory] routed PR babysitter sweep completed', stats) } - async #spawnRoutedPrBabysitter(candidate: RoutedPrCandidate, identity: string): Promise { + async #spawnRoutedPrBabysitter( + candidate: RoutedPrCandidate, + identity: string, + claimId: string, + ): Promise { const clonePath = this.#config.repos.clonePaths[candidate.repo] const issue = { key: `${candidate.repo}#${candidate.number}`, @@ -1890,18 +1909,20 @@ export class FactoryLoop implements Factory { model: this.#config.models.babysitter, cwd: clonePath, invocationId: `factory-babysit:${identity}`, + restartPolicy: { max_restarts: 0 }, }) if (!await this.#state.markRoutedPrBabysitterRunning( this.#workspaceId, identity, this.#routedPrBabysitterOwner, + claimId, result.name, this.#clock.now(), )) { await this.#fleet.release(result.name, 'routed-pr-claim-lost') throw new Error(`Routed PR babysitter claim lost after spawning ${identity}`) } - this.#routedPrBabysitterAgents.set(result.name, identity) + this.#routedPrBabysitterAgents.set(result.name, { identity, claimId }) this.#increment('routedPrBabysittersSpawned') } @@ -5930,13 +5951,15 @@ export class FactoryLoop implements Factory { return } - const routedIdentity = this.#routedPrBabysitterAgents.get(name) - if (routedIdentity) { + const routedClaim = this.#routedPrBabysitterAgents.get(name) + if (routedClaim) { this.#routedPrBabysitterAgents.delete(name) if (isCompletionReason(reason)) { await this.#state.completeRoutedPrBabysitter( this.#workspaceId, - routedIdentity, + routedClaim.identity, + this.#routedPrBabysitterOwner, + routedClaim.claimId, name, this.#clock.now(), ) @@ -5944,8 +5967,9 @@ export class FactoryLoop implements Factory { } else { await this.#state.releaseRoutedPrBabysitterClaim( this.#workspaceId, - routedIdentity, + routedClaim.identity, this.#routedPrBabysitterOwner, + routedClaim.claimId, ) this.#increment('routedPrBabysitterAbnormalExits') } @@ -6008,14 +6032,18 @@ export class FactoryLoop implements Factory { if (tracingReconciledExit) this.#logger.info?.('[factory] reconciled agent exit question replay completed', { issue: record.issue.key, name }) const exiting = record.agents.get(name) - if (isCompletionReason(reason) && exiting?.spec.role === 'babysitter' && exiting.spec.ownedPullRequest) { + const issueClaim = this.#issueBabysitterClaims.get(name) + if (isCompletionReason(reason) && exiting?.spec.role === 'babysitter' && issueClaim) { await this.#state.completeRoutedPrBabysitter( this.#workspaceId, - routedPrIdentity(exiting.spec.ownedPullRequest.repo, exiting.spec.ownedPullRequest.number), + issueClaim.identity, + this.#dispatchLifecycleOwner, + issueClaim.claimId, name, this.#clock.now(), ) } + this.#issueBabysitterClaims.delete(name) if (exiting) await this.#reportAgent(record, exiting, 'agent.exited', { releaseReason: reason }) if (tracingReconciledExit) this.#logger.info?.('[factory] reconciled agent exit telemetry completed', { issue: record.issue.key, name }) @@ -10585,7 +10613,7 @@ export class FactoryLoop implements Factory { return } const prWorkIdentity = routedPrIdentity(prRef.repo, prRef.prNumber) - const prWorkClaim = await this.#state.claimRoutedPrBabysitter( + let prWorkClaim = await this.#state.claimRoutedPrBabysitter( this.#workspaceId, prWorkIdentity, { @@ -10599,7 +10627,45 @@ export class FactoryLoop implements Factory { ROUTED_PR_BABYSITTER_CLAIM_LEASE_MS, Number.MAX_SAFE_INTEGER, ) - if (prWorkClaim.outcome === 'already-running' && prWorkClaim.claim?.source === 'routed-open-prs') { + if ( + prWorkClaim.outcome === 'already-running' && + prWorkClaim.claim?.source === 'issue-created' + ) { + const adoptedEntry = prWorkClaim.claim.agentName + ? [prWorkClaim.claim.agentName, record.agents.get(prWorkClaim.claim.agentName)] as const + : [...record.agents.entries()].find(([, agent]) => { + const owned = agent.spec.ownedPullRequest + return agent.spec.role === 'babysitter' && owned && + githubPrIdentity(owned.repo, owned.number) === githubPrIdentity(prRef.repo, prRef.prNumber) + }) + const adoptedAgentName = adoptedEntry?.[0] + const adoptedAgent = adoptedEntry?.[1] + const owned = adoptedAgent?.spec.ownedPullRequest + if ( + adoptedAgentName && + adoptedAgent && + owned && + githubPrIdentity(owned.repo, owned.number) === githubPrIdentity(prRef.repo, prRef.prNumber) + ) { + // The dispatch lifecycle ownership assertion above is the outer fence + // proving the prior Factory owner is gone. Atomically transfer this + // subordinate PR claim so the already-adopted worker is not respawned. + const adopted = await this.#state.adoptRoutedPrBabysitterClaim( + this.#workspaceId, + prWorkIdentity, + adoptedAgentName, + this.#dispatchLifecycleOwner, + this.#clock.now(), + ROUTED_PR_BABYSITTER_CLAIM_LEASE_MS, + true, + ) + if (adopted) prWorkClaim = { outcome: 'owned', claim: adopted } + } + } + if ( + (prWorkClaim.outcome !== 'claimed' && prWorkClaim.outcome !== 'owned') || + !prWorkClaim.claim + ) { this.#increment('babysitterOwnershipConflictsSuppressed') this.#logger.warn?.('[factory] issue-created babysitter suppressed by existing PR work-unit claim', { issue: record.issue.key, @@ -10608,6 +10674,7 @@ export class FactoryLoop implements Factory { }) return } + const prWorkClaimId = prWorkClaim.claim.claimId const replacedSuperseded = prRef.authoritative ? await this.#retireSupersededBabysitters(record, prRef) : false @@ -10660,13 +10727,22 @@ export class FactoryLoop implements Factory { this.#babysitterSpawned.add(babysitterKey) const ref = this.#babysitterPr.get(babysitterKey)! await this.#persistBabysitterSession(record.issue, ref, tracked) - await this.#state.markRoutedPrBabysitterRunning( + const markedRunning = await this.#state.markRoutedPrBabysitterRunning( this.#workspaceId, prWorkIdentity, this.#dispatchLifecycleOwner, + prWorkClaimId, tracked.result?.name ?? trackedName, this.#clock.now(), ) + if (!markedRunning) { + await this.#fleet.release(tracked.result?.name ?? trackedName, 'pr-work-claim-lost') + return + } + this.#issueBabysitterClaims.set(tracked.result?.name ?? trackedName, { + identity: prWorkIdentity, + claimId: prWorkClaimId, + }) await this.#ensureBabysitterResourceSubscription(record.issue, ref, tracked) await this.#retargetSlackConversationToBabysitter(record) return @@ -10770,13 +10846,22 @@ export class FactoryLoop implements Factory { }) const ref = this.#babysitterPr.get(babysitterKey)! await this.#persistBabysitterSession(record.issue, ref, tracked) - await this.#state.markRoutedPrBabysitterRunning( + const markedRunning = await this.#state.markRoutedPrBabysitterRunning( this.#workspaceId, prWorkIdentity, this.#dispatchLifecycleOwner, + prWorkClaimId, tracked?.result?.name ?? spawned.name, this.#clock.now(), ) + if (!markedRunning) { + await this.#fleet.release(tracked?.result?.name ?? spawned.name, 'pr-work-claim-lost') + return + } + this.#issueBabysitterClaims.set(tracked?.result?.name ?? spawned.name, { + identity: prWorkIdentity, + claimId: prWorkClaimId, + }) await this.#ensureBabysitterResourceSubscription(record.issue, ref, tracked) await this.#retargetSlackConversationToBabysitter(record) await this.#writeInFlightRegistry() diff --git a/src/ports/state.ts b/src/ports/state.ts index e2198d53..c30e44a4 100644 --- a/src/ports/state.ts +++ b/src/ports/state.ts @@ -92,6 +92,8 @@ export type BabysitterSessionState = { } export type RoutedPrBabysitterClaim = { + /** Opaque fencing token for one admission generation. */ + claimId: string repo: string prNumber: number /** Stable hash of the mounted PR snapshot which admitted this run. */ @@ -403,13 +405,20 @@ export interface StateStore { claimRoutedPrBabysitter( workspaceId: string, identity: string, - seed: Omit, + seed: Omit, owner: string, nowMs: number, leaseMs: number, maxActive: number, ): Promise - markRoutedPrBabysitterRunning(workspaceId: string, identity: string, owner: string, agentName: string, nowMs: number): Promise + markRoutedPrBabysitterRunning( + workspaceId: string, + identity: string, + owner: string, + claimId: string, + agentName: string, + nowMs: number, + ): Promise adoptRoutedPrBabysitterClaim( workspaceId: string, identity: string, @@ -417,9 +426,22 @@ export interface StateStore { owner: string, nowMs: number, leaseMs: number, + allowLiveLeaseTransfer?: boolean, + ): Promise + completeRoutedPrBabysitter( + workspaceId: string, + identity: string, + owner: string, + claimId: string, + agentName: string, + nowMs: number, + ): Promise + releaseRoutedPrBabysitterClaim( + workspaceId: string, + identity: string, + owner: string, + claimId: string, ): Promise - completeRoutedPrBabysitter(workspaceId: string, identity: string, agentName: string, nowMs: number): Promise - releaseRoutedPrBabysitterClaim(workspaceId: string, identity: string, owner: string): Promise listRoutedPrBabysitterClaims(workspaceId: string): Promise> recordCanonicalState(workspaceId: string, key: string, stateId: string): Promise diff --git a/src/state/file-state-store.ts b/src/state/file-state-store.ts index 3c002e66..e46d98fe 100644 --- a/src/state/file-state-store.ts +++ b/src/state/file-state-store.ts @@ -31,7 +31,7 @@ type PersistedWorkspaceState = { } type WatchStateDocument = { - version: 3 + version: 4 workspaces: Record } @@ -655,7 +655,7 @@ export class FileStateStore extends InMemoryStateStore { override async claimRoutedPrBabysitter( workspaceId: string, identity: string, - seed: Omit, + seed: Omit, owner: string, nowMs: number, leaseMs: number, @@ -666,8 +666,9 @@ export class FileStateStore extends InMemoryStateStore { const workspace = document.workspaces[workspaceId] ??= emptyWorkspaceState() for (const [key, claim] of Object.entries(workspace.routedPrBabysitterClaims)) { if ( - claim.status === 'complete' && - claim.updatedAtMs <= nowMs - ROUTED_PR_BABYSITTER_COMPLETED_RETENTION_MS + (claim.status === 'complete' && + claim.updatedAtMs <= nowMs - ROUTED_PR_BABYSITTER_COMPLETED_RETENTION_MS) || + (claim.status !== 'complete' && claim.leaseUntilMs <= nowMs) ) delete workspace.routedPrBabysitterClaims[key] } const existing = workspace.routedPrBabysitterClaims[identity] @@ -695,6 +696,7 @@ export class FileStateStore extends InMemoryStateStore { if (active >= Math.max(1, Math.trunc(maxActive))) return { outcome: 'capacity' } const claim: RoutedPrBabysitterClaim = { ...seed, + claimId: randomUUID(), status: 'claimed', owner, leaseUntilMs: nowMs + leaseMs, @@ -711,13 +713,20 @@ export class FileStateStore extends InMemoryStateStore { workspaceId: string, identity: string, owner: string, + claimId: string, agentName: string, nowMs: number, ): Promise { return await this.#exclusive(async () => this.#withMutationLock(async () => { const document = await this.#loadFromDisk() const claim = document.workspaces[workspaceId]?.routedPrBabysitterClaims[identity] - if (!claim || claim.owner !== owner || claim.status === 'complete') return false + if ( + !claim || + claim.owner !== owner || + claim.claimId !== claimId || + claim.status === 'complete' || + claim.leaseUntilMs <= nowMs + ) return false claim.status = 'running' claim.agentName = agentName claim.updatedAtMs = nowMs @@ -733,34 +742,47 @@ export class FileStateStore extends InMemoryStateStore { owner: string, nowMs: number, leaseMs: number, - ): Promise { + allowLiveLeaseTransfer = false, + ): Promise { return await this.#exclusive(async () => this.#withMutationLock(async () => { const document = await this.#loadFromDisk() const claim = document.workspaces[workspaceId]?.routedPrBabysitterClaims[identity] if ( !claim || - claim.status !== 'running' || - claim.agentName !== agentName || - (claim.owner !== owner && claim.leaseUntilMs > nowMs) - ) return false + claim.status === 'complete' || + (claim.status === 'running' && claim.agentName !== agentName) || + (claim.status === 'claimed' && !allowLiveLeaseTransfer) || + (claim.owner !== owner && claim.leaseUntilMs > nowMs && !allowLiveLeaseTransfer) + ) return undefined + if (claim.owner !== owner) claim.claimId = randomUUID() claim.owner = owner + claim.status = 'running' + claim.agentName = agentName claim.leaseUntilMs = nowMs + leaseMs claim.updatedAtMs = nowMs await this.#persist(document) - return true + return structuredClone(claim) })) } override async completeRoutedPrBabysitter( workspaceId: string, identity: string, + owner: string, + claimId: string, agentName: string, nowMs: number, ): Promise { return await this.#exclusive(async () => this.#withMutationLock(async () => { const document = await this.#loadFromDisk() const claim = document.workspaces[workspaceId]?.routedPrBabysitterClaims[identity] - if (!claim || claim.agentName !== agentName || claim.status === 'complete') return false + if ( + !claim || + claim.owner !== owner || + claim.claimId !== claimId || + claim.agentName !== agentName || + claim.status === 'complete' + ) return false claim.status = 'complete' claim.leaseUntilMs = nowMs claim.updatedAtMs = nowMs @@ -769,12 +791,17 @@ export class FileStateStore extends InMemoryStateStore { })) } - override async releaseRoutedPrBabysitterClaim(workspaceId: string, identity: string, owner: string): Promise { + override async releaseRoutedPrBabysitterClaim( + workspaceId: string, + identity: string, + owner: string, + claimId: string, + ): Promise { return await this.#exclusive(async () => this.#withMutationLock(async () => { const document = await this.#loadFromDisk() const workspace = document.workspaces[workspaceId] const claim = workspace?.routedPrBabysitterClaims[identity] - if (!workspace || !claim || claim.owner !== owner) return false + if (!workspace || !claim || claim.owner !== owner || claim.claimId !== claimId) return false delete workspace.routedPrBabysitterClaims[identity] if (workspaceIsEmpty(workspace)) delete document.workspaces[workspaceId] await this.#persist(document) @@ -964,7 +991,7 @@ export class FileStateStore extends InMemoryStateStore { return parseDocument(parsed) } catch (error) { if (!isMissingFileError(error)) throw error - return { version: 3, workspaces: {} } + return { version: 4, workspaces: {} } } } @@ -1018,7 +1045,7 @@ const parseDocument = (value: unknown): WatchStateDocument => { if (!isRecord(value) || !isRecord(value.workspaces)) { throw new Error('Factory GitHub watch state file is invalid') } - if (value.version === 3) { + if (value.version === 4 || value.version === 3) { const workspaces: Record = {} for (const [workspaceId, rawWorkspace] of Object.entries(value.workspaces)) { if (!isRecord(rawWorkspace)) throw new Error('Factory GitHub watch state file is invalid') @@ -1042,12 +1069,12 @@ const parseDocument = (value: unknown): WatchStateDocument => { githubIssueCommentWatches: watches as Record, waitingClarifications: clarifications as Record, babysitterSessions: parseBabysitterSessions(babysitters ?? {}), - routedPrBabysitterClaims: parseRoutedPrBabysitterClaims(routedClaims ?? {}), + routedPrBabysitterClaims: parseRoutedPrBabysitterClaims(routedClaims ?? {}, value.version === 3), conversationSessions: parseConversationSessions(conversations ?? {}), dispatchLifecycles: (lifecycles ?? {}) as Record, } } - return { version: 3, workspaces } + return { version: 4, workspaces } } if (value.version === 2) { const workspaces: Record = {} @@ -1068,7 +1095,7 @@ const parseDocument = (value: unknown): WatchStateDocument => { dispatchLifecycles: {}, } } - return { version: 3, workspaces } + return { version: 4, workspaces } } if (value.version === 1) { const workspaces: Record = {} @@ -1085,7 +1112,7 @@ const parseDocument = (value: unknown): WatchStateDocument => { dispatchLifecycles: {}, } } - return { version: 3, workspaces } + return { version: 4, workspaces } } throw new Error('Factory GitHub watch state file is invalid') } @@ -1099,11 +1126,15 @@ const cloneClarification = (record: WaitingClarification): WaitingClarification const cloneBabysitterSession = (session: BabysitterSessionState): BabysitterSessionState => structuredClone(session) -const parseRoutedPrBabysitterClaims = (value: Record): Record => { +const parseRoutedPrBabysitterClaims = ( + value: Record, + legacyV3 = false, +): Record => { const claims: Record = {} for (const [identity, candidate] of Object.entries(value)) { if ( !isRecord(candidate) || + (!legacyV3 && typeof candidate.claimId !== 'string') || typeof candidate.repo !== 'string' || !Number.isSafeInteger(candidate.prNumber) || (candidate.prNumber as number) < 1 || @@ -1118,7 +1149,12 @@ const parseRoutedPrBabysitterClaims = (value: Record): Record), + claimId: typeof candidate.claimId === 'string' + ? candidate.claimId + : `${candidate.owner}:${candidate.claimedAtMs}:${identity}`, + } } return claims } diff --git a/src/state/in-memory-state-store.ts b/src/state/in-memory-state-store.ts index f2160005..f5bfd9f4 100644 --- a/src/state/in-memory-state-store.ts +++ b/src/state/in-memory-state-store.ts @@ -1,3 +1,5 @@ +import { randomUUID } from 'node:crypto' + import { BatchTracker } from '../orchestrator/batch-tracker' import { githubRepositoriesMatch } from '../github/repo-identity' import { ROUTED_PR_BABYSITTER_COMPLETED_RETENTION_MS } from '../ports/state' @@ -640,7 +642,7 @@ export class InMemoryStateStore implements StateStore { async claimRoutedPrBabysitter( workspaceId: string, identity: string, - seed: Omit, + seed: Omit, owner: string, nowMs: number, leaseMs: number, @@ -649,8 +651,9 @@ export class InMemoryStateStore implements StateStore { const claims = this.#workspace(workspaceId).routedPrBabysitterClaims for (const [key, claim] of claims) { if ( - claim.status === 'complete' && - claim.updatedAtMs <= nowMs - ROUTED_PR_BABYSITTER_COMPLETED_RETENTION_MS + (claim.status === 'complete' && + claim.updatedAtMs <= nowMs - ROUTED_PR_BABYSITTER_COMPLETED_RETENTION_MS) || + (claim.status !== 'complete' && claim.leaseUntilMs <= nowMs) ) claims.delete(key) } const existing = claims.get(identity) @@ -677,6 +680,7 @@ export class InMemoryStateStore implements StateStore { if (active >= Math.max(1, Math.trunc(maxActive))) return { outcome: 'capacity' } const claim: RoutedPrBabysitterClaim = { ...seed, + claimId: randomUUID(), status: 'claimed', owner, leaseUntilMs: nowMs + leaseMs, @@ -691,11 +695,18 @@ export class InMemoryStateStore implements StateStore { workspaceId: string, identity: string, owner: string, + claimId: string, agentName: string, nowMs: number, ): Promise { const claim = this.#workspace(workspaceId).routedPrBabysitterClaims.get(identity) - if (!claim || claim.owner !== owner || claim.status === 'complete') return false + if ( + !claim || + claim.owner !== owner || + claim.claimId !== claimId || + claim.status === 'complete' || + claim.leaseUntilMs <= nowMs + ) return false claim.status = 'running' claim.agentName = agentName claim.updatedAtMs = nowMs @@ -709,38 +720,56 @@ export class InMemoryStateStore implements StateStore { owner: string, nowMs: number, leaseMs: number, - ): Promise { + allowLiveLeaseTransfer = false, + ): Promise { const claim = this.#workspace(workspaceId).routedPrBabysitterClaims.get(identity) if ( !claim || - claim.status !== 'running' || - claim.agentName !== agentName || - (claim.owner !== owner && claim.leaseUntilMs > nowMs) - ) return false + claim.status === 'complete' || + (claim.status === 'running' && claim.agentName !== agentName) || + (claim.status === 'claimed' && !allowLiveLeaseTransfer) || + (claim.owner !== owner && claim.leaseUntilMs > nowMs && !allowLiveLeaseTransfer) + ) return undefined + if (claim.owner !== owner) claim.claimId = randomUUID() claim.owner = owner + claim.status = 'running' + claim.agentName = agentName claim.leaseUntilMs = nowMs + leaseMs claim.updatedAtMs = nowMs - return true + return structuredClone(claim) } async completeRoutedPrBabysitter( workspaceId: string, identity: string, + owner: string, + claimId: string, agentName: string, nowMs: number, ): Promise { const claim = this.#workspace(workspaceId).routedPrBabysitterClaims.get(identity) - if (!claim || claim.agentName !== agentName || claim.status === 'complete') return false + if ( + !claim || + claim.owner !== owner || + claim.claimId !== claimId || + claim.agentName !== agentName || + claim.status === 'complete' + ) return false claim.status = 'complete' claim.leaseUntilMs = nowMs claim.updatedAtMs = nowMs return true } - async releaseRoutedPrBabysitterClaim(workspaceId: string, identity: string, owner: string): Promise { + async releaseRoutedPrBabysitterClaim( + workspaceId: string, + identity: string, + owner: string, + claimId: string, + ): Promise { const claims = this.#workspace(workspaceId).routedPrBabysitterClaims const claim = claims.get(identity) - if (!claim || claim.owner !== owner) return false + if (!claim || claim.owner !== owner || claim.claimId !== claimId) return false return claims.delete(identity) } diff --git a/src/state/routed-pr-babysitter-claim.test.ts b/src/state/routed-pr-babysitter-claim.test.ts index 0481b640..2d8a5f39 100644 --- a/src/state/routed-pr-babysitter-claim.test.ts +++ b/src/state/routed-pr-babysitter-claim.test.ts @@ -32,7 +32,14 @@ const verifyClaims = (label: string, create: () => Promise | StateSt const store = await create() const first = await store.claimRoutedPrBabysitter('workspace', 'agentworkforce/pear#7', seed, 'owner-a', 1_000, 60_000, 1) expect(first.outcome).toBe('claimed') - expect(await store.markRoutedPrBabysitterRunning('workspace', 'agentworkforce/pear#7', 'owner-a', 'agent-a', 1_001)).toBe(true) + expect(await store.markRoutedPrBabysitterRunning( + 'workspace', + 'agentworkforce/pear#7', + 'owner-a', + first.claim!.claimId, + 'agent-a', + 1_001, + )).toBe(true) expect((await store.claimRoutedPrBabysitter( 'workspace', 'agentworkforce/pear#8', @@ -42,7 +49,14 @@ const verifyClaims = (label: string, create: () => Promise | StateSt 60_000, 1, )).outcome).toBe('capacity') - expect(await store.completeRoutedPrBabysitter('workspace', 'agentworkforce/pear#7', 'agent-a', 1_003)).toBe(true) + expect(await store.completeRoutedPrBabysitter( + 'workspace', + 'agentworkforce/pear#7', + 'owner-a', + first.claim!.claimId, + 'agent-a', + 1_003, + )).toBe(true) expect((await store.claimRoutedPrBabysitter( 'workspace', 'agentworkforce/pear#7', @@ -83,6 +97,8 @@ const verifyClaims = (label: string, create: () => Promise | StateSt 60_000, 1, ) + expect((await store.listRoutedPrBabysitterClaims('workspace')) + .map(([identity]) => identity)).toEqual(['agentworkforce/pear#8']) expect((await store.claimRoutedPrBabysitter( 'workspace', 'agentworkforce/pear#7', @@ -117,9 +133,56 @@ const verifyClaims = (label: string, create: () => Promise | StateSt )).outcome).toBe('already-running') }) + it(`${label} fences delayed transitions from an expired claim generation`, async () => { + const store = await create() + const expired = await store.claimRoutedPrBabysitter( + 'workspace', + 'agentworkforce/pear#7', + seed, + 'owner-a', + 1_000, + 1_000, + 5, + ) + const current = await store.claimRoutedPrBabysitter( + 'workspace', + 'agentworkforce/pear#7', + { ...seed, revision: 'rev-2' }, + 'owner-a', + 2_001, + 60_000, + 5, + ) + expect(current.outcome).toBe('claimed') + expect(await store.markRoutedPrBabysitterRunning( + 'workspace', + 'agentworkforce/pear#7', + 'owner-a', + expired.claim!.claimId, + 'stale-agent', + 2_002, + )).toBe(false) + expect(await store.markRoutedPrBabysitterRunning( + 'workspace', + 'agentworkforce/pear#7', + 'owner-a', + current.claim!.claimId, + 'current-agent', + 2_002, + )).toBe(true) + expect(await store.completeRoutedPrBabysitter( + 'workspace', + 'agentworkforce/pear#7', + 'owner-a', + expired.claim!.claimId, + 'current-agent', + 2_003, + )).toBe(false) + }) + it(`${label} atomically adopts an expired restored running claim`, async () => { const store = await create() - await store.claimRoutedPrBabysitter( + const original = await store.claimRoutedPrBabysitter( 'workspace', 'agentworkforce/pear#7', seed, @@ -132,17 +195,19 @@ const verifyClaims = (label: string, create: () => Promise | StateSt 'workspace', 'agentworkforce/pear#7', 'old-owner', + original.claim!.claimId, 'agent-a', 1_001, ) - expect(await store.adoptRoutedPrBabysitterClaim( + const adopted = await store.adoptRoutedPrBabysitterClaim( 'workspace', 'agentworkforce/pear#7', 'agent-a', 'new-owner', 61_001, 60_000, - )).toBe(true) + ) + expect(adopted).toMatchObject({ owner: 'new-owner' }) expect((await store.listRoutedPrBabysitterClaims('workspace'))[0]?.[1]).toMatchObject({ owner: 'new-owner', leaseUntilMs: 121_001, @@ -153,12 +218,13 @@ const verifyClaims = (label: string, create: () => Promise | StateSt 'workspace', 'agentworkforce/pear#7', 'old-owner', + original.claim!.claimId, )).toBe(false) }) it(`${label} refuses to adopt another owner's live running lease`, async () => { const store = await create() - await store.claimRoutedPrBabysitter( + const original = await store.claimRoutedPrBabysitter( 'workspace', 'agentworkforce/pear#7', seed, @@ -171,6 +237,7 @@ const verifyClaims = (label: string, create: () => Promise | StateSt 'workspace', 'agentworkforce/pear#7', 'old-owner', + original.claim!.claimId, 'agent-a', 1_001, ) @@ -182,7 +249,7 @@ const verifyClaims = (label: string, create: () => Promise | StateSt 'new-owner', 2_000, 60_000, - )).toBe(false) + )).toBeUndefined() expect((await store.listRoutedPrBabysitterClaims('workspace'))[0]?.[1]).toMatchObject({ owner: 'old-owner', leaseUntilMs: 61_000, @@ -191,7 +258,7 @@ const verifyClaims = (label: string, create: () => Promise | StateSt it(`${label} lets the current owner renew a live running lease`, async () => { const store = await create() - await store.claimRoutedPrBabysitter( + const original = await store.claimRoutedPrBabysitter( 'workspace', 'agentworkforce/pear#7', seed, @@ -204,6 +271,7 @@ const verifyClaims = (label: string, create: () => Promise | StateSt 'workspace', 'agentworkforce/pear#7', 'owner-a', + original.claim!.claimId, 'agent-a', 1_001, ) @@ -215,7 +283,7 @@ const verifyClaims = (label: string, create: () => Promise | StateSt 'owner-a', 2_000, 60_000, - )).toBe(true) + )).toMatchObject({ owner: 'owner-a', claimId: original.claim!.claimId }) expect((await store.listRoutedPrBabysitterClaims('workspace'))[0]?.[1]).toMatchObject({ owner: 'owner-a', leaseUntilMs: 62_000, @@ -224,7 +292,7 @@ const verifyClaims = (label: string, create: () => Promise | StateSt it(`${label} prunes completed claims after bounded retention`, async () => { const store = await create() - await store.claimRoutedPrBabysitter( + const original = await store.claimRoutedPrBabysitter( 'workspace', 'agentworkforce/pear#7', seed, @@ -237,12 +305,15 @@ const verifyClaims = (label: string, create: () => Promise | StateSt 'workspace', 'agentworkforce/pear#7', 'owner-a', + original.claim!.claimId, 'agent-a', 1_001, ) await store.completeRoutedPrBabysitter( 'workspace', 'agentworkforce/pear#7', + 'owner-a', + original.claim!.claimId, 'agent-a', 1_002, )