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..64806188 100644 --- a/src/config/schema.test.ts +++ b/src/config/schema.test.ts @@ -3,6 +3,7 @@ import { homedir } from 'node:os' import { join } from 'node:path' import { FactoryConfigSchema, NodeConfigSchema, loadFactoryConfig } from './schema' +import { routedPrRepos } from '../github/routed-pr-babysitter' describe('FactoryConfigSchema', () => { it('parses a valid config and applies defaults', () => { @@ -36,7 +37,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') @@ -314,6 +321,65 @@ 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') + + expect(() => FactoryConfigSchema.parse({ + babysitter: { enabled: true, mode: 'routed-open-prs' }, + repos: { names: ['pear'] }, + })).toThrow(/must resolve at least one owner\/repository route/u) + }) + + it('does not reject a byLabel bare-name entry that org would successfully resolve at runtime', () => { + // requireRoutedBabysitterRepos runs in a superRefine BEFORE the + // normalizeFactoryConfig transform, so it only sees the raw pre-merge + // repos.byLabel the user wrote -- it previously replicated the org-prefix + // fallback by hand, but only for names with no byLabel/overrides entry at + // all. An explicit byLabel value that itself lacks a slash (a common + // shorthand: map a label to a bare repo name, let `org` supply the + // owner) got no such rescue and was wrongly rejected here, even though + // routedPrRepos applies exactly that rescue to any resolved value + // lacking a slash, regardless of where it came from. + const parsed = FactoryConfigSchema.parse({ + babysitter: { enabled: true, mode: 'routed-open-prs' }, + repos: { names: ['pear'], org: 'AgentWorkforce', byLabel: { pear: 'pear-fork' } }, + }) + expect(parsed.repos.byLabel.pear).toBe('pear-fork') + expect(routedPrRepos(parsed)).toEqual(['AgentWorkforce/pear-fork']) + }) + + 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('mirrors the owner-segment length rule onto the repo segment', () => { + const withRepo = (repo: string) => FactoryConfigSchema.parse({ + babysitter: { enabled: true, excludePullRequests: [`owner/${repo}#1`] }, + repos: {}, + }) + + expect(() => withRepo('')).toThrow(/expected owner\/repo#number/u) + expect(withRepo('r').babysitter.excludePullRequests).toEqual(['owner/r#1']) + expect(withRepo('r'.repeat(100)).babysitter.excludePullRequests).toEqual([`owner/${'r'.repeat(100)}#1`]) + expect(() => withRepo('r'.repeat(101))).toThrow(/expected owner\/repo#number/u) + }) + 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 aed87a56..9129d065 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](?:[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 + // notification shape observed in a read-only sweep. + notifyHumans: z.boolean().default(false), }).default({}) const reportingSchema = z.object({ @@ -307,9 +323,55 @@ const NodeConfigObjectSchema = z.object({ const FactoryConfigObjectSchema = WorkspaceConfigObjectSchema.merge(NodeConfigObjectSchema) -export const WorkspaceConfigSchema = WorkspaceConfigObjectSchema.transform((cfg) => normalizeWorkspaceConfig(cfg)) +// Shared with routedPrRepos (src/github/routed-pr-babysitter.ts): the last +// step in resolving a repos.names entry to a routable owner/repo slug. A +// resolved value that already has a slash is used as-is; one that doesn't +// gets `org` prefixed as a rescue, regardless of whether it came from an +// explicit byLabel entry, overrides, or the bare name itself -- that rescue +// used to live only in routedPrRepos, so a validator that ran before +// normalizeFactoryConfig's transform (and so never saw org get applied) +// could reject a config routedPrRepos would resolve fine at runtime. +export const resolveRoutedRepo = (configured: string, org: string | undefined): string | undefined => { + const repo = configured.includes('/') ? configured : (org ? `${org}/${configured}` : undefined) + return repo && /^[^/]+\/[^/]+$/u.test(repo) ? repo : undefined +} + +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', + }) + return + } + if (cfg.babysitter.mode === 'routed-open-prs') { + // Mirror resolveRepos' own derivation (overrides win over a bare + // org/name, explicit byLabel wins over that) instead of a second, + // hand-rolled copy of that fallback chain that can drift from it. + const { byLabel } = resolveRepos(cfg.repos, cfg.repos.cloneRoot) + const routedRepos = (cfg.repos.names ?? []) + .map((name) => resolveRoutedRepo(byLabel[name] ?? name, cfg.repos.org)) + if (!routedRepos.some(Boolean)) { + 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', + }) + } + } +} + +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/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..e0f5bb74 --- /dev/null +++ b/src/github/routed-pr-babysitter.test.ts @@ -0,0 +1,149 @@ +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('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' + 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({ + 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..16333fe5 --- /dev/null +++ b/src/github/routed-pr-babysitter.ts @@ -0,0 +1,237 @@ +import type { FactoryConfig } from '../config/schema' +import { resolveRoutedRepo } 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 repo = resolveRoutedRepo(config.repos.byLabel[name] ?? name, config.repos.org) + if (!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() + 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)}`, + }) + } + } + if (!readableRoot) 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 + } + 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 + } + seen.add(identity) + 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\/(?:([^/]+)\/([^/]+)|([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] + 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_changed ?? 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) => { + 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 +} + +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..d468096b 100644 --- a/src/orchestrator/factory.test.ts +++ b/src/orchestrator/factory.test.ts @@ -16241,6 +16241,215 @@ 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]?.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': { + 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([]) + }) + + 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', @@ -17053,6 +17262,280 @@ describe('FactoryLoop PR babysitter', () => { expect(fleet.spawns.filter((s) => s.name === 'ar-403-babysit')).toHaveLength(1) }) + // This asserts what the fix actually guarantees: the claim gets + // renegotiated (markRoutedPrBabysitterRunning is called again) instead of + // latching forever on stale local bookkeeping. It does NOT assert a real + // second OS-level process spawns — that is separately gated by + // #spawnAgent's own crash-recovery idempotency layer (batch.shouldSpawn on + // a deterministic invocationId), which has no way to learn that a later + // release invalidated an earlier "already dispatched" record. Tracked as + // AgentWorkforce/factory#229; not fixed here (different subsystem). + // + // The store override below forces markRoutedPrBabysitterRunning to fail + // WITHOUT any underlying claim-state change, which isolates the local + // bookkeeping fix under test but does not correspond to any reachable + // production path — see the "suppresses the retry..." test directly below + // for what a *real* markedRunning failure (another owner adopted the + // claim, or the lease genuinely expired) actually does on retry, which is + // materially different (the retry's own claim call detects the state + // change and either suppresses or mints a fresh claim, so it can never + // reach markRoutedPrBabysitterRunning on stale claim data). Do not read + // this test as evidence about what happens when a real claim is lost. + it('retries ensureBabysitter after a lost PR work claim instead of latching the spawn flag forever', async () => { + class LoseNextClaimStore extends InMemoryStateStore { + failNext = false + markRunningCalls = 0 + + override async markRoutedPrBabysitterRunning( + workspaceId: string, + identity: string, + owner: string, + claimId: string, + agentName: string, + nowMs: number, + ): Promise { + this.markRunningCalls += 1 + if (this.failNext) { + this.failNext = false + return false + } + return super.markRoutedPrBabysitterRunning(workspaceId, identity, owner, claimId, agentName, nowMs) + } + } + + const issue = realIssueFile(505, ready, { title: 'Real babysitter claim-lost retry' }) + const mount = new FakeMountClient({ [issuePath(505)]: issue }) + const fleet = new FakeFleetClient() + const stateStore = new LoseNextClaimStore({ batchSize: 2 }) + const factory = createFactory(babysitterConfig(), { + mount, + fleet, + triage: new StaticTriage(), + stateStore, + probePrResolver: async () => ({ repo: 'AgentWorkforce/pear', prNumber: 505 }), + }) + + await factory.start({ mode: 'live', liveSubscription: { transport: 'subscribe' } }) + try { + await factory.dispatch(await factory.triageIssue(parseLinearIssue(issuePath(505), issue))) + + stateStore.failNext = true + fleet.emitAgentExit('ar-505-impl-pear', 'worker_exited') + await vi.waitFor(() => expect(fleet.releases.map((release) => release.name)).toContain('ar-505-babysit')) + expect(stateStore.markRunningCalls).toBe(1) + + // Before the fix, #babysitterSpawned/#babysitterPr/#babysitterIssueRefs + // stayed set after the claim-lost release, so a later PR event's + // ensureBabysitter call short-circuited on the stale "already spawned" + // flag and never renegotiated the claim at all — markRunningCalls would + // stay latched at 1 forever, even though the babysitter that owned + // that claim no longer exists. + // markRunningCalls alone is not sufficient to prove the retry took the + // fresh-spawn branch specifically: ensureBabysitter's adopt branch + // (reusing a *stale, released* tracked agent still sitting in + // record.agents) calls markRoutedPrBabysitterRunning too, so it would + // pass this same assertion by silently re-confirming a phantom agent + // as "the" babysitter instead of retrying for real. #readIssue is + // called only inside the fresh-spawn branch's try block -- the adopt + // branch never calls it -- so a second read of the issue file is a + // real discriminator between the two, unlike anything both branches + // call. + const readsBeforeRetry = mount.reads.filter((path) => path === issuePath(505)).length + + const prPath = '/github/repos/AgentWorkforce/pear/pulls/505/metadata.json' + mount.files.set(prPath, { content: { number: 505, state: 'open', head_ref: 'ar-505-fix', draft: false } }) + mount.emit(changeEvent(prPath, 'pr-505-retry')) + + await vi.waitFor(() => expect(stateStore.markRunningCalls).toBe(2)) + expect(mount.reads.filter((path) => path === issuePath(505)).length).toBeGreaterThan(readsBeforeRetry) + } finally { + await factory.stop() + } + }) + + it('releases the PR work-unit claim when an issue-created babysitter exits abnormally', async () => { + const issue = realIssueFile(507, ready, { title: 'Real babysitter abnormal exit releases claim' }) + const mount = new FakeMountClient({ [issuePath(507)]: issue }) + const fleet = new FakeFleetClient() + const stateStore = new InMemoryStateStore({ batchSize: 2 }) + const factory = createFactory(babysitterConfig(), { + mount, + fleet, + triage: new StaticTriage(), + stateStore, + probePrResolver: async () => ({ repo: 'AgentWorkforce/pear', prNumber: 507 }), + }) + + await factory.dispatch(await factory.triageIssue(parseLinearIssue(issuePath(507), issue))) + fleet.emitAgentExit('ar-507-impl-pear', 'worker_exited') + await vi.waitFor(() => expect(fleet.spawns.map((spawn) => spawn.name)).toContain('ar-507-babysit')) + + const identity = 'agentworkforce/pear#507' + await vi.waitFor(async () => { + const claim = (await stateStore.listRoutedPrBabysitterClaims('factory-test')) + .find(([key]) => key === identity)?.[1] + expect(claim?.status).toBe('running') + }) + + // A crash/non-completion exit, unlike a normal 'issue-done' completion. + fleet.emitAgentExit('ar-507-babysit', 'worker_exited') + + // Before the fix, only the routed-PR exit path released the claim on a + // non-completion reason (mirroring isCompletionReason(reason) ? complete + // : release); the issue-created path here only ever called + // completeRoutedPrBabysitter, gated behind isCompletionReason, and did + // nothing at all on a non-completion exit — leaving the claim marked + // 'running' for a process that no longer exists until its lease expires + // naturally, blocking retry for that whole window. + await vi.waitFor(async () => { + const claim = (await stateStore.listRoutedPrBabysitterClaims('factory-test')) + .find(([key]) => key === identity)?.[1] + expect(claim).toBeUndefined() + }) + }) + + it('retries after a late failure in ensureBabysitter\'s fresh-spawn attempt, not just after a lost claim', async () => { + // Uncited by any reviewer -- found while fixing 3741674681 (the two + // markedRunning-failure branches) by noticing this pre-existing catch + // block hand-rolled the same three-key cleanup, missing #babysitterReady, + // and (like the markedRunning branches before their fix) never dropped + // the just-spawned agent from record.agents. Same class, a third + // pre-existing instance, found by reading code we were already touching + // rather than by a new review pass. + class FailNextPersistStore extends InMemoryStateStore { + failNext = false + markRunningCalls = 0 + + override async setBabysitterSession( + workspaceId: string, + key: string, + session: Parameters[2], + ): Promise { + if (this.failNext) { + this.failNext = false + throw new Error('durable session write unavailable') + } + await super.setBabysitterSession(workspaceId, key, session) + } + + override async markRoutedPrBabysitterRunning( + ...args: Parameters + ): ReturnType { + this.markRunningCalls += 1 + return super.markRoutedPrBabysitterRunning(...args) + } + } + + const issue = realIssueFile(508, ready, { title: 'Real babysitter late spawn failure retry' }) + const mount = new FakeMountClient({ [issuePath(508)]: issue }) + const fleet = new FakeFleetClient() + const stateStore = new FailNextPersistStore({ batchSize: 2 }) + const factory = createFactory(babysitterConfig(), { + mount, + fleet, + triage: new StaticTriage(), + stateStore, + probePrResolver: async () => ({ repo: 'AgentWorkforce/pear', prNumber: 508 }), + }) + + await factory.start({ mode: 'live', liveSubscription: { transport: 'subscribe' } }) + try { + await factory.dispatch(await factory.triageIssue(parseLinearIssue(issuePath(508), issue))) + + stateStore.failNext = true + fleet.emitAgentExit('ar-508-impl-pear', 'worker_exited') + await vi.waitFor(() => expect(factory.status().counters.babysitterSpawnFailures).toBe(1)) + // The failure happened before markRoutedPrBabysitterRunning was ever + // reached (setBabysitterSession is called first in the try block). + expect(stateStore.markRunningCalls).toBe(0) + + // markRunningCalls alone can't tell fresh-spawn apart from adopt (see + // the sibling test above) -- the adopt branch, reusing the phantom + // agent this catch block left behind pre-fix, also calls + // markRoutedPrBabysitterRunning. #readIssue is fresh-spawn-only, so a + // second read of the issue file is the real discriminator. + const readsBeforeRetry = mount.reads.filter((path) => path === issuePath(508)).length + + const prPath = '/github/repos/AgentWorkforce/pear/pulls/508/metadata.json' + mount.files.set(prPath, { content: { number: 508, state: 'open', head_ref: 'ar-508-fix', draft: false } }) + mount.emit(changeEvent(prPath, 'pr-508-retry')) + + // Before this fix, the catch block left #babysitterSpawned set and the + // spawned agent in record.agents, so this retry would have + // short-circuited on the stale flag (adopting the phantom record) and + // never reached markRoutedPrBabysitterRunning at all. + await vi.waitFor(() => expect(stateStore.markRunningCalls).toBe(1)) + expect(mount.reads.filter((path) => path === issuePath(508)).length).toBeGreaterThan(readsBeforeRetry) + } finally { + await factory.stop() + } + }) + + it('suppresses the retry instead of re-marking running when a lost claim was genuinely taken by another owner', async () => { + // Unlike the artificial-failure test above (which forces + // markRoutedPrBabysitterRunning to fail without changing any underlying + // state, to isolate the local-bookkeeping fix), this simulates the one + // realistic way that call actually fails: another owner adopts the claim + // via a live-lease transfer between our claim and our markRunning call — + // the same mechanism #restoreRoutedPrBabysitterClaims uses on restart. + class StolenByAnotherOwnerStore extends InMemoryStateStore { + stealNext = false + markRunningCalls = 0 + + override async markRoutedPrBabysitterRunning( + workspaceId: string, + identity: string, + owner: string, + claimId: string, + agentName: string, + nowMs: number, + ): Promise { + this.markRunningCalls += 1 + if (this.stealNext) { + this.stealNext = false + await this.adoptRoutedPrBabysitterClaim(workspaceId, identity, 'other-owner-agent', 'other-owner', nowMs, 15 * 60_000, true) + } + return super.markRoutedPrBabysitterRunning(workspaceId, identity, owner, claimId, agentName, nowMs) + } + } + + const issue = realIssueFile(506, ready, { title: 'Real babysitter claim genuinely stolen' }) + const mount = new FakeMountClient({ [issuePath(506)]: issue }) + const fleet = new FakeFleetClient() + const stateStore = new StolenByAnotherOwnerStore({ batchSize: 2 }) + const factory = createFactory(babysitterConfig(), { + mount, + fleet, + triage: new StaticTriage(), + stateStore, + probePrResolver: async () => ({ repo: 'AgentWorkforce/pear', prNumber: 506 }), + }) + + await factory.start({ mode: 'live', liveSubscription: { transport: 'subscribe' } }) + try { + await factory.dispatch(await factory.triageIssue(parseLinearIssue(issuePath(506), issue))) + + stateStore.stealNext = true + fleet.emitAgentExit('ar-506-impl-pear', 'worker_exited') + await vi.waitFor(() => expect(fleet.releases.map((release) => release.name)).toContain('ar-506-babysit')) + expect(stateStore.markRunningCalls).toBe(1) + + const prPath = '/github/repos/AgentWorkforce/pear/pulls/506/metadata.json' + mount.files.set(prPath, { content: { number: 506, state: 'open', head_ref: 'ar-506-fix', draft: false } }) + mount.emit(changeEvent(prPath, 'pr-506-retry')) + + await vi.waitFor(() => expect(factory.status().counters.babysitterOwnershipConflictsSuppressed).toBeGreaterThan(0)) + // The retry's own claimRoutedPrBabysitter call must see the other + // owner's live claim and suppress before ever reaching + // markRoutedPrBabysitterRunning again — proving the local-bookkeeping + // fix does not let a genuinely-taken claim be re-marked running out + // from under its real owner (no phantom-running claim). + expect(stateStore.markRunningCalls).toBe(1) + } finally { + await factory.stop() + } + }) + it('uses Relayfile by-id delivery claims for renamed PR activity, renews idempotently, and retires terminal claims after the local queue is durable', async () => { const number = 604 const issue = realIssueFile(number, ready, { title: 'Real durable resource babysitter' }) diff --git a/src/orchestrator/factory.ts b/src/orchestrator/factory.ts index 927ec21a..d3a5ecc6 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,10 @@ 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() + readonly #issueBabysitterClaims = new Map() + #routedPrBabysitterSweep?: Promise readonly #dispatchLifecycleEpochs = new Map() readonly #dispatchTerminalWaiters = new Map void>>() readonly #dispatchLifecycleRetryTimers = new Map>() @@ -804,6 +818,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 +942,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 +962,8 @@ export class FactoryLoop implements Factory { this.#babysitterSubscriptionOwners.clear() 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()) @@ -1246,6 +1264,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 +1554,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 +1689,243 @@ export class FactoryLoop implements Factory { return undefined } + async #restoreRoutedPrBabysitterClaims(): Promise { + 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)) + 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, + claimId: adopted.claimId, + }) + 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) + .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 + } + + async #runRoutedPrBabysitterSweep(dryRun: boolean): Promise { + const discovery = await discoverRoutedPullRequests(this.#mount, this.#config) + 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 = { + 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, + spawnFailures: 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 + } + if (claim.outcome !== 'claimed') continue + try { + 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 + 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 - stats.spawnFailures + break + } + this.#logger.info?.('[factory] routed PR babysitter sweep completed', stats) + } + + async #spawnRoutedPrBabysitter( + candidate: RoutedPrCandidate, + identity: string, + claimId: 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}`, + 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, claimId }) + this.#increment('routedPrBabysittersSpawned') + } + async runOnce(opts: { dryRun?: boolean } = {}): Promise { const dryRun = opts.dryRun ?? this.#config.dryRun const startedAtMs = this.#clock.now() @@ -1678,6 +1935,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 +5951,32 @@ export class FactoryLoop implements Factory { return } + const routedClaim = this.#routedPrBabysitterAgents.get(name) + if (routedClaim) { + this.#routedPrBabysitterAgents.delete(name) + if (isCompletionReason(reason)) { + await this.#state.completeRoutedPrBabysitter( + this.#workspaceId, + routedClaim.identity, + this.#routedPrBabysitterOwner, + routedClaim.claimId, + name, + this.#clock.now(), + ) + this.#increment('routedPrBabysittersCompleted') + } else { + await this.#state.releaseRoutedPrBabysitterClaim( + this.#workspaceId, + routedClaim.identity, + this.#routedPrBabysitterOwner, + routedClaim.claimId, + ) + this.#increment('routedPrBabysitterAbnormalExits') + } + 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 +6032,27 @@ 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) + const issueClaim = this.#issueBabysitterClaims.get(name) + if (exiting?.spec.role === 'babysitter' && issueClaim) { + if (isCompletionReason(reason)) { + await this.#state.completeRoutedPrBabysitter( + this.#workspaceId, + issueClaim.identity, + this.#dispatchLifecycleOwner, + issueClaim.claimId, + name, + this.#clock.now(), + ) + } else { + await this.#state.releaseRoutedPrBabysitterClaim( + this.#workspaceId, + issueClaim.identity, + this.#dispatchLifecycleOwner, + issueClaim.claimId, + ) + } + } + 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 }) @@ -10041,6 +10346,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 +10621,69 @@ export class FactoryLoop implements Factory { this.#increment('babysitterLifecycleOwnershipRejected') return } + const prWorkIdentity = routedPrIdentity(prRef.repo, prRef.prNumber) + let 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 === '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, + repo: prRef.repo, + prNumber: prRef.prNumber, + }) + return + } + const prWorkClaimId = prWorkClaim.claim.claimId const replacedSuperseded = prRef.authoritative ? await this.#retireSupersededBabysitters(record, prRef) : false @@ -10363,6 +10736,23 @@ export class FactoryLoop implements Factory { this.#babysitterSpawned.add(babysitterKey) const ref = this.#babysitterPr.get(babysitterKey)! await this.#persistBabysitterSession(record.issue, ref, tracked) + 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') + await this.#cancelBabysitterWake(babysitterKey) + return + } + this.#issueBabysitterClaims.set(tracked.result?.name ?? trackedName, { + identity: prWorkIdentity, + claimId: prWorkClaimId, + }) await this.#ensureBabysitterResourceSubscription(record.issue, ref, tracked) await this.#retargetSlackConversationToBabysitter(record) return @@ -10372,6 +10762,10 @@ export class FactoryLoop implements Factory { let finishSpawn!: () => void const spawnFinished = new Promise((resolve) => { finishSpawn = resolve }) this.#babysitterSpawnInFlight.set(babysitterKey, spawnFinished) + // Hoisted above the try so the catch block can drop the phantom + // record.agents entry if a later step in this attempt fails after the + // spawn itself succeeded (see the cancelBabysitterWake call in catch). + let spawnedAgentName: string | undefined try { const issue = await this.#readIssue(record.issue.path) @@ -10457,6 +10851,7 @@ export class FactoryLoop implements Factory { task, ownedPullRequest: { repo: prRef.repo, number: prRef.prNumber, path: prRef.path }, }, false) + spawnedAgentName = spawned.name const tracked = record.agents.get(spawned.name) this.#babysitterPr.set(babysitterKey, { repo: prRef.repo, @@ -10466,6 +10861,29 @@ export class FactoryLoop implements Factory { }) const ref = this.#babysitterPr.get(babysitterKey)! await this.#persistBabysitterSession(record.issue, ref, tracked) + 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') + // This agent was spawned solely for this claim attempt (unlike the + // adopt-branch case above, which reuses an agent that already owned + // this PR before the claim check ran) — drop it so a retry's + // trackedBabysitter search doesn't find a released process and skip + // straight to (incorrectly) re-adopting it instead of spawning fresh. + record.agents.delete(spawned.name) + await this.#cancelBabysitterWake(babysitterKey) + 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() @@ -10494,13 +10912,18 @@ export class FactoryLoop implements Factory { await this.#state.recordCritical(this.#workspaceId, ack.eventId, { issue: record.issue, input }) } } catch (error) { - // Allow a later event to retry the spawn. - this.#babysitterSpawned.delete(babysitterKey) - this.#babysitterPr.delete(babysitterKey) - this.#babysitterIssueRefs.delete(babysitterKey) - if (await this.#assertIssueDispatchLifecycleOwner(record.issue)) { - await this.#state.clearBabysitterSession(this.#workspaceId, babysitterKey) - } + // Allow a later event to retry the spawn. Same teardown as the + // markedRunning-failure branch above: #cancelBabysitterWake clears all + // four tracking keys (this hand-rolled version was missing + // #babysitterReady) and the durable session under the same + // "is this still ours to clear" guard this block already applied by + // hand. If the spawn itself succeeded before a later step in this try + // threw, drop the now-orphaned record.agents entry too -- otherwise a + // retry's trackedBabysitter search finds a released process and + // silently "adopts" it instead of spawning fresh (the same failure + // mode fixed for the markedRunning case above). + if (spawnedAgentName) record.agents.delete(spawnedAgentName) + await this.#cancelBabysitterWake(babysitterKey) this.#increment('babysitterSpawnFailures') this.#error(error, record.issue) } finally { diff --git a/src/ports/state.ts b/src/ports/state.ts index 78dbf98c..c30e44a4 100644 --- a/src/ports/state.ts +++ b/src/ports/state.ts @@ -91,6 +91,28 @@ export type BabysitterSessionState = { pendingDeliveryClaims?: Array<{ deliveryId: string; claimToken: string }> } +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. */ + revision: string + source: 'issue-created' | 'routed-open-prs' + status: 'claimed' | 'running' | 'complete' + owner: string + leaseUntilMs: number + claimedAtMs: number + updatedAtMs: number + 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 } + export type ConversationMessage = { id: string text: string @@ -380,6 +402,48 @@ 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, + claimId: string, + agentName: string, + nowMs: number, + ): Promise + adoptRoutedPrBabysitterClaim( + workspaceId: string, + identity: string, + agentName: string, + 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 + 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..e46d98fe 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, @@ -13,6 +14,8 @@ import type { GithubIssueCommentWatchState, ConversationMessage, ConversationSessionState, + RoutedPrBabysitterClaim, + RoutedPrBabysitterClaimResult, WaitingClarification, } from '../ports/state' import { InMemoryStateStore, type InMemoryStateStoreOptions } from './in-memory-state-store' @@ -22,12 +25,13 @@ type PersistedWorkspaceState = { githubIssueCommentWatches: Record waitingClarifications: Record babysitterSessions: Record + routedPrBabysitterClaims: Record conversationSessions: Record dispatchLifecycles: Record } type WatchStateDocument = { - version: 3 + version: 4 workspaces: Record } @@ -648,6 +652,171 @@ 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() + 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.leaseUntilMs <= nowMs) + ) delete workspace.routedPrBabysitterClaims[key] + } + const existing = workspace.routedPrBabysitterClaims[identity] + 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) + 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, + claimId: randomUUID(), + 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, + 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.claimId !== claimId || + claim.status === 'complete' || + claim.leaseUntilMs <= nowMs + ) return false + claim.status = 'running' + claim.agentName = agentName + claim.updatedAtMs = nowMs + await this.#persist(document) + return true + })) + } + + override async adoptRoutedPrBabysitterClaim( + workspaceId: string, + identity: string, + agentName: string, + owner: string, + nowMs: number, + leaseMs: number, + 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 === '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 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.owner !== owner || + claim.claimId !== claimId || + 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, + 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 || claim.claimId !== claimId) 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, @@ -822,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: {} } } } @@ -876,19 +1045,21 @@ 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') 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,11 +1069,12 @@ const parseDocument = (value: unknown): WatchStateDocument => { githubIssueCommentWatches: watches as Record, waitingClarifications: clarifications as Record, babysitterSessions: parseBabysitterSessions(babysitters ?? {}), + 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 = {} @@ -918,11 +1090,12 @@ const parseDocument = (value: unknown): WatchStateDocument => { githubIssueCommentWatches: watches as Record, waitingClarifications: clarifications as Record, babysitterSessions: parseBabysitterSessions(babysitters ?? {}), + routedPrBabysitterClaims: {}, conversationSessions: {}, dispatchLifecycles: {}, } } - return { version: 3, workspaces } + return { version: 4, workspaces } } if (value.version === 1) { const workspaces: Record = {} @@ -934,11 +1107,12 @@ const parseDocument = (value: unknown): WatchStateDocument => { githubIssueCommentWatches: watches as Record, waitingClarifications: {}, babysitterSessions: {}, + routedPrBabysitterClaims: {}, conversationSessions: {}, dispatchLifecycles: {}, } } - return { version: 3, workspaces } + return { version: 4, workspaces } } throw new Error('Factory GitHub watch state file is invalid') } @@ -952,6 +1126,39 @@ const cloneClarification = (record: WaitingClarification): WaitingClarification const cloneBabysitterSession = (session: BabysitterSessionState): BabysitterSessionState => structuredClone(session) +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 || + 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 Omit), + claimId: typeof candidate.claimId === 'string' + ? candidate.claimId + : `${candidate.owner}:${candidate.claimedAtMs}:${identity}`, + } + } + return claims +} + const cloneConversationSession = (session: ConversationSessionState): ConversationSessionState => structuredClone(session) @@ -1134,6 +1341,7 @@ const emptyWorkspaceState = (): PersistedWorkspaceState => ({ githubIssueCommentWatches: {}, waitingClarifications: {}, babysitterSessions: {}, + routedPrBabysitterClaims: {}, conversationSessions: {}, dispatchLifecycles: {}, }) @@ -1142,6 +1350,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..f5bfd9f4 100644 --- a/src/state/in-memory-state-store.ts +++ b/src/state/in-memory-state-store.ts @@ -1,5 +1,8 @@ +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' import type { BatchSnapshot, BabysitterSessionState, @@ -9,6 +12,8 @@ import type { DispatchAttemptState, GithubIssueCommentWatchState, RegistryHandoffAgent, + RoutedPrBabysitterClaim, + RoutedPrBabysitterClaimResult, ConversationMessage, ConversationSessionState, StateStore, @@ -31,6 +36,7 @@ type WorkspaceState = { canonicalIssueStates: Map dispatchFailureReaperHandoffs: Map babysitterSessions: Map + routedPrBabysitterClaims: Map dispatchLifecycles: Map } @@ -633,6 +639,145 @@ 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 + for (const [key, claim] of claims) { + if ( + (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) + 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) } + } + 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, + claimId: randomUUID(), + 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, + claimId: string, + agentName: string, + nowMs: number, + ): Promise { + const claim = this.#workspace(workspaceId).routedPrBabysitterClaims.get(identity) + 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 + return true + } + + async adoptRoutedPrBabysitterClaim( + workspaceId: string, + identity: string, + agentName: string, + owner: string, + nowMs: number, + leaseMs: number, + allowLiveLeaseTransfer = false, + ): Promise { + const claim = this.#workspace(workspaceId).routedPrBabysitterClaims.get(identity) + if ( + !claim || + 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 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.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, + claimId: string, + ): Promise { + const claims = this.#workspace(workspaceId).routedPrBabysitterClaims + const claim = claims.get(identity) + if (!claim || claim.owner !== owner || claim.claimId !== claimId) 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 +803,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..2d8a5f39 --- /dev/null +++ b/src/state/routed-pr-babysitter-claim.test.ts @@ -0,0 +1,340 @@ +import { mkdtemp } from 'node:fs/promises' +import { tmpdir } from 'node:os' +import { join } from 'node:path' + +import { describe, expect, it } from 'vitest' + +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' + +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', + first.claim!.claimId, + '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', + 'owner-a', + first.claim!.claimId, + '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') + }) + + 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.listRoutedPrBabysitterClaims('workspace')) + .map(([identity]) => identity)).toEqual(['agentworkforce/pear#8']) + 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} 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() + const original = await store.claimRoutedPrBabysitter( + 'workspace', + 'agentworkforce/pear#7', + seed, + 'old-owner', + 1_000, + 60_000, + 5, + ) + await store.markRoutedPrBabysitterRunning( + 'workspace', + 'agentworkforce/pear#7', + 'old-owner', + original.claim!.claimId, + 'agent-a', + 1_001, + ) + const adopted = await store.adoptRoutedPrBabysitterClaim( + 'workspace', + 'agentworkforce/pear#7', + 'agent-a', + 'new-owner', + 61_001, + 60_000, + ) + expect(adopted).toMatchObject({ owner: 'new-owner' }) + expect((await store.listRoutedPrBabysitterClaims('workspace'))[0]?.[1]).toMatchObject({ + owner: 'new-owner', + leaseUntilMs: 121_001, + agentName: 'agent-a', + status: 'running', + }) + expect(await store.releaseRoutedPrBabysitterClaim( + '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() + const original = await store.claimRoutedPrBabysitter( + 'workspace', + 'agentworkforce/pear#7', + seed, + 'old-owner', + 1_000, + 60_000, + 5, + ) + await store.markRoutedPrBabysitterRunning( + 'workspace', + 'agentworkforce/pear#7', + 'old-owner', + original.claim!.claimId, + 'agent-a', + 1_001, + ) + + expect(await store.adoptRoutedPrBabysitterClaim( + 'workspace', + 'agentworkforce/pear#7', + 'agent-a', + 'new-owner', + 2_000, + 60_000, + )).toBeUndefined() + 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() + const original = await store.claimRoutedPrBabysitter( + 'workspace', + 'agentworkforce/pear#7', + seed, + 'owner-a', + 1_000, + 60_000, + 5, + ) + await store.markRoutedPrBabysitterRunning( + 'workspace', + 'agentworkforce/pear#7', + 'owner-a', + original.claim!.claimId, + 'agent-a', + 1_001, + ) + + expect(await store.adoptRoutedPrBabysitterClaim( + 'workspace', + 'agentworkforce/pear#7', + 'agent-a', + 'owner-a', + 2_000, + 60_000, + )).toMatchObject({ owner: 'owner-a', claimId: original.claim!.claimId }) + 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() + const original = await store.claimRoutedPrBabysitter( + 'workspace', + 'agentworkforce/pear#7', + seed, + 'owner-a', + 1_000, + 60_000, + 5, + ) + await store.markRoutedPrBabysitterRunning( + '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, + ) + 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', () => { + 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') }) + }) +})