diff --git a/.github/workflows/ci.yml b/.github/workflows/ci.yml index 9517ca6..4dd1c70 100644 --- a/.github/workflows/ci.yml +++ b/.github/workflows/ci.yml @@ -25,7 +25,9 @@ jobs: - uses: actions/setup-node@v4 with: node-version: 20 - cache: npm - - run: npm ci - - run: npm run check - - run: npm run pack:dry + - uses: oven-sh/setup-bun@v2 + with: + bun-version: 1.3.7 + - run: bun install --no-save --ignore-scripts + - run: bun run check + - run: bun run pack:dry diff --git a/CHANGELOG.md b/CHANGELOG.md index 6d70e34..75d2cd5 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -10,7 +10,9 @@ ### Added -- share documentation workflows +- add public Herdr fleet workflow +- add plugin packaging and question TUI (#26) [#26] +- share documentation workflows (#30) [#30] - add skill-inspector plugin + verdict-aware renderer - Add skill-inspector skill: skillspector wrapper with readable reports (#28) [#28] - add ship-issue Claude Code plugin (#27) [#27] @@ -28,6 +30,8 @@ ### Changed +- bump ws from 8.20.0 to 8.21.1 in the npm_and_yarn group across 1 directory (#31) [#31] +- bump undici in the npm_and_yarn group across 1 directory (#29) [#29] - bump shell-quote (#16) [#16] - bump the npm_and_yarn group across 2 directories with 1 update (#12) [#12] - Port claude-md-improver skill to Claude Code plugin (#13) [#13] @@ -36,6 +40,10 @@ - bump the npm_and_yarn group across 1 directory with 2 updates (#5) [#5] - Initial commit +### Fixed + +- harden Herdr fleet orchestration + ### Security - bump protobufjs to 7.5.6 (#6) [#6] diff --git a/docs/catalog.md b/docs/catalog.md index 23ddcfc..b76110d 100644 --- a/docs/catalog.md +++ b/docs/catalog.md @@ -30,6 +30,7 @@ Keep this as the human-readable record of what lives in the package. | `grill-with-docs` | Skill | `skills/grill-with-docs/SKILL.md` | active | Stress-tests plans against project domain language, updates `CONTEXT.md`/ADRs as decisions crystallize, and uses recommendation-first `AskUserQuestion` decision prompts. | | `harness-audit` | Skill | `skills/harness-audit/SKILL.md` | active | Audits repo harness readiness and fix gaps; copy into harness-specific inventories for daily use when isolation matters. | | `harness-worktrees` | Skill | `skills/harness-worktrees/SKILL.md` | active | Manages Pi/Superconductor worktree refreshes and resets after PR merges. | +| `herdr-fleet` | Skill | `skills/herdr-fleet/SKILL.md` | active | Orchestrates user-confirmed, project-scoped Herdr worker rosters from one control pane. | | `scaffold-notes` | Skill | `skills/scaffold-notes/SKILL.md` | active | Maintenance skill for adding resources to this repo consistently. | | `skill-inspector` | Skill | `skills/skill-inspector/SKILL.md` | active | Global-first skill (symlinked into `~/.claude/skills/skill-inspector`); security-scans agent skills with the `skillspector` CLI and renders a plain-English verdict report (safe/caution/do-not-install, threat breakdown, top findings) for chat. | | `critical-bug-hunt.prompt` | Prompt | `prompts/critical-bug-hunt.prompt.md` | active | Recent-commit audit prompt for high-severity correctness bugs and minimal fixes. | diff --git a/package.json b/package.json index 7f4bf15..5a7fab9 100644 --- a/package.json +++ b/package.json @@ -42,20 +42,21 @@ ] }, "scripts": { - "check": "npm run lint && npm run typecheck && npm test && npm run validate:plugins && node scripts/validate-package-contents.mjs && npm run validate:skills && node scripts/list-resources.mjs", - "list": "node scripts/list-resources.mjs", - "validate:plugins": "node scripts/validate-plugin-ports.mjs && node scripts/validate-codex-plugins.mjs", + "check": "bun run lint && bun run typecheck && bun run test && bun run validate:plugins && bun scripts/validate-package-contents.mjs && bun run validate:skills && bun scripts/list-resources.mjs", + "list": "bun scripts/list-resources.mjs", + "validate:plugins": "bun scripts/validate-plugin-ports.mjs && bun scripts/validate-codex-plugins.mjs", "install:bash-guard:global": "node scripts/install-bash-guard-global.mjs", "new:extension": "node scripts/new-extension.mjs", "new:skill": "node scripts/new-skill.mjs", - "pack:dry": "npm pack --dry-run", + "pack:dry": "bun pm pack --dry-run", "format": "biome format --write .", "lint": "biome lint .", "lint:fix": "biome check --write .", "prepare": "husky", - "test": "node extensions/autopilot/v2-smoke-test.mjs && node extensions/question/question-smoke-test.mjs && node extensions/conditional-hooks/smoke-test.mjs && node scripts/lib/frontmatter-test.mjs && node scripts/lib/bundle-refs-test.mjs && node scripts/cli-entrypoint-test.mjs && python3 skills/diataxis-docs-site/tests/test_create_site.py", + "test": "node extensions/autopilot/v2-smoke-test.mjs && node extensions/question/question-smoke-test.mjs && node extensions/conditional-hooks/smoke-test.mjs && node scripts/lib/frontmatter-test.mjs && node scripts/lib/bundle-refs-test.mjs && node scripts/cli-entrypoint-test.mjs && python3 skills/diataxis-docs-site/tests/test_create_site.py && bun test scripts/lib/package-links.test.mjs && bun run test:herdr-fleet", + "test:herdr-fleet": "bun skills/herdr-fleet/scripts/resolve-project-key.mjs --self-test && bun skills/herdr-fleet/scripts/watch-fleet.mjs --self-test && bun skills/herdr-fleet/scripts/consume-events.mjs --self-test && bun test skills/herdr-fleet/scripts/fleet-state.test.mjs skills/herdr-fleet/scripts/review-thread-gate.test.mjs", "typecheck": "tsc --noEmit", - "validate:skills": "node scripts/validate-agent-skills.mjs" + "validate:skills": "bun scripts/validate-agent-skills.mjs" }, "peerDependencies": { "@mariozechner/pi-ai": "*", diff --git a/scripts/lib/package-links.mjs b/scripts/lib/package-links.mjs new file mode 100644 index 0000000..503f4f2 --- /dev/null +++ b/scripts/lib/package-links.mjs @@ -0,0 +1,38 @@ +import path from "node:path"; + +function headingAnchors(content) { + return new Set( + content + .split("\n") + .map((line) => line.match(/^#{1,6}\s+(.+)$/)?.[1]) + .filter(Boolean) + .map((heading) => + heading + .toLowerCase() + .replace(/[`*_~]/g, "") + .replace(/[^\p{L}\p{N}\s-]/gu, "") + .trim() + .replace(/\s+/g, "-"), + ), + ); +} + +export function findBrokenPackedLinks({ sourcePath, content, packedPathSet, readTarget }) { + const broken = []; + for (const match of content.matchAll(/\[[^\]]+\]\(([^)]+)\)/g)) { + const target = match[1].trim().split(/\s+/, 1)[0]; + if (/^[a-z][a-z0-9+.-]*:/i.test(target)) continue; + const [relativeTarget, anchor] = target.split("#", 2); + const targetPath = relativeTarget + ? path.posix.normalize(path.posix.join(path.posix.dirname(sourcePath), relativeTarget)) + : sourcePath; + if (!packedPathSet.has(targetPath)) { + broken.push(`${sourcePath} -> ${target}`); + continue; + } + if (anchor && !headingAnchors(readTarget(targetPath)).has(anchor)) { + broken.push(`${sourcePath} -> #${anchor}`); + } + } + return broken; +} diff --git a/scripts/lib/package-links.test.mjs b/scripts/lib/package-links.test.mjs new file mode 100644 index 0000000..052ab42 --- /dev/null +++ b/scripts/lib/package-links.test.mjs @@ -0,0 +1,45 @@ +import { test } from "bun:test"; +import assert from "node:assert/strict"; +import { findBrokenPackedLinks } from "./package-links.mjs"; + +const readTarget = () => "# Included target\n"; + +test("checkout-only targets fail packed-link validation", () => { + const broken = findBrokenPackedLinks({ + sourcePath: "skills/herdr-fleet/README.md", + content: "[private](private.md)", + packedPathSet: new Set(["skills/herdr-fleet/README.md"]), + readTarget, + }); + assert.deepEqual(broken, ["skills/herdr-fleet/README.md -> private.md"]); +}); + +test("packed targets with valid anchors pass", () => { + const broken = findBrokenPackedLinks({ + sourcePath: "skills/herdr-fleet/README.md", + content: "[included](included.md#included-target)", + packedPathSet: new Set(["skills/herdr-fleet/README.md", "skills/herdr-fleet/included.md"]), + readTarget, + }); + assert.deepEqual(broken, []); +}); + +test("link destinations ignore surrounding whitespace and optional titles", () => { + const broken = findBrokenPackedLinks({ + sourcePath: "skills/herdr-fleet/README.md", + content: '[included]( included.md#included-target "Title" ) [external]( https://example.com "Title" )', + packedPathSet: new Set(["skills/herdr-fleet/README.md", "skills/herdr-fleet/included.md"]), + readTarget, + }); + assert.deepEqual(broken, []); +}); + +test("whitespace-prefixed missing targets remain broken after title removal", () => { + const broken = findBrokenPackedLinks({ + sourcePath: "skills/herdr-fleet/README.md", + content: '[missing]( missing.md "Title" )', + packedPathSet: new Set(["skills/herdr-fleet/README.md"]), + readTarget, + }); + assert.deepEqual(broken, ["skills/herdr-fleet/README.md -> missing.md"]); +}); diff --git a/scripts/validate-package-contents.mjs b/scripts/validate-package-contents.mjs index bf0e53d..6267be9 100644 --- a/scripts/validate-package-contents.mjs +++ b/scripts/validate-package-contents.mjs @@ -1,17 +1,24 @@ import { spawnSync } from "node:child_process"; +import { readdirSync, readFileSync } from "node:fs"; import path from "node:path"; import { fileURLToPath } from "node:url"; +import { findBrokenPackedLinks } from "./lib/package-links.mjs"; const root = path.resolve(path.dirname(fileURLToPath(import.meta.url)), ".."); -const result = spawnSync("npm", ["pack", "--dry-run", "--json", "--ignore-scripts"], { cwd: root, encoding: "utf8" }); +const result = spawnSync("bun", ["pm", "pack", "--dry-run", "--ignore-scripts"], { + cwd: root, + encoding: "utf8", +}); if (result.status !== 0) { process.stderr.write(result.stderr || result.stdout); process.exit(result.status ?? 1); } -const [pack] = JSON.parse(result.stdout); -const packedPaths = pack.files.map((file) => file.path); +const packedPaths = result.stdout + .split("\n") + .map((line) => line.match(/^packed\s+\S+\s+(.+)$/)?.[1]) + .filter(Boolean); const packedPathSet = new Set(packedPaths); const requiredPaths = [ "skills/diataxis-docs-site/SKILL.md", @@ -20,16 +27,62 @@ const requiredPaths = [ "skills/diataxis-docs-site/scripts/create_site.py", "skills/github-wiki/SKILL.md", "skills/github-wiki/agents/openai.yaml", + "skills/herdr-fleet/SKILL.md", + "skills/herdr-fleet/README.md", + "skills/herdr-fleet/launch-fleet.md", + "skills/herdr-fleet/protocols.md", + "skills/herdr-fleet/fixtures/compaction-failures.json", + "skills/herdr-fleet/fixtures/fleet-state-race.json", + "skills/herdr-fleet/fixtures/project-key-collision.json", + "skills/herdr-fleet/fixtures/review-thread-pages.json", + "skills/herdr-fleet/fixtures/watcher-identity.json", + "skills/herdr-fleet/scripts/consume-events.mjs", + "skills/herdr-fleet/scripts/fleet-labels.mjs", + "skills/herdr-fleet/scripts/fleet-state.mjs", + "skills/herdr-fleet/scripts/fleet-state.test.mjs", + "skills/herdr-fleet/scripts/resolve-project-key.mjs", + "skills/herdr-fleet/scripts/review-thread-gate.mjs", + "skills/herdr-fleet/scripts/review-thread-gate.test.mjs", + "skills/herdr-fleet/scripts/watch-fleet.mjs", + "scripts/lib/package-links.mjs", + "scripts/lib/package-links.test.mjs", ]; const missing = requiredPaths.filter((requiredPath) => !packedPathSet.has(requiredPath)); const forbidden = packedPaths.filter( (packedPath) => /(?:^|\/)__pycache__(?:\/|$)/.test(packedPath) || /\.py[cod]$/.test(packedPath), ); -if (missing.length > 0 || forbidden.length > 0) { - if (missing.length > 0) console.error(`Missing required package files:\n${missing.join("\n")}`); - if (forbidden.length > 0) console.error(`Forbidden generated artifacts in package:\n${forbidden.join("\n")}`); +function markdownFiles(directory) { + return readdirSync(directory, { withFileTypes: true }).flatMap((entry) => { + const entryPath = path.join(directory, entry.name); + if (entry.isDirectory()) return markdownFiles(entryPath); + return entry.isFile() && entry.name.endsWith(".md") ? [entryPath] : []; + }); +} + +const brokenLinks = []; +for (const markdownPath of markdownFiles(path.join(root, "skills/herdr-fleet"))) { + const sourcePath = path.relative(root, markdownPath).split(path.sep).join(path.posix.sep); + if (!packedPathSet.has(sourcePath)) { + brokenLinks.push(`${sourcePath} is not packed`); + continue; + } + brokenLinks.push( + ...findBrokenPackedLinks({ + sourcePath, + content: readFileSync(markdownPath, "utf8"), + packedPathSet, + readTarget: (targetPath) => readFileSync(path.join(root, targetPath), "utf8"), + }), + ); +} + +if (missing.length > 0 || forbidden.length > 0 || brokenLinks.length > 0) { + if (missing.length > 0) process.stderr.write(`Missing required package files:\n${missing.join("\n")}\n`); + if (forbidden.length > 0) + process.stderr.write(`Forbidden generated artifacts in package:\n${forbidden.join("\n")}\n`); + if (brokenLinks.length > 0) process.stderr.write(`Broken herdr-fleet links:\n${brokenLinks.join("\n")}\n`); process.exit(1); } -console.log("✓ required skill package files are included"); +process.stdout.write("✓ package contents and herdr-fleet links are valid\n"); diff --git a/skills/README.md b/skills/README.md index c6636f8..14c2f42 100644 --- a/skills/README.md +++ b/skills/README.md @@ -19,11 +19,12 @@ This directory is the repo's generic skills lane. The skills CLI also discovers | `grill-with-docs` | [`grill-with-docs/`](grill-with-docs/) | Stress-tests plans against project domain language and records resolved terms/ADRs as decisions crystallize. | Generic Agent Skill. | | `harness-audit` | [`harness-audit/`](harness-audit/) | Audits repos for autonomous-agent harness readiness and unattended ticket execution gaps. | Generic Agent Skill; also available as a Claude Code plugin. | | `harness-worktrees` | [`harness-worktrees/`](harness-worktrees/) | Manages Pi/Superconductor worktree refresh and reset workflows after PR merges. | Generic Agent Skill; also available as a Claude Code plugin. | +| `herdr-fleet` | [`herdr-fleet/`](herdr-fleet/) | Launches and reconciles user-confirmed, project-scoped Herdr worker fleets from one control pane. | Requires `HERDR_ENV=1`; defaults to report-only merge policy. | | `scaffold-notes` | [`scaffold-notes/`](scaffold-notes/) | Maintains this repo's Pi package resources and docs when adding or refactoring skills/extensions/prompts/themes. | Repo maintenance skill. | ## Validate ```bash -npm run validate:skills -npx skills add . --list +bun run validate:skills +bunx skills add . --list ``` diff --git a/skills/herdr-fleet/README.md b/skills/herdr-fleet/README.md new file mode 100644 index 0000000..800ec4c --- /dev/null +++ b/skills/herdr-fleet/README.md @@ -0,0 +1,36 @@ +# Herdr Fleet + +`herdr-fleet` helps one control pane launch or rebuild a project-scoped Herdr fleet. It asks the user for the roster, previews the pane map, reuses ownership-proven workers, and creates only confirmed missing panes. There is no default worker count or fixed worker roster. + +See [SKILL.md](SKILL.md) for the agent contract, [launch-fleet.md](launch-fleet.md) for intake and reconciliation, and [protocols.md](protocols.md) for the live event loop. + +## Roster intake + +Each worker is user-selected: + +- label; +- launch command; +- role (`implementer`, `reviewer`, or another role); +- optional assignment or lane constraint; +- pane placement. + +The current control pane is fixed and is not included in the worker count. Launch and rebuild require an `AskUserQuestion` roster preview and confirmation before worker-pane mutation. + +## Launcher menu + +These are launcher options, not a default roster: + +| Command | Routed model | Effort | +| --- | --- | --- | +| `claudex` | GPT-5.6 Sol | xhigh | +| `claudex --model fable` | GPT-5.6 Sol | xhigh | +| `claudex --model opus` | GPT-5.6 Sol | high | +| `claudex --model sonnet` | Grok 4.5 | high | +| `claudex --model haiku` | GPT-5.6 Terra | medium | +| `claude` | Native Claude Code | native model selection | + +The roster can also use `pi`, `codex`, or any user-provided command. + +Claudex keeps the Claude Code harness—its tools, permissions, skills, hooks, context management, and interface—while routing model traffic through CLIProxyAPI. Ossie's setup article, [The Fable Effect](https://aojdevstudio.me/blog/the-fable-effect/#how-i-ran-gpt-56-sol-inside-the-claude-code-harness), is the canonical setup reference. The article is currently a draft, so the URL may return 404 until publication. + +This package intentionally does not duplicate proxy configuration, credentials, local paths, or setup secrets. diff --git a/skills/herdr-fleet/SKILL.md b/skills/herdr-fleet/SKILL.md new file mode 100644 index 0000000..72f1ecd --- /dev/null +++ b/skills/herdr-fleet/SKILL.md @@ -0,0 +1,48 @@ +--- +name: herdr-fleet +description: This skill should be used when a project requires Herdr fleet launch, worker-pane startup, surviving-pane rebuilds, or control of a project-scoped issue and pull-request queue. Requires HERDR_ENV=1. +--- + +# Herdr Fleet + +Act as the fleet's control pane. The principal talks to this pane; the control pane dispatches work, consumes review verdicts, and reports progress. Workers implement or review. Only the control pane operates Herdr. Merge authority is established explicitly during launch and defaults to report-only. + +Choose the active branch: + +- **Cold start** — no fleet exists or panes were lost: follow [launch-fleet.md](launch-fleet.md), then enter the live loop. +- **Live loop** — the fleet exists: follow [protocols.md](protocols.md). + +## Standing rules + +- Derive a collision-checked control label from the current repository as `-control-pane`; never copy a label from another project. [Launch Step 2](launch-fleet.md#step-2-resolve-the-project-identity) defines the derivation. +- Scope every inventory, watcher, split, discovery, and event action to both the injected workspace and tab IDs. +- Reconcile surviving project-owned panes before creating anything. Reuse healthy panes, retire only proven stale owned panes, and create only missing roles. +- Reserve Herdr remote control for the control pane. Broadcast this boundary before assigning work. +- Default to report-only. Merge only when explicit launch-time authorization covers the repository, base branch, and strategy, and a fresh verdict names the current head with green required checks. If policy or tooling requires approval, ask the principal directly; routing it through a worker is permission laundering. +- Delegate parallel work and retain only findings in the control context. Read worker transcripts with `herdr pane read`; dispatch with `herdr pane run`. +- Report events in proportion to their importance. Use the environment's notification mechanism for unattended merges, blockers, and decisions. Ask the principal only for genuine product, scope, cost, or irreversible decisions, with a recommendation first. +- Resolve reversible, PR-gated worker questions in the control pane. Verify every claim that the principal "approved" a choice. + +## User-selected roster + +The current control pane is the only fixed pane. Before creating workers, follow the `AskUserQuestion` intake and confirmation in [launch-fleet.md](launch-fleet.md#step-3-collect-and-confirm-the-roster). Accept any worker count, labels, commands, roles, assignments, and confirmed pane map. Rebuilds reuse only ownership-proven roster metadata. + +Read the human [launcher menu](README.md#launcher-menu) before presenting command choices. Claudex, native Claude Code, Pi, Codex, and arbitrary user-provided commands are options, never roster defaults. + +## Operational gotchas + +1. **Dialogs can swallow dispatches.** A message sent while a worker has a blocking permission or question dialog can become the dialog's answer. After dispatching to a blocked pane, read the transcript and confirm the message became a prompt. Audit any resulting approval claim. +2. **Hooks parse dispatch command text.** Repository hooks may regex-match the shell command containing `herdr pane run`, including words inside the worker message. When a hook trips, use its documented satisfaction path in separate commands rather than bypass environment variables. +3. **Capacity failures need bounded failover.** Retry a model-capacity error once. On a second failure, hand the task to another compatible worker and identify any orphaned review claim that must be superseded. +4. **Review claims race.** Two reviewers can claim after checking the same head. Duplicate reviews are acceptable, but only a review newer than the pull-request head is fresh. Explicitly override orphaned claims. +5. **Verdicts expire on every head change.** Content pushes require full review. Mechanical base syncs may use a focused delta review, but that review must issue a new verdict naming the post-sync SHA; source-file resolutions receive semantic scrutiny. +6. **Union merge drivers do not apply on GitHub.** A shared append-only file can make sibling pull requests dirty after every merge. Merge the base branch locally in each branch worktree, resolve with the repository's intended union behavior, and push. +7. **Context wedges masquerade as work.** A full context and a large nested-agent swarm can leave a pane reporting `working` while its worktree is frozen. Prevent this with incremental commits, bounded delegation, and compaction near 75%. Confirm liveness through worktree modification times. Salvage verified work, retire the pane, and start fresh. +8. **Long-lived sessions drift.** Recycle workers that accumulate very large contexts, use stale directories, repeat obsolete recaps, or leave dispatches unsubmitted. A fresh pane with a self-contained brief is safer. +9. **Runner starvation resembles test failure.** Classify failures before rerunning. A local pass, the same failure on unrelated runs, and high concurrent load point to infrastructure. Repeated failures of the same class warrant a structural CI fix. +10. **Workers exceed roles unless corrected.** Reviewers review; implementers implement; the control pane operates Herdr and applies the recorded merge policy. Watch queued intents and correct role drift before execution. +11. **Background deliverables may duplicate or disappear.** Request missing output explicitly. Act only on delivered content and deduplicate retries. +12. **Nested agents can finish files but fail to report.** If a worker waits on a silent child, inspect output modification times. When writes have stopped, tell the worker to verify disk state and finish inline. +13. **Long dispatches may paste without submitting.** Confirm every dispatch changes the pane to `working`. If a paste placeholder remains at the prompt, submit it with an empty `herdr pane run ""`, then verify again. +14. **Repository policy is runtime input.** Before assigning work, derive lane labels, branch prefixes, provenance variables, hook requirements, issue-link rules, and worktree cleanup commands from the repository's checked-in instructions. Include only the applicable rules in each brief. +15. **Context footer formats vary.** The scoped watcher recognizes forms such as `[79% ...]`, `90% context used`, and `9.3%/372k`. It re-arms compaction after usage drops below 60%; a permanent "already compacted" set would disable protection for the rest of the session. diff --git a/skills/herdr-fleet/fixtures/compaction-failures.json b/skills/herdr-fleet/fixtures/compaction-failures.json new file mode 100644 index 0000000..04f46ab --- /dev/null +++ b/skills/herdr-fleet/fixtures/compaction-failures.json @@ -0,0 +1,35 @@ +{ + "maxAttempts": 3, + "retryMs": 60000, + "cases": [ + { + "name": "failed dispatches stop at the configured limit", + "samples": [80, 80, 80, 80], + "agentStatuses": ["idle", "idle", "idle", "idle"], + "outcomes": ["fail", "fail", "fail", "fail"], + "expectedAttempts": 3, + "expectedBlockedEvents": 1 + }, + { + "name": "usage below the rearm threshold permits a new attempt", + "samples": [80, 80, 80, 50, 80], + "agentStatuses": ["idle", "idle", "idle", "idle", "idle"], + "outcomes": ["fail", "fail", "fail", "none", "fail"], + "expectedAttempts": 1, + "expectedBlockedEvents": 1 + }, + { + "name": "exhausted successful dispatches block before dialog deferral", + "initialState": { + "compactionAttempts": 3, + "lastCompactionRequestAt": 0, + "compactionBlocked": false + }, + "samples": [80], + "agentStatuses": ["blocked"], + "outcomes": ["none"], + "expectedAttempts": 3, + "expectedBlockedEvents": 1 + } + ] +} diff --git a/skills/herdr-fleet/fixtures/fleet-state-race.json b/skills/herdr-fleet/fixtures/fleet-state-race.json new file mode 100644 index 0000000..9c8f08b --- /dev/null +++ b/skills/herdr-fleet/fixtures/fleet-state-race.json @@ -0,0 +1,80 @@ +{ + "scope": { + "workspaceId": "w1", + "tabId": "w1:t1" + }, + "before": { + "result": { + "panes": [ + { + "pane_id": "w1:p1", + "workspace_id": "w1", + "tab_id": "w1:t1", + "label": "app-builder", + "foreground_cwd": "/repos/app", + "tokens": { + "fleet_key": "app", + "fleet_owner": "owner-1", + "fleet_kind": "worker", + "fleet_role": "implementer" + } + }, + { + "pane_id": "w1:p9", + "workspace_id": "w1", + "tab_id": "w1:t2", + "label": "other-worker" + } + ] + } + }, + "afterTopologyRace": { + "result": { + "panes": [ + { + "pane_id": "w1:p1", + "workspace_id": "w1", + "tab_id": "w1:t1", + "label": "app-reviewer", + "foreground_cwd": "/repos/app", + "tokens": { + "fleet_key": "app", + "fleet_owner": "owner-1", + "fleet_kind": "worker", + "fleet_role": "reviewer" + } + } + ] + } + }, + "rosterWithReservedLabel": [ + { + "label": "control-pane", + "launchCommand": "pi", + "role": "implementer", + "placement": "right" + } + ], + "metadataReadback": { + "pane": { + "label": "app-builder", + "tokens": { + "fleet_owner": "owner-1", + "fleet_key": "app", + "fleet_kind": "worker", + "fleet_label": "builder", + "fleet_role": "implementer", + "fleet_command": "claudex --model fa", + "fleet_placement": "right" + } + }, + "expected": { + "ownerToken": "owner-1", + "projectKey": "app", + "label": "builder", + "role": "implementer", + "command": "claudex --model fable", + "placement": "right" + } + } +} diff --git a/skills/herdr-fleet/fixtures/project-key-collision.json b/skills/herdr-fleet/fixtures/project-key-collision.json new file mode 100644 index 0000000..9598545 --- /dev/null +++ b/skills/herdr-fleet/fixtures/project-key-collision.json @@ -0,0 +1,35 @@ +{ + "baseKey": "ba", + "repoRoot": "/repos/billing-api", + "expectedKey": "ba", + "panes": [ + { + "pane_id": "w1:p1", + "workspace_id": "w1", + "tab_id": "w1:t1", + "label": "ba-control-pane", + "cwd": "/repos/billing-api" + }, + { + "pane_id": "w1:p2", + "workspace_id": "w1", + "tab_id": "w1:t1", + "label": "ba-pi-impl", + "cwd": "/repos/billing-api" + }, + { + "pane_id": "w1:p3", + "workspace_id": "w1", + "tab_id": "w1:t1", + "label": "ba-1234-control-pane", + "cwd": "/repos/other-service" + }, + { + "pane_id": "w1:p4", + "workspace_id": "w1", + "tab_id": "w1:t1", + "label": "ba-1234-codex-impl", + "cwd": "/repos/other-service" + } + ] +} diff --git a/skills/herdr-fleet/fixtures/review-thread-pages.json b/skills/herdr-fleet/fixtures/review-thread-pages.json new file mode 100644 index 0000000..54d5975 --- /dev/null +++ b/skills/herdr-fleet/fixtures/review-thread-pages.json @@ -0,0 +1,56 @@ +{ + "pages": [ + { + "data": { + "repository": { + "pullRequest": { + "headRefOid": "abc123", + "reviewThreads": { + "nodes": [ + { + "id": "thread-unresolved", + "isResolved": false, + "isOutdated": false, + "comments": { "nodes": [{ "url": "https://example.test/unresolved" }] } + }, + { + "id": "thread-resolved", + "isResolved": true, + "isOutdated": false, + "comments": { "nodes": [{ "url": "https://example.test/resolved" }] } + } + ], + "pageInfo": { "hasNextPage": true, "endCursor": "cursor-1" } + } + } + } + } + }, + { + "data": { + "repository": { + "pullRequest": { + "headRefOid": "abc123", + "reviewThreads": { + "nodes": [ + { + "id": "thread-outdated", + "isResolved": false, + "isOutdated": true, + "comments": { "nodes": [{ "url": "https://example.test/outdated" }] } + }, + { + "id": "thread-page-two-unresolved", + "isResolved": false, + "isOutdated": false, + "comments": { "nodes": [{ "url": "https://example.test/page-two" }] } + } + ], + "pageInfo": { "hasNextPage": false, "endCursor": null } + } + } + } + } + } + ] +} diff --git a/skills/herdr-fleet/fixtures/watcher-identity.json b/skills/herdr-fleet/fixtures/watcher-identity.json new file mode 100644 index 0000000..57911ec --- /dev/null +++ b/skills/herdr-fleet/fixtures/watcher-identity.json @@ -0,0 +1,28 @@ +{ + "cases": [ + { + "name": "exact watcher instance", + "pid": 4242, + "instanceId": "instance-1234", + "expectedCommandSuffix": "watch-fleet.mjs --project-key app --instance-token instance-1234", + "command": "bun /skill/watch-fleet.mjs --project-key app --instance-token instance-1234", + "expected": true + }, + { + "name": "recycled PID with unrelated command", + "pid": 4242, + "instanceId": "instance-1234", + "expectedCommandSuffix": "watch-fleet.mjs --project-key app --instance-token instance-1234", + "command": "sleep 9999", + "expected": false + }, + { + "name": "trailing argv does not match exact watcher scope", + "pid": 4242, + "instanceId": "instance-1234", + "expectedCommandSuffix": "watch-fleet.mjs --project-key app --instance-token instance-1234", + "command": "bun /skill/watch-fleet.mjs --project-key app --instance-token instance-1234-extra", + "expected": false + } + ] +} diff --git a/skills/herdr-fleet/launch-fleet.md b/skills/herdr-fleet/launch-fleet.md new file mode 100644 index 0000000..af2a687 --- /dev/null +++ b/skills/herdr-fleet/launch-fleet.md @@ -0,0 +1,363 @@ +# Launch or rebuild the fleet + +Build or reconcile a user-selected worker fleet inside the current Herdr tab. The current control pane is the only fixed pane and is not part of the worker count. Complete the steps in order. + +## Step 1: Verify the environment and scope + +```bash +test "${HERDR_ENV:-}" = 1 || exit 1 +test -n "${HERDR_WORKSPACE_ID:-}" || exit 1 +test -n "${HERDR_TAB_ID:-}" || exit 1 +test -n "${HERDR_PANE_ID:-}" || exit 1 +herdr pane +CURRENT_PANE_JSON="$(herdr pane current --current)" +printf '%s' "$CURRENT_PANE_JSON" | \ + EXPECTED_WORKSPACE="$HERDR_WORKSPACE_ID" EXPECTED_TAB="$HERDR_TAB_ID" EXPECTED_PANE="$HERDR_PANE_ID" \ + bun -e ' +const payload = JSON.parse(await Bun.stdin.text()); +const pane = payload?.result?.pane ?? payload?.pane ?? payload?.result ?? payload; +if ( + pane.workspace_id !== process.env.EXPECTED_WORKSPACE || + pane.tab_id !== process.env.EXPECTED_TAB || + pane.pane_id !== process.env.EXPECTED_PANE +) process.exit(1); +' +printf '%s\n' "$HERDR_WORKSPACE_ID" "$HERDR_TAB_ID" "$HERDR_PANE_ID" +``` + +The installed `herdr pane` output is the syntax authority. Stop before mutation if the session IDs are unavailable. + +`herdr pane list` can filter by workspace but not tab. Every inventory must filter returned records by both `workspace_id` and `tab_id`; never treat a workspace-only result as the fleet. + +## Step 2: Resolve the project identity + +Derive the base key from the current repository. Multiword names use each word's first letter; a single word uses its first three characters. + +```bash +REPO_ROOT="$(git rev-parse --show-toplevel 2>/dev/null)" || exit 1 +project="$(basename "$REPO_ROOT")" +BASE_KEY="$(printf '%s' "$project" | tr '[:upper:]' '[:lower:]' | awk -F'[-_ ]+' '{s=""; for (i=1;i<=NF;i++) s=s substr($i,1,1); if(length(s)<2)s=substr($1,1,3); print s}')" +ALL_PANES_JSON="$(herdr pane list --workspace "$HERDR_WORKSPACE_ID")" +TAB_PANES_JSON="$(printf '%s' "$ALL_PANES_JSON" | TAB_ID="$HERDR_TAB_ID" WORKSPACE_ID="$HERDR_WORKSPACE_ID" bun -e ' +const payload = JSON.parse(await Bun.stdin.text()); +const panes = payload?.result?.panes ?? payload?.panes ?? payload?.result ?? payload; +const scoped = (Array.isArray(panes) ? panes : []).filter( + (pane) => pane.workspace_id === process.env.WORKSPACE_ID && pane.tab_id === process.env.TAB_ID, +); +process.stdout.write(JSON.stringify(scoped)); +')" +KEY_RESULT="$(printf '%s' "$TAB_PANES_JSON" | bun /scripts/resolve-project-key.mjs \ + --base-key "$BASE_KEY" --repo-root "$REPO_ROOT" --json)" +PROJECT_KEY="$(printf '%s' "$KEY_RESULT" | bun -e 'process.stdout.write(JSON.parse(await Bun.stdin.text()).projectKey)')" +FLEET_OWNER_TOKEN="$(printf '%s' "$KEY_RESULT" | bun -e 'process.stdout.write(JSON.parse(await Bun.stdin.text()).ownerToken)')" +CONTROL_LABEL="${PROJECT_KEY}-control-pane" +RECONCILIATION_FINGERPRINT="$(printf '%s' "$TAB_PANES_JSON" | \ + bun /scripts/fleet-state.mjs --fingerprint \ + --workspace-id "$HERDR_WORKSPACE_ID" --tab-id "$HERDR_TAB_ID" | \ + bun -e 'process.stdout.write(JSON.parse(await Bun.stdin.text()).fingerprint)')" +``` + +The resolver prefers an established ownership-proven key. It compares metadata keys and legacy key/role labels by complete boundaries, so `ba-1234-pi-impl` occupies `ba-1234`, not `ba`. If the exact base key is foreign-occupied, it derives a deterministic suffix, rechecks that complete key, and lengthens the suffix until free. Multiple owned keys stop the launch for reconciliation. + +## Step 3: Collect and confirm the roster + +Read the human-facing [launcher menu](README.md#launcher-menu), then use `AskUserQuestion` before creating any worker pane. + +### Rebuild intake + +Use the current-tab list only for discovery. For every reuse candidate, immediately read `herdr pane get ` and classify the fresh pane record; never trust list tokens alone. + +Require this read-back metadata: + +- `tokens.fleet_owner` equals `$FLEET_OWNER_TOKEN`; +- `tokens.fleet_key` equals `$PROJECT_KEY`; +- `tokens.fleet_kind` is `worker`; +- repository cwd is owned; +- `fleet_label`, `fleet_role`, `fleet_command`, `fleet_placement`, and optional `fleet_assignment` are present; +- `fleet_metadata_sha` matches the canonical hash recomputed from all reconstructable fields, proving Step 7 stored the untruncated values; +- the exact stored command and placement reconstruct the prior roster, and `pane process-info` is consistent with the stored launcher. + +Map the fresh readback tokens to the expected worker shape and run `fleet-state.mjs --verify-metadata`; this recomputes `fleet_metadata_sha` and rejects truncated or normalized values. A mismatch makes the pane non-reusable. + +When every surviving worker has complete, consistent metadata, reconstruct the prior roster and use `AskUserQuestion` with **Reuse detected roster**, **Edit roster**, and **Cancel**. Include the full pane-map preview. Reuse only after confirmation. + +If any worker is ambiguous, metadata is incomplete, command evidence differs, or no owned roster exists, collect the roster again. + +### New or edited intake + +Use a free-form `AskUserQuestion` asking for one worker per line: + +```text +label | launch command | role | optional assignment/lane | desired placement +``` + +Requirements: + +- accept any number of workers, including zero; +- require a unique non-empty label and launch command per worker; +- reject `control-pane` and any rendered worker label equal to `$CONTROL_LABEL`; +- accept `implementer`, `reviewer`, or any user-specified role; +- preserve an optional assignment or lane constraint; +- accept `pi`, `codex`, every documented Claudex/Claude launcher, and arbitrary user commands; +- treat commands and assignments as data: never interpolate credentials or secrets into pane labels or metadata. + +Parse the answer into `ROSTER_JSON`, then validate final rendered labels before showing the preview: + +```bash +ROSTER_JSON="$(printf '%s' "$ROSTER_JSON" | bun /scripts/fleet-state.mjs \ + --validate-roster --project-key "$PROJECT_KEY")" || exit 1 +``` + +This rejects duplicate labels and any worker that would render as `$CONTROL_LABEL`. Derive the exact split/placement plan and render a preview containing the control pane plus every worker's label, command, role, assignment/lane, and placement. Use a second `AskUserQuestion` with **Confirm**, **Edit**, and **Cancel**. No pane mutation occurs before **Confirm**. + +Immediately after **Confirm**, refresh and compare the classified topology with the pre-intake fingerprint: + +```bash +verify_reconciliation_inventory() { + fresh_all="$(herdr pane list --workspace "$HERDR_WORKSPACE_ID")" || return 1 + TAB_PANES_JSON="$(printf '%s' "$fresh_all" | bun /scripts/fleet-state.mjs \ + --assert-fingerprint "$RECONCILIATION_FINGERPRINT" \ + --workspace-id "$HERDR_WORKSPACE_ID" --tab-id "$HERDR_TAB_ID")" || return 1 +} + +capture_reconciliation_inventory() { + fresh_all="$(herdr pane list --workspace "$HERDR_WORKSPACE_ID")" || return 1 + TAB_PANES_JSON="$(printf '%s' "$fresh_all" | bun /scripts/fleet-state.mjs \ + --workspace-id "$HERDR_WORKSPACE_ID" --tab-id "$HERDR_TAB_ID")" || return 1 + RECONCILIATION_FINGERPRINT="$(printf '%s' "$TAB_PANES_JSON" | \ + bun /scripts/fleet-state.mjs --fingerprint \ + --workspace-id "$HERDR_WORKSPACE_ID" --tab-id "$HERDR_TAB_ID" | \ + bun -e 'process.stdout.write(JSON.parse(await Bun.stdin.text()).fingerprint)')" +} + +verify_mutation_target() { + target_pane_id="$1" + expected_target_json="$2" + verify_reconciliation_inventory || return 1 + TARGET_PANE_JSON="$(herdr pane get "$target_pane_id")" || return 1 + if TARGET_PROCESS_INFO="$(herdr pane process-info "$target_pane_id" 2>&1)"; then + TARGET_PROCESS_INFO_STATUS=running + else + TARGET_PROCESS_INFO_STATUS=exited + fi + VERIFY_PAYLOAD="$(ACTUAL_PANE_JSON="$TARGET_PANE_JSON" EXPECTED_TARGET_JSON="$expected_target_json" bun -e ' +const actual = JSON.parse(process.env.ACTUAL_PANE_JSON); +const pane = actual?.result?.pane ?? actual?.pane ?? actual?.result ?? actual; +process.stdout.write(JSON.stringify({ pane, expected: JSON.parse(process.env.EXPECTED_TARGET_JSON) })); +')" + printf '%s' "$VERIFY_PAYLOAD" | bun /scripts/fleet-state.mjs --verify-target +} + +verify_reconciliation_inventory || exit 1 +``` + +If this fails, topology changed during intake. Abort mutation and ask the user to confirm a newly classified preview. The confirmed roster is the source of truth for all later role, launch, and assignment decisions. There is no default worker count or default worker roster. + +## Step 4: Establish merge policy + +Use `AskUserQuestion` to establish merge policy. Default to `report-only` and record the answer in the control transcript. + +| Policy | Behavior | +|---|---| +| `report-only` | Stop at a fresh `MERGE_READY` verdict and report it. This is the default. | +| `authorized-merge` | Merge only on explicitly allowed base branches, with the explicitly selected strategy, after a fresh verdict and green required checks. | + +Authorization must name the repository, allowed base branches, merge strategy, and branch-deletion policy. It does not bypass tool permissions; approval prompts go directly to the principal. + +## Step 5: Reconcile confirmed workers + +Before renaming, splitting, or launching, compare the confirmed roster with exact ownership metadata from `$TAB_PANES_JSON`. Inspect each candidate with `herdr pane get`, `herdr pane process-info`, and a recent transcript read. Immediately before **every** close or rename, call `verify_mutation_target` with the expected pane ID, scope, label, role, and owned metadata classification. It refreshes the list, reads the pane back with `pane get`, captures `pane process-info`, and rejects any mismatch. Abort or ask again on any change. After each deliberate mutation, call `capture_reconciliation_inventory` to establish the next baseline. + +Classify each confirmed entry: + +- **Healthy owned match:** identity metadata and command evidence match the confirmed entry, and status is `idle`, `working`, `blocked`, or `done`. Reuse it. +- **Stale owned match:** identity matches, but the process exited, returned to a shell, or is otherwise proven abandoned. Retire only this pane. +- **User-confirmed legacy match:** repository cwd and a complete legacy key/role label match, and the user explicitly maps it to a confirmed entry after command inspection. Reuse it, then stamp current metadata in Step 7. +- **Foreign or ambiguous pane:** ownership evidence is missing or mismatched. Leave it untouched and do not create a same-label replacement without another user decision. +- **Duplicate healthy label:** stop and ask which pane to keep. +- **Missing entry:** record it for Step 6. + +Also identify owned workers absent from the confirmed roster. Ask before retiring them; roster confirmation alone is not process-kill consent. + +Close a stale pane only after recording evidence and its pane ID, then refresh the workspace inventory and re-filter the current tab: + +```bash +EXPECTED_TARGET_JSON='' +verify_mutation_target "$EXPECTED_TARGET_JSON" || exit 1 +# Require TARGET_PROCESS_INFO_STATUS=exited and the same fresh owned-worker classification. +herdr pane close +capture_reconciliation_inventory || exit 1 +``` + +If another healthy `${CONTROL_LABEL}` exists outside the current pane, stop and direct the principal to it. Replace it only when proven stale or explicitly authorized. Then claim and verify the current control pane: + +```bash +EXPECTED_TARGET_JSON='' +verify_mutation_target "$HERDR_PANE_ID" "$EXPECTED_TARGET_JSON" || exit 1 +herdr pane rename "$HERDR_PANE_ID" "$CONTROL_LABEL" +capture_reconciliation_inventory || exit 1 +herdr pane layout --pane "$HERDR_PANE_ID" +``` + +## Step 6: Create only missing confirmed workers + +Follow the confirmed pane-map placements. Use explicit source pane IDs from the current-tab inventory for every split. Immediately before each split, call `verify_mutation_target` with the source pane's expected ID, workspace, tab, label, and owned classification from the confirmed plan. Abort or ask again on any list, readback, process, ownership, or label change. Always use `--no-focus`; read `result.pane.pane_id` from JSON; verify the returned workspace and tab; never predict IDs. After each successful split, call `capture_reconciliation_inventory` before considering another mutation. + +Create exactly one pane for each missing confirmed roster entry. Any number of workers is valid. A partial surviving fleet becomes reused-plus-missing, never duplicated. + +## Step 7: Launch new workers and stamp all confirmed workers + +Start each new worker with its confirmed command, wait for its interactive agent when applicable, and verify the command through `pane process-info`. Reused workers keep their sessions. Before watcher startup, stamp every confirmed new, reused, or user-confirmed legacy worker with current metadata. + +```bash +EXPECTED_TARGET_JSON='' +verify_mutation_target "$EXPECTED_TARGET_JSON" || exit 1 +herdr pane rename "${PROJECT_KEY}-" +capture_reconciliation_inventory || exit 1 +herdr pane run "" +herdr wait agent-status --status idle --timeout 60000 +``` + +Record reconstructable ownership metadata on every confirmed worker. Omit `fleet_assignment` only when the user left it empty. + +```bash +# Generate this JSON from the confirmed roster; never rebuild it by parsing shell text. +EXPECTED_WORKER_JSON='' +METADATA_SHA="$(printf '%s' "$EXPECTED_WORKER_JSON" | \ + bun /scripts/fleet-state.mjs --metadata-hash)" || exit 1 +# Add --token "fleet_assignment=" only when non-empty. +herdr pane report-metadata \ + --source user:herdr-fleet \ + --token "fleet_owner=$FLEET_OWNER_TOKEN" \ + --token "fleet_key=$PROJECT_KEY" \ + --token "fleet_kind=worker" \ + --token "fleet_label=" \ + --token "fleet_role=" \ + --token "fleet_command=" \ + --token "fleet_placement=" \ + --token "fleet_metadata_sha=$METADATA_SHA" +``` + +Herdr metadata values are bounded. Store commands only when they contain no credential and fit without truncation. Read every value back before trusting the pane as reusable: + +```bash +PANE_JSON="$(herdr pane get )" || exit 1 +VERIFY_PAYLOAD="$(ACTUAL_PANE_JSON="$PANE_JSON" EXPECTED_WORKER_JSON="$EXPECTED_WORKER_JSON" bun -e ' +const actual = JSON.parse(process.env.ACTUAL_PANE_JSON); +const pane = actual?.result?.pane ?? actual?.pane ?? actual?.result ?? actual; +process.stdout.write(JSON.stringify({ pane, expected: JSON.parse(process.env.EXPECTED_WORKER_JSON) })); +')" +printf '%s' "$VERIFY_PAYLOAD" | bun /scripts/fleet-state.mjs --verify-metadata || exit 1 +``` + +The verifier compares every field, the rendered pane label, and `fleet_metadata_sha`, which is computed from the confirmed pre-write values. A truncated or omitted value cannot reproduce that hash. If readback differs, record the pane as non-reusable and do not arm ownership-based monitoring for it. Correct the metadata or require intake on the next rebuild rather than claiming proof that does not exist. + +## Step 8: Broadcast standing constraints + +Send this to every new or reused worker before assignment: + +> STANDING CONSTRAINT: you are a worker pane. Remote control of this Herdr session is reserved for the control pane (`$CONTROL_LABEL`). Follow your confirmed role and assignment. Never run Herdr commands or merge pull requests. Never switch branches in the canonical checkout; use short-lived worktrees from the required remote base. Commit incrementally. Report results only in your own transcript. + +For workers prone to nested delegation, add bounded-delegation guidance. Read each transcript and verify delivery; re-send after blocked dialogs or unsubmitted pastes. + +## Step 9: Dispatch confirmed assignments + +Inspect repository instructions, contribution guidance, pull-request templates, issue labels, open pull requests, and worktree conventions first. + +- **Implementers:** include claim procedure, branch/worktree rules, gates, hooks, and the confirmed assignment/lane. +- **Reviewers:** include the review workflow, claim mutex, peer labels, fresh-head verdict requirement, and confirmed queue scope. +- **Other roles:** dispatch only the user-confirmed responsibility and boundaries. +- **Unassigned workers:** hold idle or ask; do not invent work merely to occupy panes. + +## Step 10: Arm the scoped watcher + +A confirmed zero-worker roster needs no watcher. Otherwise, start or reuse exactly one project-scoped watcher: + +```bash +MONITOR_DIR="${TMPDIR:-/tmp}/herdr-fleet-${HERDR_WORKSPACE_ID//:/_}-${HERDR_TAB_ID//:/_}-${PROJECT_KEY}" +mkdir -p "$MONITOR_DIR" +reuse_running_watcher() { + [ -f "$MONITOR_DIR/pid" ] && [ -f "$MONITOR_DIR/instance-token" ] || return 1 + existing_pid="$(cat "$MONITOR_DIR/pid")" + existing_instance_token="$(cat "$MONITOR_DIR/instance-token")" + case "$existing_pid" in ''|*[!0-9]*) return 1 ;; esac + [ "$existing_pid" -gt 1 ] || return 1 + case "$existing_instance_token" in ''|*[!A-Za-z0-9-]*) return 1 ;; esac + existing_command="$(ps -ww -p "$existing_pid" -o command= 2>/dev/null || true)" + expected_suffix="watch-fleet.mjs --project-key $PROJECT_KEY --owner-token $FLEET_OWNER_TOKEN --workspace-id $HERDR_WORKSPACE_ID --tab-id $HERDR_TAB_ID --instance-token $existing_instance_token" + case "$existing_command" in + *"$expected_suffix") kill -0 "$existing_pid" 2>/dev/null ;; + *) return 1 ;; + esac +} + +release_start_lock() { + [ "$(cat "$MONITOR_DIR/start.lock" 2>/dev/null || true)" = "$$" ] && rm -f "$MONITOR_DIR/start.lock" +} + +REUSE_WATCHER=no +START_LOCK_HELD=no +attempt=0 +while [ "$attempt" -lt 30 ]; do + if reuse_running_watcher; then REUSE_WATCHER=yes; break; fi + if shlock -f "$MONITOR_DIR/start.lock" -p "$$"; then + START_LOCK_HELD=yes + trap release_start_lock EXIT INT TERM + if reuse_running_watcher; then REUSE_WATCHER=yes; fi + break + fi + attempt=$((attempt + 1)) + sleep 1 +done + +if [ "$REUSE_WATCHER" = yes ]; then + FLEET_WATCHER_PID="$existing_pid" +elif [ "$START_LOCK_HELD" = yes ]; then + rm -f "$MONITOR_DIR/pid" "$MONITOR_DIR/instance-token" "$MONITOR_DIR/events.cursor" + WATCHER_INSTANCE_TOKEN="$(bun -e 'process.stdout.write(crypto.randomUUID())')" + printf '%s\n' "$WATCHER_INSTANCE_TOKEN" >"$MONITOR_DIR/instance-token.tmp-$$" + mv "$MONITOR_DIR/instance-token.tmp-$$" "$MONITOR_DIR/instance-token" + bun /scripts/watch-fleet.mjs \ + --project-key "$PROJECT_KEY" --owner-token "$FLEET_OWNER_TOKEN" \ + --workspace-id "$HERDR_WORKSPACE_ID" --tab-id "$HERDR_TAB_ID" \ + --instance-token "$WATCHER_INSTANCE_TOKEN" \ + >"$MONITOR_DIR/events.ndjson" 2>"$MONITOR_DIR/errors.ndjson" & + FLEET_WATCHER_PID=$! + printf '%s\n' "$FLEET_WATCHER_PID" >"$MONITOR_DIR/pid.tmp-$$" + mv "$MONITOR_DIR/pid.tmp-$$" "$MONITOR_DIR/pid" + sleep 1 + if ! reuse_running_watcher; then + rm -f "$MONITOR_DIR/pid" "$MONITOR_DIR/instance-token" + release_start_lock + trap - EXIT INT TERM + exit 1 + fi +else + exit 1 +fi +if [ "$START_LOCK_HELD" = yes ]; then + release_start_lock + trap - EXIT INT TERM +fi +``` + +The watcher filters by workspace, tab, exact project identity, and ownership token; requires a worker during bounded startup; tracks status changes; parses only current context footers; retries compaction protection; and stops after bounded failures. Treat watcher exit as a blocker. Consume only its NDJSON, never global Herdr events. + +For shutdown: + +```bash +if reuse_running_watcher; then + kill "$existing_pid" + wait "$existing_pid" 2>/dev/null || true + rm -f "$MONITOR_DIR/pid" "$MONITOR_DIR/instance-token" +else + printf '%s\n' "watcher PID no longer matches this fleet; refusing to kill it" >&2 + exit 1 +fi +``` + +Create bounded CI waits per pull request covering success, failure, cancellation, and timeout. + +## Step 11: Report and enter the loop + +Report merge policy, workspace/tab IDs, project key, confirmed roster, pane IDs, reused/created/retired workers, assignments, watcher PID, and any blockers. Then follow [protocols.md](protocols.md). diff --git a/skills/herdr-fleet/protocols.md b/skills/herdr-fleet/protocols.md new file mode 100644 index 0000000..099d66e --- /dev/null +++ b/skills/herdr-fleet/protocols.md @@ -0,0 +1,159 @@ +# Fleet event-loop protocols + +Consume only the watcher created for the current `$HERDR_WORKSPACE_ID`, `$HERDR_TAB_ID`, and `$PROJECT_KEY`. Before acting on an event, reject any pane whose live record does not match all three scope checks. Never consume another tab's fleet events. + +Consume events one at a time with a persisted byte cursor. `--next` returns one complete, scope-validated NDJSON event without advancing the cursor; `--ack` advances atomically only after the control pane successfully handles it. + +```bash +EVENT_SESSION_ID="fleet:${HERDR_WORKSPACE_ID}:${HERDR_TAB_ID}:${PROJECT_KEY}" +WATCHER_INSTANCE_TOKEN="$(cat "$MONITOR_DIR/instance-token")" || exit 1 +WATCHER_COMMAND_SUFFIX="watch-fleet.mjs --project-key $PROJECT_KEY --owner-token $FLEET_OWNER_TOKEN --workspace-id $HERDR_WORKSPACE_ID --tab-id $HERDR_TAB_ID --instance-token $WATCHER_INSTANCE_TOKEN" +while true; do + NEXT="$(bun /scripts/consume-events.mjs --next \ + --events "$MONITOR_DIR/events.ndjson" \ + --cursor "$MONITOR_DIR/events.cursor" \ + --pid-file "$MONITOR_DIR/pid" \ + --instance-token-file "$MONITOR_DIR/instance-token" \ + --watcher-command-suffix "$WATCHER_COMMAND_SUFFIX" \ + --instance-token "$WATCHER_INSTANCE_TOKEN" \ + --session-id "$EVENT_SESSION_ID" \ + --wait-seconds 30)" || break + status="$(printf '%s' "$NEXT" | bun -e ' +const value = JSON.parse(await Bun.stdin.text()); +process.stdout.write(value.status ?? "event"); +')" + [ "$status" = no_event ] && continue + + NEXT_CURSOR="$(printf '%s' "$NEXT" | bun -e ' +process.stdout.write(String(JSON.parse(await Bun.stdin.text()).nextCursor)); +')" + if [ "$status" = event ]; then + EVENT="$(printf '%s' "$NEXT" | bun -e ' +process.stdout.write(JSON.stringify(JSON.parse(await Bun.stdin.text()).event)); +')" + + # Re-read any referenced pane and reject it unless workspace, tab, key, and owner metadata still match. + # Read the transcript, perform the event action, and record evidence. Ack only after success. + fi + + # Acknowledge a handled event or a blank-line skip placeholder. + bun /scripts/consume-events.mjs --ack "$NEXT_CURSOR" \ + --events "$MONITOR_DIR/events.ndjson" \ + --cursor "$MONITOR_DIR/events.cursor" \ + --pid-file "$MONITOR_DIR/pid" \ + --instance-token-file "$MONITOR_DIR/instance-token" \ + --start-lock "$MONITOR_DIR/start.lock" \ + --watcher-command-suffix "$WATCHER_COMMAND_SUFFIX" \ + --instance-token "$WATCHER_INSTANCE_TOKEN" \ + --session-id "$EVENT_SESSION_ID" +done +``` + +A blank line returns `status: "skip"`; the loop bypasses event extraction and acknowledges its exact cursor, so later events remain reachable. A crash before acknowledgment replays the unhandled event; a restart after acknowledgment resumes at the next byte. Acknowledgment acquires the same `start.lock` used by watcher replacement, revalidates PID, instance token, and exact command identity, then re-reads the pending event and verifies its session, generation, and next cursor before advancing. The lock prevents a replacement generation from starting between the final identity check and cursor rename. The consumer reads only a bounded chunk from the saved offset, so the append-only stream does not get reloaded on each event. Partial trailing lines remain buffered on disk. Before waiting, it validates the PID, stored instance token, and exact scoped command suffix. A recycled PID or exited watcher fails closed and blocks the fleet rather than silently missing events. + +For each accepted event, read the relevant transcript tail, act, and give the principal a proportionate update: a line for routine movement and a structured report for milestones. + +## Consume review verdicts + +Reviewers end each pull request with a `VERDICT:` block containing: + +- `MERGE_READY`, `NEEDS_WORK`, or `BLOCKED`; +- the reviewed head SHA; +- required-gate status; +- bounded fix scope when applicable. + +A verdict is fresh only when its SHA exactly equals the current pull-request head. Every head change invalidates every earlier verdict. + +Before accepting `MERGE_READY` or reporting readiness, run the paginated review-thread gate and record its complete JSON evidence in the control transcript: + +```bash +set +e +REVIEW_THREAD_EVIDENCE="$(bun /scripts/review-thread-gate.mjs \ + --repo "$REPOSITORY_OWNER_AND_NAME" --pr "$PULL_REQUEST_NUMBER" \ + --expected-head "$CURRENT_HEAD_SHA")" +REVIEW_THREAD_STATUS=$? +set -e +printf '%s\n' "$REVIEW_THREAD_EVIDENCE" +(( REVIEW_THREAD_STATUS == 0 )) || exit "$REVIEW_THREAD_STATUS" +``` + +The gate records every thread ID, URL, resolution state, and outdated state. Any unresolved, non-outdated thread blocks readiness. Verify repository identity and current head through GitHub before constructing these arguments. + +Act by verdict: + +- **`MERGE_READY` plus green required checks and a passing review-thread gate:** apply the launch-time merge policy. Under `report-only`, report readiness and stop. Under `authorized-merge`, verify the repository, base branch, strategy, branch-deletion setting, current head SHA, and required checks all match the recorded authorization, then re-run the review-thread gate immediately before merging from the control pane. A tool permission prompt still goes to the principal. +- **`NEEDS_WORK`:** dispatch the fix scope to the authoring implementer when healthy, otherwise to a free implementer with a self-contained brief. A push starts a fresh verdict cycle. +- **`BLOCKED`:** record the blocking dependency and owner. Escalate only decisions reserved for the principal. +- **Conflicting verdicts:** prioritize concrete, reproducible findings. Preserve complementary reviews and require all blocking findings to be resolved. A substantive `NEEDS_WORK` verdict supersedes a less substantive ready verdict at the same head. + +A merge cycle is complete only when the policy decision, remote pull-request state, cleanup result, tracking update, and affected sibling pull requests are known. + +## Handle every head change + +Any content push, conflict resolution, metadata commit, or base sync invalidates the prior verdict. + +- **Content change:** require a full review and a new verdict naming the new head SHA. +- **Mechanical base sync without source-file merges:** allow a short delta review, but require it to inspect the new head and issue a new verdict naming that SHA. +- **Base sync that auto-merges source files:** require a semantic delta review of each resolution and a new verdict naming the new head SHA. + +Immediately before merge, compare the verdict SHA with the live pull-request head and re-run `review-thread-gate.mjs` against that SHA. Record the second evidence payload in the control transcript. A head mismatch or newly unresolved non-outdated thread returns the pull request to review; it never inherits the old verdict. + +## Resolve append-only shared-file conflicts + +GitHub does not honor local `merge=union` drivers. A merge to the base branch can mark every open pull request carrying an append-only file as dirty, and server-side branch updates may refuse the conflict. + +For each affected pull request, use its own worktree: + +```bash +git fetch origin +git merge origin/ --no-edit +# Resolve append-only files according to the repository's documented policy. +git push +``` + +Apply repository-required provenance and hooks. The resulting head requires a new delta verdict. Sequence merges to minimize repeated syncs, and defer control-authored base updates until the current pull-request wave lands. + +## Triage red CI + +Classify before rerunning: + +1. Read the failing job log and identify the exact step or test. +2. Compare the failure with the pull-request diff. +3. Reproduce locally in the owning worktree or find the same failure on the base branch or unrelated runs. +4. For an environmental failure, record evidence and rerun once. On the second recurrence of the same class, create or assign a structural fix. +5. For a real regression, dispatch a `NEEDS_WORK` fix scope to the owner. + +Triage is complete when the failure has an evidence-backed class, owner, and next action. + +## Detect and salvage wedges + +Suspect a wedge when context is near full, nested-agent timers run unusually long, or dispatches are ignored. Confirm with worktree modification times; a `working` status alone is not liveness evidence. + +Before salvage, inspect the diff and run the smallest repository gate that checks the changed scope. Preserve valid work with an incremental commit in the worker's short-lived worktree, push it, and either finish the pull request with an explicit salvage note or hand it to a healthy implementer. Retire only a pane proven owned by this project, workspace, and tab. Start its replacement in the same scoped role with the full standing constraints and a self-contained brief. + +## Recycle panes + +Recycle a pane before a hard wedge when it shows excessive context growth, stale working directories, obsolete recaps, or unsubmitted input. Every replacement receives: + +1. the full standing-constraint broadcast; +2. the repository-derived rules relevant to its role; +3. a self-contained assignment with current issue, branch, worktree, PR, and head details. + +Update the watcher state by letting the old scoped pane disappear and the new scoped pane appear. Never broaden watcher scope to find a replacement. + +## Handle watcher failure and shutdown + +- Three consecutive scoped inventory failures stop the watcher. Treat this as `BLOCKED`, inspect its NDJSON error stream, restore socket health, and start one replacement watcher. +- Three context-read failures for one pane emit `pane_monitor_blocked`; inspect that pane directly without changing scope. +- On normal fleet shutdown, terminate the recorded watcher PID, wait for it, remove its PID file, and report the last consumed event. +- If the watcher PID is missing or no longer belongs to the recorded command, do not kill it; reconcile the monitor first. + +## Communicate with the principal + +- **Milestones:** report merges, tracking updates, salvages, and role changes in a short structured block. +- **Routine events:** acknowledge only when useful. +- **Unattended session:** use the configured notification mechanism for merges, blockers, and decisions. +- **Decisions:** use the available question tool for merge authorization, product direction, scope forks, spend, destructive actions, or policy exceptions. Put the recommended option first. +- **Unexpected actors or changes:** document evidence on the relevant tracking item and route the output through review with heightened scrutiny rather than automatically reverting or duplicating it. + +The event is closed only when the action and its evidence are recorded in the control transcript. diff --git a/skills/herdr-fleet/scripts/consume-events.mjs b/skills/herdr-fleet/scripts/consume-events.mjs new file mode 100644 index 0000000..8bfb729 --- /dev/null +++ b/skills/herdr-fleet/scripts/consume-events.mjs @@ -0,0 +1,406 @@ +#!/usr/bin/env bun + +import assert from "node:assert/strict"; +import { spawnSync } from "node:child_process"; +import { + closeSync, + existsSync, + fstatSync, + openSync, + readFileSync, + readSync, + renameSync, + statSync, + unlinkSync, + writeFileSync, +} from "node:fs"; +import { tmpdir } from "node:os"; +import path from "node:path"; + +function option(name) { + const index = process.argv.indexOf(name); + const value = index === -1 ? undefined : process.argv[index + 1]; + return !value || value.startsWith("--") ? undefined : value; +} + +function cursorAt(cursorPath) { + if (!existsSync(cursorPath)) return 0; + const value = Number(readFileSync(cursorPath, "utf8").trim()); + if (!Number.isSafeInteger(value) || value < 0) throw new Error("invalid cursor file"); + return value; +} + +export function readFromCursor(eventsPath, cursor, maxBytes = 65_536) { + const descriptor = openSync(eventsPath, "r"); + try { + const size = fstatSync(descriptor).size; + if (cursor > size) throw new Error("event file truncated behind saved cursor"); + const length = Math.min(size - cursor, maxBytes); + const buffer = Buffer.alloc(length); + return buffer.subarray(0, readSync(descriptor, buffer, 0, length, cursor)); + } finally { + closeSync(descriptor); + } +} + +function parseJson(text, context) { + try { + return JSON.parse(text); + } catch (error) { + throw new Error(`${context}: ${error.message}`); + } +} + +export function nextEvent(eventsPath, cursor, expectedSessionId, expectedGeneration) { + const data = readFromCursor(eventsPath, cursor); + const newline = data.indexOf(0x0a); + if (newline === -1) { + if (data.length === 65_536) throw new Error("event exceeds maximum line size"); + return undefined; + } + const line = data.subarray(0, newline).toString("utf8").trim(); + if (!line) return { status: "skip", nextCursor: cursor + newline + 1 }; + const event = parseJson(line, "invalid watcher event"); + if (event.sessionId !== expectedSessionId) throw new Error("event session scope mismatch"); + if (event.generation !== expectedGeneration) throw new Error("event watcher generation mismatch"); + return { event, nextCursor: cursor + newline + 1 }; +} + +export function watcherIdentityMatches(candidate) { + const { pid, storedToken, requestedToken, expectedCommandSuffix, command } = candidate; + if (!Number.isSafeInteger(pid) || pid <= 1 || !/^[A-Za-z0-9-]+$/.test(storedToken ?? "")) return false; + if (requestedToken !== storedToken) return false; + if (!expectedCommandSuffix?.endsWith(`--instance-token ${storedToken}`)) return false; + return command?.trimEnd().endsWith(expectedCommandSuffix) ?? false; +} + +function watcherRunning(pidPath, instanceTokenPath, expectedCommandSuffix, expectedGeneration) { + if (!existsSync(pidPath) || !existsSync(instanceTokenPath)) return false; + const pid = Number(readFileSync(pidPath, "utf8").trim()); + const storedToken = readFileSync(instanceTokenPath, "utf8").trim(); + const result = spawnSync("ps", ["-ww", "-p", String(pid), "-o", "command="], { + encoding: "utf8", + timeout: 5000, + }); + if (result.error || result.status !== 0) return false; + return watcherIdentityMatches({ + pid, + storedToken, + requestedToken: expectedGeneration, + expectedCommandSuffix, + command: result.stdout, + }); +} + +function startLockManager(lockPath, fileSystem = { existsSync, readFileSync, unlinkSync }) { + return { + acquire() { + if (!lockPath) return false; + const result = spawnSync("shlock", ["-f", lockPath, "-p", String(process.pid)], { timeout: 5000 }); + return !result.error && result.status === 0; + }, + release() { + try { + if (!fileSystem.existsSync(lockPath)) return; + if (fileSystem.readFileSync(lockPath, "utf8").trim() === String(process.pid)) { + fileSystem.unlinkSync(lockPath); + } + } catch (error) { + if (error?.code !== "ENOENT") throw error; + } + }, + }; +} + +function acknowledgeLocked(options) { + const { + eventsPath, + cursorPath, + nextCursor, + expectedSessionId, + expectedGeneration, + pidPath, + instanceTokenPath, + expectedCommandSuffix, + identityValidator, + } = options; + if (!identityValidator(pidPath, instanceTokenPath, expectedCommandSuffix, expectedGeneration)) { + throw new Error("watcher identity changed before acknowledgment"); + } + const current = cursorAt(cursorPath); + const size = statSync(eventsPath).size; + if (!Number.isSafeInteger(nextCursor) || nextCursor <= current || nextCursor > size) { + throw new Error("ack cursor is outside the pending event range"); + } + const pending = nextEvent(eventsPath, current, expectedSessionId, expectedGeneration); + if (!pending || pending.nextCursor !== nextCursor) throw new Error("ack does not match pending event"); + const temporary = `${cursorPath}.tmp-${process.pid}`; + writeFileSync(temporary, `${nextCursor}\n`, { mode: 0o600 }); + if (!identityValidator(pidPath, instanceTokenPath, expectedCommandSuffix, expectedGeneration)) { + unlinkSync(temporary); + throw new Error("watcher identity changed during acknowledgment"); + } + renameSync(temporary, cursorPath); +} + +export function acknowledge(options) { + const lockManager = options.lockManager ?? startLockManager(options.startLockPath); + if (!lockManager.acquire()) throw new Error("watcher start lock unavailable during acknowledgment"); + try { + acknowledgeLocked({ ...options, identityValidator: options.identityValidator ?? watcherRunning }); + } finally { + lockManager.release(); + } +} + +async function readNext( + eventsPath, + cursorPath, + pidPath, + instanceTokenPath, + expectedCommandSuffix, + expectedSessionId, + expectedGeneration, + waitMs, +) { + const deadline = Date.now() + waitMs; + while (Date.now() < deadline) { + if (!watcherRunning(pidPath, instanceTokenPath, expectedCommandSuffix, expectedGeneration)) { + throw new Error("watcher identity changed before the next complete event"); + } + const cursor = cursorAt(cursorPath); + if (existsSync(eventsPath)) { + const result = nextEvent(eventsPath, cursor, expectedSessionId, expectedGeneration); + if (result) return result; + } + await new Promise((resolve) => setTimeout(resolve, 500)); + } + return undefined; +} + +function selfTest() { + const root = path.join(tmpdir(), `herdr-fleet-consumer-${process.pid}`); + const eventsPath = `${root}.ndjson`; + const cursorPath = `${root}.cursor`; + const sessionId = "fleet:w1:w1:t1:app"; + const generation = "instance-a"; + const first = `${JSON.stringify({ sessionId, generation, event: "first" })}\n`; + writeFileSync(eventsPath, `${first}${JSON.stringify({ sessionId, generation, event: "partial" })}`); + const result = nextEvent(eventsPath, 0, sessionId, generation); + assert.equal(result.event.event, "first"); + const ackOptions = { + eventsPath, + cursorPath, + nextCursor: result.nextCursor, + expectedSessionId: sessionId, + expectedGeneration: generation, + pidPath: `${root}.pid`, + instanceTokenPath: `${root}.instance-token`, + expectedCommandSuffix: "watch-fleet.mjs --instance-token instance-a", + lockManager: { acquire: () => true, release: () => {} }, + }; + const replacementIdentity = watcherIdentityMatches({ + pid: 4242, + storedToken: "instance-b", + requestedToken: generation, + expectedCommandSuffix: "watch-fleet.mjs --instance-token instance-b", + command: "bun watch-fleet.mjs --instance-token instance-b", + }); + assert.throws(() => acknowledge({ ...ackOptions, identityValidator: () => replacementIdentity }), /identity changed/); + assert.equal(cursorAt(cursorPath), 0); + const missingLockError = Object.assign(new Error("lock disappeared"), { code: "ENOENT" }); + let racedLockRead = false; + const racedLock = startLockManager(`${root}.race-lock`, { + existsSync: () => true, + readFileSync: () => { + racedLockRead = true; + throw missingLockError; + }, + unlinkSync: () => assert.fail("missing lock must not be unlinked"), + }); + assert.throws( + () => + acknowledge({ + ...ackOptions, + identityValidator: () => false, + lockManager: { acquire: () => true, release: () => racedLock.release() }, + }), + /watcher identity changed/, + ); + assert.equal(racedLockRead, true); + let identityChecks = 0; + assert.throws( + () => + acknowledge({ + ...ackOptions, + identityValidator: () => { + identityChecks += 1; + return identityChecks === 1; + }, + }), + /identity changed during acknowledgment/, + ); + assert.equal(cursorAt(cursorPath), 0); + assert.throws( + () => + acknowledge({ + ...ackOptions, + expectedSessionId: "fleet:other", + identityValidator: () => true, + }), + /session scope mismatch/, + ); + assert.equal(cursorAt(cursorPath), 0); + assert.throws( + () => + acknowledge({ + ...ackOptions, + expectedGeneration: "instance-b", + identityValidator: () => true, + }), + /generation mismatch/, + ); + assert.equal(cursorAt(cursorPath), 0); + assert.throws( + () => + acknowledge({ + ...ackOptions, + nextCursor: result.nextCursor + 1, + identityValidator: () => true, + }), + /pending event/, + ); + assert.equal(cursorAt(cursorPath), 0); + const lockCalls = []; + acknowledge({ + ...ackOptions, + identityValidator: () => { + lockCalls.push("identity"); + return true; + }, + lockManager: { + acquire: () => { + lockCalls.push("acquire"); + return true; + }, + release: () => { + lockCalls.push("release"); + assert.equal(cursorAt(cursorPath), result.nextCursor); + }, + }, + }); + assert.deepEqual(lockCalls, ["acquire", "identity", "identity", "release"]); + assert.equal(cursorAt(cursorPath), Buffer.byteLength(first)); + assert.equal(nextEvent(eventsPath, cursorAt(cursorPath), sessionId, generation), undefined); + assert.throws(() => nextEvent(eventsPath, 0, "fleet:other", generation)); + assert.throws(() => nextEvent(eventsPath, 0, sessionId, "instance-b")); + const tail = readFromCursor(eventsPath, cursorAt(cursorPath), 9); + assert.equal(tail.length, 9); + assert.equal(tail.toString("utf8"), '{"session'); + + const blankEventsPath = `${root}.blank.ndjson`; + const blankCursorPath = `${root}.blank.cursor`; + writeFileSync(blankEventsPath, "\n"); + const blank = nextEvent(blankEventsPath, 0, sessionId, generation); + assert.equal(blank.status, "skip"); + assert.equal(blank.nextCursor, 1); + acknowledge({ + ...ackOptions, + eventsPath: blankEventsPath, + cursorPath: blankCursorPath, + nextCursor: blank.nextCursor, + identityValidator: () => true, + }); + assert.equal(cursorAt(blankCursorPath), 1); + + const identityFixture = parseJson( + readFileSync(new URL("../fixtures/watcher-identity.json", import.meta.url), "utf8"), + "invalid watcher identity fixture", + ); + for (const testCase of identityFixture.cases) { + const candidate = { + ...testCase, + storedToken: testCase.instanceId, + requestedToken: testCase.instanceId, + }; + assert.equal(watcherIdentityMatches(candidate), testCase.expected, testCase.name); + } + unlinkSync(eventsPath); + unlinkSync(cursorPath); + unlinkSync(blankEventsPath); + unlinkSync(blankCursorPath); + process.stdout.write(`${JSON.stringify({ status: "pass", checks: 27 })}\n`); +} + +if (process.argv.includes("--self-test")) { + selfTest(); + process.exit(0); +} + +const eventsPath = option("--events"); +const cursorPath = option("--cursor"); +const pidPath = option("--pid-file"); +const instanceTokenPath = option("--instance-token-file"); +const startLockPath = option("--start-lock"); +const expectedCommandSuffix = option("--watcher-command-suffix"); +const expectedSessionId = option("--session-id"); +const expectedGeneration = option("--instance-token"); +if (!eventsPath || !cursorPath || !expectedSessionId || !expectedGeneration) { + process.stderr.write( + `${JSON.stringify({ + level: "fatal", + event: "invalid_arguments", + sessionId: expectedSessionId ?? "fleet:unscoped", + })}\n`, + ); + process.exit(2); +} + +try { + if (process.argv.includes("--next")) { + if (!pidPath || !instanceTokenPath || !expectedCommandSuffix) { + throw new Error("--next requires PID, instance-token, and command-suffix identity"); + } + const waitSeconds = Number(option("--wait-seconds") ?? "30"); + const waitMs = Number.isFinite(waitSeconds) && waitSeconds > 0 ? waitSeconds * 1000 : 30_000; + const result = await readNext( + eventsPath, + cursorPath, + pidPath, + instanceTokenPath, + expectedCommandSuffix, + expectedSessionId, + expectedGeneration, + waitMs, + ); + process.stdout.write(`${JSON.stringify(result ?? { status: "no_event" })}\n`); + } else if (process.argv.includes("--ack")) { + if (!pidPath || !instanceTokenPath || !expectedCommandSuffix || !startLockPath) { + throw new Error("--ack requires PID, instance-token, command-suffix, and start-lock identity"); + } + acknowledge({ + eventsPath, + cursorPath, + nextCursor: Number(option("--ack")), + expectedSessionId, + expectedGeneration, + pidPath, + instanceTokenPath, + expectedCommandSuffix, + startLockPath, + }); + process.stdout.write(`${JSON.stringify({ status: "acknowledged" })}\n`); + } else { + throw new Error("choose --next or --ack"); + } +} catch (error) { + process.stderr.write( + `${JSON.stringify({ + level: "error", + event: "consumer_failed", + sessionId: expectedSessionId, + message: error.message, + })}\n`, + ); + process.exit(1); +} diff --git a/skills/herdr-fleet/scripts/fleet-labels.mjs b/skills/herdr-fleet/scripts/fleet-labels.mjs new file mode 100644 index 0000000..dd31815 --- /dev/null +++ b/skills/herdr-fleet/scripts/fleet-labels.mjs @@ -0,0 +1,26 @@ +const LEGACY_ROLE_SUFFIXES = ["control-pane", "claude-impl", "codex-impl", "pi-impl", "codex-review", "claude-review"]; + +export function parseLegacyFleetLabel(label) { + if (typeof label !== "string") return undefined; + for (const role of LEGACY_ROLE_SUFFIXES) { + const suffix = `-${role}`; + if (label.endsWith(suffix) && label.length > suffix.length) { + return { key: label.slice(0, -suffix.length), role, source: "legacy-label" }; + } + } + return undefined; +} + +export function paneFleetIdentity(pane) { + const tokens = pane?.tokens; + if (tokens?.fleet_key) { + return { + key: tokens.fleet_key, + role: tokens.fleet_role, + kind: tokens.fleet_kind, + owner: tokens.fleet_owner, + source: "metadata", + }; + } + return parseLegacyFleetLabel(pane?.label); +} diff --git a/skills/herdr-fleet/scripts/fleet-state.mjs b/skills/herdr-fleet/scripts/fleet-state.mjs new file mode 100644 index 0000000..c079e78 --- /dev/null +++ b/skills/herdr-fleet/scripts/fleet-state.mjs @@ -0,0 +1,158 @@ +#!/usr/bin/env bun + +import { createHash } from "node:crypto"; +import { paneFleetIdentity } from "./fleet-labels.mjs"; + +function extractPanes(payload) { + const value = payload?.result?.panes ?? payload?.panes ?? payload?.result ?? payload; + return Array.isArray(value) ? value : []; +} + +function scopedPanes(payload, { workspaceId, tabId }) { + return extractPanes(payload).filter((pane) => pane.workspace_id === workspaceId && pane.tab_id === tabId); +} + +export function classifiedInventory(payload, scope) { + return scopedPanes(payload, scope) + .map((pane) => ({ + paneId: pane.pane_id, + label: pane.label, + cwd: pane.foreground_cwd ?? pane.cwd, + identity: paneFleetIdentity(pane), + })) + .sort((left, right) => String(left.paneId).localeCompare(String(right.paneId))); +} + +export function inventoryFingerprint(inventory) { + return createHash("sha256").update(JSON.stringify(inventory)).digest("hex"); +} + +export function validateRosterLabels(roster, projectKey) { + if (!Array.isArray(roster)) throw new Error("roster must be an array"); + const controlLabel = `${projectKey}-control-pane`.toLowerCase(); + const renderedLabels = new Set(); + for (const worker of roster) { + const label = typeof worker.label === "string" ? worker.label.trim() : ""; + const commandValue = worker.launchCommand ?? worker.command; + const command = typeof commandValue === "string" ? commandValue.trim() : ""; + if (!label || !command) throw new Error("each worker requires a label and launch command"); + const rendered = `${projectKey}-${label}`.toLowerCase(); + if (rendered === controlLabel) throw new Error(`reserved control label: ${label}`); + if (renderedLabels.has(rendered)) throw new Error(`duplicate worker label: ${label}`); + renderedLabels.add(rendered); + } + return roster; +} + +export function mutationTargetMismatches(pane, expected) { + const identity = paneFleetIdentity(pane) ?? {}; + const actual = { + paneId: pane?.pane_id, + workspaceId: pane?.workspace_id, + tabId: pane?.tab_id, + label: pane?.label, + projectKey: identity.key, + ownerToken: identity.owner, + kind: identity.kind, + role: identity.role, + }; + return Object.keys(actual).filter((key) => expected[key] !== undefined && actual[key] !== expected[key]); +} + +export function workerMetadataHash(expected) { + const canonical = { + ownerToken: expected.ownerToken, + projectKey: expected.projectKey, + label: expected.label, + role: expected.role, + assignment: expected.assignment ?? null, + command: expected.command, + placement: expected.placement, + }; + return createHash("sha256").update(JSON.stringify(canonical)).digest("hex").slice(0, 24); +} + +export function metadataMismatches(pane, expected) { + const tokens = pane?.tokens ?? {}; + const required = { + fleet_owner: expected.ownerToken, + fleet_key: expected.projectKey, + fleet_kind: "worker", + fleet_label: expected.label, + fleet_role: expected.role, + fleet_command: expected.command, + fleet_placement: expected.placement, + fleet_metadata_sha: workerMetadataHash(expected), + }; + if (expected.assignment) required.fleet_assignment = expected.assignment; + const mismatches = Object.entries(required) + .filter(([key, value]) => tokens[key] !== value) + .map(([key]) => key); + if (!expected.assignment && Object.hasOwn(tokens, "fleet_assignment")) mismatches.push("fleet_assignment"); + if (pane?.label !== `${expected.projectKey}-${expected.label}`) mismatches.push("pane_label"); + return mismatches; +} + +function parseJson(text) { + try { + return JSON.parse(text); + } catch (error) { + throw new Error(`invalid fleet-state JSON: ${error.message}`); + } +} + +function option(name) { + const index = process.argv.indexOf(name); + const value = index === -1 ? undefined : process.argv[index + 1]; + return !value || value.startsWith("--") ? undefined : value; +} + +async function main() { + const payload = parseJson(await Bun.stdin.text()); + const workspaceId = option("--workspace-id"); + const tabId = option("--tab-id"); + if (process.argv.includes("--validate-roster")) { + validateRosterLabels(payload, option("--project-key")); + process.stdout.write(`${JSON.stringify(payload)}\n`); + return; + } + if (process.argv.includes("--metadata-hash")) { + process.stdout.write(`${workerMetadataHash(payload)}\n`); + return; + } + if (process.argv.includes("--verify-target")) { + const mismatches = mutationTargetMismatches(payload.pane, payload.expected); + if (mismatches.length > 0) throw new Error(`mutation target changed: ${mismatches.join(", ")}`); + process.stdout.write(`${JSON.stringify({ status: "verified" })}\n`); + return; + } + if (process.argv.includes("--verify-metadata")) { + const mismatches = metadataMismatches(payload.pane, payload.expected); + if (mismatches.length > 0) throw new Error(`metadata mismatch: ${mismatches.join(", ")}`); + process.stdout.write(`${JSON.stringify({ status: "verified" })}\n`); + return; + } + if (!workspaceId || !tabId) throw new Error("inventory mode requires workspace and tab IDs"); + const scope = { workspaceId, tabId }; + const fingerprint = inventoryFingerprint(classifiedInventory(payload, scope)); + const expected = option("--assert-fingerprint"); + if (expected && expected !== fingerprint) throw new Error("fleet topology changed during roster intake"); + const result = process.argv.includes("--fingerprint") ? { fingerprint } : scopedPanes(payload, scope); + process.stdout.write(`${JSON.stringify(result)}\n`); +} + +if (import.meta.path === Bun.main) { + try { + await main(); + } catch (error) { + process.stderr.write( + `${JSON.stringify({ + level: "fatal", + event: "fleet_state_invalid", + sessionId: "fleet-state", + message: error.message, + })}\n`, + ); + process.exit(1); + } +} diff --git a/skills/herdr-fleet/scripts/fleet-state.test.mjs b/skills/herdr-fleet/scripts/fleet-state.test.mjs new file mode 100644 index 0000000..a81a496 --- /dev/null +++ b/skills/herdr-fleet/scripts/fleet-state.test.mjs @@ -0,0 +1,55 @@ +import { test } from "bun:test"; +import assert from "node:assert/strict"; +import { + classifiedInventory, + inventoryFingerprint, + metadataMismatches, + mutationTargetMismatches, + validateRosterLabels, +} from "./fleet-state.mjs"; + +const fixture = await Bun.file(new URL("../fixtures/fleet-state-race.json", import.meta.url)).json(); +const before = classifiedInventory(fixture.before, fixture.scope); +const afterRace = classifiedInventory(fixture.afterTopologyRace, fixture.scope); + +test("topology races change the scoped inventory fingerprint", () => { + assert.notEqual(inventoryFingerprint(before), inventoryFingerprint(afterRace)); +}); + +test("inventory excludes panes from another tab", () => { + assert.equal(before.length, 1); +}); + +test("worker roster rejects the reserved control label", () => { + assert.throws(() => validateRosterLabels(fixture.rosterWithReservedLabel, "app"), /reserved control label/); +}); + +test("metadata readback detects a truncated command", () => { + assert.deepEqual(metadataMismatches(fixture.metadataReadback.pane, fixture.metadataReadback.expected), [ + "fleet_command", + "fleet_metadata_sha", + ]); +}); + +const expectedMutationTarget = { + paneId: "w1:p1", + workspaceId: "w1", + tabId: "w1:t1", + projectKey: "app", + ownerToken: "owner-1", + kind: "worker", + label: "app-builder", + role: "implementer", +}; + +test("fresh pane readback detects label and role races before mutation", () => { + const racedPane = fixture.afterTopologyRace.result.panes[0]; + assert.deepEqual(mutationTargetMismatches(racedPane, expectedMutationTarget), ["label", "role"]); +}); + +test("fresh pane readback detects scope and ownership races before mutation", () => { + const racedPane = structuredClone(fixture.before.result.panes[0]); + racedPane.workspace_id = "w2"; + racedPane.tokens.fleet_owner = "owner-2"; + assert.deepEqual(mutationTargetMismatches(racedPane, expectedMutationTarget), ["workspaceId", "ownerToken"]); +}); diff --git a/skills/herdr-fleet/scripts/resolve-project-key.mjs b/skills/herdr-fleet/scripts/resolve-project-key.mjs new file mode 100644 index 0000000..873c42d --- /dev/null +++ b/skills/herdr-fleet/scripts/resolve-project-key.mjs @@ -0,0 +1,132 @@ +#!/usr/bin/env bun + +import assert from "node:assert/strict"; +import { createHash } from "node:crypto"; +import { readFileSync } from "node:fs"; +import path from "node:path"; +import { paneFleetIdentity } from "./fleet-labels.mjs"; + +function option(name) { + const index = process.argv.indexOf(name); + return index === -1 ? undefined : process.argv[index + 1]; +} + +function digestPath(repoRoot) { + return createHash("sha256").update(path.resolve(repoRoot)).digest("hex"); +} + +export function fleetOwnerToken(repoRoot) { + return digestPath(repoRoot).slice(0, 16); +} + +function ownsPath(pane, repoRoot) { + const value = pane.foreground_cwd ?? pane.cwd; + if (!value) return false; + const root = path.resolve(repoRoot); + const candidate = path.resolve(value); + return candidate === root || candidate.startsWith(`${root}${path.sep}`); +} + +function provesOwnership(pane, repoRoot, ownerToken) { + const identity = paneFleetIdentity(pane); + if (!identity || !ownsPath(pane, repoRoot)) return false; + return identity.source === "legacy-label" || identity.owner === ownerToken; +} + +function occupants(panes, key) { + return panes.filter((pane) => paneFleetIdentity(pane)?.key === key); +} + +export function resolveProjectKey({ panes, baseKey, repoRoot }) { + const ownerToken = fleetOwnerToken(repoRoot); + const established = new Set( + panes + .filter((pane) => provesOwnership(pane, repoRoot, ownerToken)) + .map((pane) => paneFleetIdentity(pane)?.key) + .filter((key) => key === baseKey || key?.startsWith(`${baseKey}-`)), + ); + + if (established.size > 1) { + throw new Error(`multiple ownership-proven project keys: ${[...established].join(", ")}`); + } + if (established.size === 1) return [...established][0]; + if (occupants(panes, baseKey).length === 0) return baseKey; + + const digest = digestPath(repoRoot); + for (const length of [4, 8, 12, 16, 24, 32, 64]) { + const candidate = `${baseKey}-${digest.slice(0, length)}`; + if (occupants(panes, candidate).length === 0) return candidate; + } + throw new Error("no unoccupied deterministic project key available"); +} + +function selfTest() { + const fixture = JSON.parse(readFileSync(new URL("../fixtures/project-key-collision.json", import.meta.url), "utf8")); + assert.equal(resolveProjectKey(fixture), fixture.expectedKey); + + const repoRoot = "/repos/new-billing-api"; + const digest = digestPath(repoRoot); + assert.equal( + resolveProjectKey({ + baseKey: "ba", + repoRoot, + panes: [ + { label: "ba-control-pane", cwd: "/repos/foreign" }, + { label: `ba-${digest.slice(0, 4)}-pi-impl`, cwd: "/repos/another-foreign" }, + ], + }), + `ba-${digest.slice(0, 8)}`, + ); + assert.equal( + resolveProjectKey({ + baseKey: "ba", + repoRoot, + panes: [ + { + cwd: repoRoot, + tokens: { + fleet_key: "ba-owned", + fleet_owner: fleetOwnerToken(repoRoot), + fleet_kind: "worker", + }, + }, + ], + }), + "ba-owned", + ); + assert.throws(() => + resolveProjectKey({ + baseKey: "ba", + repoRoot, + panes: [ + { label: "ba-control-pane", cwd: repoRoot }, + { label: "ba-abcd-pi-impl", cwd: repoRoot }, + ], + }), + ); + process.stdout.write(`${JSON.stringify({ status: "pass", checks: 4 })}\n`); +} + +if (process.argv.includes("--self-test")) { + selfTest(); + process.exit(0); +} + +const baseKey = option("--base-key"); +const repoRoot = option("--repo-root"); +if (!baseKey || !repoRoot) { + process.stderr.write( + `${JSON.stringify({ + level: "fatal", + event: "invalid_arguments", + sessionId: "project-key-resolver", + message: "resolve-project-key requires --base-key and --repo-root", + })}\n`, + ); + process.exit(2); +} + +const panes = JSON.parse(await Bun.stdin.text()); +const projectKey = resolveProjectKey({ panes, baseKey, repoRoot }); +const result = { projectKey, ownerToken: fleetOwnerToken(repoRoot) }; +process.stdout.write(process.argv.includes("--json") ? `${JSON.stringify(result)}\n` : `${projectKey}\n`); diff --git a/skills/herdr-fleet/scripts/review-thread-gate.mjs b/skills/herdr-fleet/scripts/review-thread-gate.mjs new file mode 100644 index 0000000..271da31 --- /dev/null +++ b/skills/herdr-fleet/scripts/review-thread-gate.mjs @@ -0,0 +1,151 @@ +#!/usr/bin/env bun + +import { spawnSync } from "node:child_process"; +import { readFileSync } from "node:fs"; + +const QUERY = `query($owner: String!, $name: String!, $number: Int!, $after: String) { + repository(owner: $owner, name: $name) { + pullRequest(number: $number) { + headRefOid + reviewThreads(first: 100, after: $after) { + nodes { + id + isResolved + isOutdated + comments(first: 1) { nodes { url } } + } + pageInfo { hasNextPage endCursor } + } + } + } +}`; + +function parseJson(text, context) { + try { + return JSON.parse(text); + } catch (error) { + throw new Error(`${context}: ${error.message}`); + } +} + +export async function collectReviewThreads(fetchPage) { + const threads = []; + const seenCursors = new Set(); + let cursor; + let headSha; + for (let pageNumber = 0; pageNumber < 100; pageNumber += 1) { + const payload = await fetchPage(cursor); + const pullRequest = payload?.data?.repository?.pullRequest; + if (!pullRequest) throw new Error("pull request not found"); + if (headSha && pullRequest.headRefOid !== headSha) throw new Error("pull request head changed during pagination"); + headSha = pullRequest.headRefOid; + const connection = pullRequest.reviewThreads; + if (!Array.isArray(connection?.nodes)) throw new Error("invalid review thread response"); + threads.push(...connection.nodes); + if (!connection.pageInfo?.hasNextPage) return { headSha, threads }; + cursor = connection.pageInfo.endCursor; + if (!cursor || seenCursors.has(cursor)) throw new Error("invalid review thread pagination cursor"); + seenCursors.add(cursor); + } + throw new Error("review thread pagination exceeded 100 pages"); +} + +export function reviewThreadEvidence(result) { + const threads = result.threads.map((thread) => { + const url = thread.comments?.nodes?.[0]?.url; + if (!thread.id || !url) throw new Error("review thread is missing ID or URL"); + return { + id: thread.id, + url, + isResolved: thread.isResolved === true, + isOutdated: thread.isOutdated === true, + }; + }); + return { + headSha: result.headSha, + threads, + blockingThreadIds: threads.filter((thread) => !thread.isResolved && !thread.isOutdated).map((thread) => thread.id), + }; +} + +function option(name) { + const index = process.argv.indexOf(name); + const value = index === -1 ? undefined : process.argv[index + 1]; + return !value || value.startsWith("--") ? undefined : value; +} + +function fixturePageFetcher(fixturePath) { + const fixture = parseJson(readFileSync(fixturePath, "utf8"), "invalid review-thread fixture"); + let pageIndex = 0; + return (cursor) => { + if (pageIndex > 0) { + const expectedCursor = fixture.pages[pageIndex - 1].data.repository.pullRequest.reviewThreads.pageInfo.endCursor; + if (cursor !== expectedCursor) throw new Error("fixture pagination cursor mismatch"); + } + const page = fixture.pages[pageIndex]; + pageIndex += 1; + if (!page) throw new Error("fixture page missing"); + return page; + }; +} + +function githubPageFetcher(owner, name, number) { + return (cursor) => { + const args = [ + "api", + "graphql", + "-f", + `query=${QUERY}`, + "-F", + `owner=${owner}`, + "-F", + `name=${name}`, + "-F", + `number=${number}`, + ]; + if (cursor) args.push("-f", `after=${cursor}`); + const result = spawnSync("gh", args, { encoding: "utf8", timeout: 15_000 }); + if (result.error || result.status !== 0) { + throw new Error(result.stderr?.trim() || result.error?.message || "GitHub review-thread query failed"); + } + return parseJson(result.stdout, "invalid GitHub review-thread response"); + }; +} + +async function main() { + const repository = option("--repo"); + const number = Number(option("--pr")); + const expectedHead = option("--expected-head"); + const fixturePath = option("--fixture"); + const [owner, name, extra] = repository?.split("/") ?? []; + if (!owner || !name || extra || !Number.isSafeInteger(number) || number <= 0 || !expectedHead) { + throw new Error("review-thread-gate requires --repo owner/name, --pr, and --expected-head"); + } + const fetchPage = fixturePath ? fixturePageFetcher(fixturePath) : githubPageFetcher(owner, name, number); + const result = await collectReviewThreads(fetchPage); + if (result.headSha !== expectedHead) throw new Error("pull request head does not match expected SHA"); + const evidence = { + repository, + pullRequest: number, + checkedAt: new Date().toISOString(), + ...reviewThreadEvidence(result), + }; + process.stdout.write(`${JSON.stringify(evidence)}\n`); + if (evidence.blockingThreadIds.length > 0) process.exitCode = 3; +} + +if (import.meta.path === Bun.main) { + try { + await main(); + } catch (error) { + process.stderr.write( + `${JSON.stringify({ + level: "fatal", + event: "review_thread_gate_failed", + sessionId: "review-thread-gate", + message: error.message, + })}\n`, + ); + process.exit(1); + } +} diff --git a/skills/herdr-fleet/scripts/review-thread-gate.test.mjs b/skills/herdr-fleet/scripts/review-thread-gate.test.mjs new file mode 100644 index 0000000..5a9b263 --- /dev/null +++ b/skills/herdr-fleet/scripts/review-thread-gate.test.mjs @@ -0,0 +1,92 @@ +import { test } from "bun:test"; +import assert from "node:assert/strict"; +import { fileURLToPath } from "node:url"; +import { collectReviewThreads, reviewThreadEvidence } from "./review-thread-gate.mjs"; + +const fixture = await Bun.file(new URL("../fixtures/review-thread-pages.json", import.meta.url)).json(); + +test("review thread collection follows every page", async () => { + const cursors = []; + const result = await collectReviewThreads(async (cursor) => { + cursors.push(cursor ?? null); + return fixture.pages[cursors.length - 1]; + }); + assert.deepEqual(cursors, [null, "cursor-1"]); + assert.equal(result.threads.length, 4); + assert.equal(result.headSha, "abc123"); +}); + +test("review thread collection rejects head drift between pages", async () => { + const pages = structuredClone(fixture.pages); + pages[1].data.repository.pullRequest.headRefOid = "changed-head"; + let pageIndex = 0; + await assert.rejects( + collectReviewThreads(async () => pages[pageIndex++]), + /head changed during pagination/, + ); +}); + +test("review gate records all states and blocks only current unresolved threads", async () => { + let pageIndex = 0; + const result = await collectReviewThreads(async () => fixture.pages[pageIndex++]); + const evidence = reviewThreadEvidence(result); + assert.deepEqual( + evidence.threads.map(({ id, url, isResolved, isOutdated }) => ({ id, url, isResolved, isOutdated })), + [ + { + id: "thread-unresolved", + url: "https://example.test/unresolved", + isResolved: false, + isOutdated: false, + }, + { + id: "thread-resolved", + url: "https://example.test/resolved", + isResolved: true, + isOutdated: false, + }, + { + id: "thread-outdated", + url: "https://example.test/outdated", + isResolved: false, + isOutdated: true, + }, + { + id: "thread-page-two-unresolved", + url: "https://example.test/page-two", + isResolved: false, + isOutdated: false, + }, + ], + ); + assert.deepEqual(evidence.blockingThreadIds, ["thread-unresolved", "thread-page-two-unresolved"]); +}); + +const scriptPath = fileURLToPath(new URL("./review-thread-gate.mjs", import.meta.url)); +const fixturePath = fileURLToPath(new URL("../fixtures/review-thread-pages.json", import.meta.url)); + +function runFixtureGate(expectedHead) { + return Bun.spawnSync([ + "bun", + scriptPath, + "--repo", + "fixture/repository", + "--pr", + "32", + "--expected-head", + expectedHead, + "--fixture", + fixturePath, + ]); +} + +test("review gate CLI rejects a stale expected head", () => { + assert.equal(runFixtureGate("stale-head").exitCode, 1); +}); + +test("review gate CLI exits three for current unresolved threads", async () => { + const result = runFixtureGate("abc123"); + assert.equal(result.exitCode, 3); + const evidence = await new Response(result.stdout).json(); + assert.deepEqual(evidence.blockingThreadIds, ["thread-unresolved", "thread-page-two-unresolved"]); +}); diff --git a/skills/herdr-fleet/scripts/watch-fleet.mjs b/skills/herdr-fleet/scripts/watch-fleet.mjs new file mode 100644 index 0000000..92e8a20 --- /dev/null +++ b/skills/herdr-fleet/scripts/watch-fleet.mjs @@ -0,0 +1,410 @@ +#!/usr/bin/env bun + +import assert from "node:assert/strict"; +import { spawnSync } from "node:child_process"; +import { readFileSync } from "node:fs"; +import { paneFleetIdentity } from "./fleet-labels.mjs"; + +const STATUS_VALUES = new Set(["idle", "working", "blocked", "done", "unknown"]); + +export function extractPanes(payload) { + const value = payload?.result?.panes ?? payload?.panes ?? payload?.result ?? payload; + return Array.isArray(value) ? value : []; +} + +export function scopePanes(panes, workspaceId, tabId, projectKey, ownerToken) { + return panes.filter((pane) => { + const identity = paneFleetIdentity(pane); + const ownedWorker = + identity?.source === "metadata" && + identity.key === projectKey && + identity.owner === ownerToken && + identity.kind === "worker"; + return pane.workspace_id === workspaceId && pane.tab_id === tabId && ownedWorker; + }); +} + +export function parseContext(text) { + const result = {}; + for (const rawLine of text.split(/\r?\n/).toReversed()) { + const line = rawLine.trim(); + if (!line) continue; + + const bracketed = line.match(/^\[(\d+(?:\.\d+)?)%[^\]]*\]$/i); + const contextUsed = line.match(/^(\d+(?:\.\d+)?)%\s+context used$/i); + const tokenFooter = line.match(/^(\d+(?:\.\d+)?)%\s*\/\s*\d+(?:\.\d+)?[km]?$/i); + const until = line.match(/^(\d+(?:\.\d+)?)%\s+until auto-compact$/i); + const used = bracketed?.[1] ?? contextUsed?.[1] ?? tokenFooter?.[1]; + + if (used !== undefined && result.usedPercent === undefined) result.usedPercent = Number(used); + if (until !== null && result.untilAutoCompactPercent === undefined) { + result.untilAutoCompactPercent = Number(until[1]); + } + if (used === undefined && until === null) break; + } + return result; +} + +function option(name, fallback) { + const index = process.argv.indexOf(name); + const value = index === -1 ? undefined : process.argv[index + 1]; + return !value || value.startsWith("--") ? fallback : value; +} + +function positiveSeconds(value, fallback) { + const parsed = Number(value); + return Number.isFinite(parsed) && parsed > 0 ? parsed : fallback; +} + +function positiveInteger(value, fallback) { + const parsed = Number(value); + return Number.isInteger(parsed) && parsed > 0 ? parsed : fallback; +} + +function emit(stream, level, event, fields = {}) { + stream.write(`${JSON.stringify({ level, event, sessionId, generation: instanceToken, ...fields })}\n`); +} + +function parseJson(text, context) { + try { + return JSON.parse(text); + } catch (error) { + throw new Error(`${context}: ${error.message}`); + } +} + +function runHerdr(args) { + const result = spawnSync("herdr", args, { + encoding: "utf8", + timeout: 15_000, + killSignal: "SIGKILL", + }); + if (result.error || result.status !== 0) { + const stderr = typeof result.stderr === "string" ? result.stderr.trim() : ""; + throw new Error(stderr || result.error?.message || `herdr ${args.join(" ")} exited ${result.status}`); + } + return result.stdout; +} + +export function nextCompactionAction(state, options) { + const { usedPercent, agentStatus, now, retryMs, maxAttempts } = options; + if (usedPercent !== undefined && usedPercent <= 60 && state.compactionAttempts > 0) { + state.compactionAttempts = 0; + state.lastCompactionRequestAt = undefined; + state.compactionBlocked = false; + return { type: "rearmed" }; + } + if (usedPercent === undefined || usedPercent < 75 || state.compactionBlocked) return { type: "none" }; + if (state.compactionAttempts >= maxAttempts) { + state.compactionBlocked = true; + return { type: "blocked" }; + } + if (agentStatus === "blocked") return { type: "deferred" }; + if (state.lastCompactionRequestAt !== undefined && now - state.lastCompactionRequestAt < retryMs) { + return { type: "none" }; + } + state.compactionAttempts += 1; + state.lastCompactionRequestAt = now; + return { type: "dispatch", attempt: state.compactionAttempts }; +} + +export function recordCompactionFailure(state, maxAttempts) { + if (state.compactionAttempts < maxAttempts || state.compactionBlocked) return false; + state.compactionBlocked = true; + return true; +} + +function selfTest() { + assert.deepEqual(parseContext("[79% ▮▮]"), { usedPercent: 79 }); + assert.equal(parseContext("90% context used").usedPercent, 90); + assert.equal(parseContext("9.3%/372k").usedPercent, 9.3); + assert.equal(parseContext("7% until auto-compact").untilAutoCompactPercent, 7); + assert.equal(extractPanes({ result: { panes: [{ pane_id: "w1:p1" }] } }).length, 1); + assert.deepEqual( + scopePanes( + [ + { pane_id: "w1:p1", workspace_id: "w1", tab_id: "w1:t1", label: "app-pi-impl" }, + { pane_id: "w1:p2", workspace_id: "w1", tab_id: "w1:t2", label: "app-pi-impl" }, + { pane_id: "w1:p3", workspace_id: "w1", tab_id: "w1:t1", label: "app-1234-pi-impl" }, + { + pane_id: "w1:p4", + workspace_id: "w1", + tab_id: "w1:t1", + label: "app-custom", + tokens: { fleet_key: "app", fleet_owner: "owner-1", fleet_kind: "worker" }, + }, + ], + "w1", + "w1:t1", + "app", + "owner-1", + ).map((pane) => pane.pane_id), + ["w1:p4"], + ); + assert.equal(parseContext("task output: 99% context used\n[79% ▮▮]").usedPercent, 79); + assert.equal(parseContext("90% context used\n[79% ▮▮]").usedPercent, 79); + assert.equal(parseContext("12% until auto-compact\n7% until auto-compact").untilAutoCompactPercent, 7); + assert.equal(parseContext("[79% ▮▮]\nnew prompt output").usedPercent, undefined); + assert.equal(positiveSeconds("0", 20), 20); + assert.equal(positiveSeconds("nope", 20), 20); + + const fixture = parseJson( + readFileSync(new URL("../fixtures/compaction-failures.json", import.meta.url), "utf8"), + "invalid compaction fixture", + ); + for (const testCase of fixture.cases) { + const state = { + compactionAttempts: 0, + lastCompactionRequestAt: undefined, + compactionBlocked: false, + ...testCase.initialState, + }; + let blockedEvents = 0; + testCase.samples.forEach((usedPercent, index) => { + const action = nextCompactionAction(state, { + usedPercent, + agentStatus: testCase.agentStatuses[index], + now: index * fixture.retryMs, + retryMs: fixture.retryMs, + maxAttempts: fixture.maxAttempts, + }); + if (action.type === "blocked") blockedEvents += 1; + if (action.type === "dispatch" && testCase.outcomes[index] === "fail") { + if (recordCompactionFailure(state, fixture.maxAttempts)) blockedEvents += 1; + } + }); + assert.equal(state.compactionAttempts, testCase.expectedAttempts, testCase.name); + assert.equal(blockedEvents, testCase.expectedBlockedEvents, testCase.name); + } + process.stdout.write(`${JSON.stringify({ status: "pass", checks: 18 })}\n`); +} + +if (process.argv.includes("--self-test")) { + selfTest(); + process.exit(0); +} + +const workspaceId = process.env.HERDR_WORKSPACE_ID; +const tabId = process.env.HERDR_TAB_ID; +const projectKey = option("--project-key"); +const ownerToken = option("--owner-token"); +const workspaceOption = option("--workspace-id"); +const tabOption = option("--tab-id"); +const instanceToken = option("--instance-token"); +const statusIntervalMs = positiveSeconds(option("--status-interval", "20"), 20) * 1000; +const contextIntervalMs = positiveSeconds(option("--context-interval", "300"), 300) * 1000; +const startupGraceMs = positiveSeconds(option("--startup-grace", "30"), 30) * 1000; +const maxCompactionAttempts = positiveInteger(option("--compaction-attempts", "3"), 3); + +if ( + !workspaceId || + !tabId || + !projectKey || + !ownerToken || + workspaceOption !== workspaceId || + tabOption !== tabId || + !instanceToken || + !/^[A-Za-z0-9-]+$/.test(instanceToken) +) { + process.stderr.write( + `${JSON.stringify({ + level: "fatal", + event: "invalid_scope", + sessionId: "fleet:unscoped", + message: "watch-fleet requires matching workspace/tab arguments plus project, owner, and instance tokens", + })}\n`, + ); + process.exit(2); +} + +const sessionId = `fleet:${workspaceId}:${tabId}:${projectKey}`; +const states = new Map(); +let consecutiveListFailures = 0; +let nextContextPoll = 0; +let stopping = false; + +for (const signal of ["SIGINT", "SIGTERM"]) { + process.on(signal, () => { + stopping = true; + emit(process.stderr, "info", "watcher_shutdown_requested", { signal }); + }); +} + +function validateCallerScope() { + const payload = parseJson(runHerdr(["pane", "current", "--current"]), "invalid current-pane response"); + const pane = payload?.result?.pane ?? payload?.pane ?? payload?.result ?? payload; + return pane.workspace_id === workspaceId && pane.tab_id === tabId && pane.pane_id === process.env.HERDR_PANE_ID; +} + +function listScopedPanes() { + const payload = parseJson(runHerdr(["pane", "list", "--workspace", workspaceId]), "invalid pane-list response"); + return scopePanes(extractPanes(payload), workspaceId, tabId, projectKey, ownerToken); +} + +async function discoverInitialPanes() { + const deadline = Date.now() + startupGraceMs; + let message = "no owned worker panes discovered"; + while (!stopping && Date.now() < deadline) { + try { + if (!validateCallerScope()) throw new Error("injected workspace, tab, and pane do not match"); + const panes = listScopedPanes(); + if (panes.length > 0) return panes; + } catch (error) { + message = error.message; + } + await new Promise((resolve) => setTimeout(resolve, 1000)); + } + emit(process.stderr, "fatal", "invalid_scope", { message }); + process.exitCode = 2; + return undefined; +} + +function emitCompactionBlocked(pane, state) { + emit(process.stdout, "error", "pane_monitor_blocked", { + paneId: pane.pane_id, + reason: "compaction_unconfirmed", + attempts: state.compactionAttempts, + }); +} + +function dispatchCompaction(pane, state, usedPercent, attempt) { + try { + runHerdr(["pane", "run", pane.pane_id, "/compact"]); + emit(process.stdout, "info", "compaction_requested", { paneId: pane.pane_id, usedPercent, attempt }); + } catch (error) { + emit(process.stderr, "error", "compaction_request_failed", { + paneId: pane.pane_id, + attempt, + message: error.message, + }); + if (recordCompactionFailure(state, maxCompactionAttempts)) emitCompactionBlocked(pane, state); + } +} + +function pollContext(pane, state) { + let output; + try { + output = runHerdr(["pane", "read", pane.pane_id, "--source", "visible", "--lines", "80"]); + state.contextFailures = 0; + } catch (error) { + state.contextFailures += 1; + emit(process.stderr, "warn", "context_read_failed", { + paneId: pane.pane_id, + failures: state.contextFailures, + message: error.message, + }); + if (state.contextFailures >= 3) { + emit(process.stdout, "error", "pane_monitor_blocked", { paneId: pane.pane_id }); + } + return; + } + + const context = parseContext(output); + const action = nextCompactionAction(state, { + usedPercent: context.usedPercent, + agentStatus: pane.agent_status, + now: Date.now(), + retryMs: Math.max(contextIntervalMs, 60_000), + maxAttempts: maxCompactionAttempts, + }); + if (action.type === "rearmed") { + emit(process.stdout, "info", "compaction_rearmed", { paneId: pane.pane_id }); + } else if (action.type === "deferred") { + emit(process.stdout, "warn", "compaction_deferred_blocked_dialog", { + paneId: pane.pane_id, + usedPercent: context.usedPercent, + }); + } else if (action.type === "blocked") { + emitCompactionBlocked(pane, state); + } else if (action.type === "dispatch") { + dispatchCompaction(pane, state, context.usedPercent, action.attempt); + } + + const lowRemaining = context.untilAutoCompactPercent !== undefined && context.untilAutoCompactPercent <= 8; + if (lowRemaining && !state.autoCompactWarning) { + state.autoCompactWarning = true; + emit(process.stdout, "warn", "auto_compact_near", { + paneId: pane.pane_id, + remainingPercent: context.untilAutoCompactPercent, + }); + } else if (!lowRemaining) { + state.autoCompactWarning = false; + } +} + +async function main() { + const initialPanes = await discoverInitialPanes(); + if (!initialPanes) return; + + emit(process.stderr, "info", "watcher_started", { + workspaceId, + tabId, + projectKey, + ownerToken, + statusIntervalMs, + contextIntervalMs, + startupGraceMs, + maxCompactionAttempts, + initialPaneCount: initialPanes.length, + }); + + while (!stopping) { + let panes; + try { + panes = listScopedPanes(); + consecutiveListFailures = 0; + } catch (error) { + consecutiveListFailures += 1; + emit(process.stderr, "error", "pane_list_failed", { + failures: consecutiveListFailures, + message: error.message, + }); + if (consecutiveListFailures >= 3) { + emit(process.stderr, "fatal", "watcher_stopped_after_failures"); + process.exitCode = 1; + break; + } + await new Promise((resolve) => setTimeout(resolve, statusIntervalMs)); + continue; + } + + const liveIds = new Set(panes.map((pane) => pane.pane_id)); + for (const [paneId] of states) { + if (!liveIds.has(paneId)) { + states.delete(paneId); + emit(process.stdout, "info", "pane_removed", { paneId }); + } + } + + const contextDue = Date.now() >= nextContextPoll; + for (const pane of panes) { + const state = states.get(pane.pane_id) ?? { + status: undefined, + compactionAttempts: 0, + lastCompactionRequestAt: undefined, + compactionBlocked: false, + autoCompactWarning: false, + contextFailures: 0, + }; + const status = STATUS_VALUES.has(pane.agent_status) ? pane.agent_status : "unknown"; + if (state.status !== status) { + emit(process.stdout, "info", "pane_status_changed", { + paneId: pane.pane_id, + label: pane.label, + previous: state.status, + status, + }); + state.status = status; + } + states.set(pane.pane_id, state); + if (contextDue) pollContext({ ...pane, agent_status: status }, state); + } + + if (contextDue) nextContextPoll = Date.now() + contextIntervalMs; + await new Promise((resolve) => setTimeout(resolve, statusIntervalMs)); + } + + emit(process.stderr, "info", "watcher_stopped"); +} + +await main();