From 7316625a4d6abb51d9c606123e33fa05ab5e8bb6 Mon Sep 17 00:00:00 2001 From: John McChesney TenEyck Jr <59268465+jmcte@users.noreply.github.com> Date: Tue, 14 Jul 2026 09:23:54 +0100 Subject: [PATCH] Harden Flow inspector safety Fail closed on incomplete check evidence, reconcile mutually exclusive labels, and keep all mutations behind explicit flags. Add atomic locked assignment updates, portable configuration, table-driven coverage, and operator guidance. Closes #24. Signed-off-by: John McChesney TenEyck Jr <59268465+jmcte@users.noreply.github.com> --- .github/workflows/ci.yml | 2 +- README.md | 13 +- docs/flow-inspector.md | 85 ++++ scripts/flow/inspect_repo_flow.py | 646 +++++++++++------------------- scripts/flow/inspector_core.py | 368 +++++++++++++++++ tests/test_flow_inspector.py | 311 ++++++++++++++ 6 files changed, 1009 insertions(+), 416 deletions(-) create mode 100644 docs/flow-inspector.md create mode 100644 scripts/flow/inspector_core.py create mode 100644 tests/test_flow_inspector.py diff --git a/.github/workflows/ci.yml b/.github/workflows/ci.yml index 35fce44..ac85cf8 100644 --- a/.github/workflows/ci.yml +++ b/.github/workflows/ci.yml @@ -43,4 +43,4 @@ jobs: run: python -m unittest discover -s tests -v - name: Compile flow inspector - run: python -m py_compile scripts/flow/inspect_repo_flow.py + run: python -m py_compile scripts/flow/inspector_core.py scripts/flow/inspect_repo_flow.py diff --git a/README.md b/README.md index 2e7c9f8..9c938b6 100644 --- a/README.md +++ b/README.md @@ -21,7 +21,7 @@ For the high-level relationship between `flow`, `bootstrap`, `.github`, individu - [`schemas/`](schemas/) - issue and PR contract schemas. - [`github/`](github/) - canonical labels, issue templates, project fields, and PR template inputs. - [`dashboards/`](dashboards/) - saved query patterns for operational review. -- [`scripts/flow/inspect_repo_flow.py`](scripts/flow/inspect_repo_flow.py) - local inspector for repo flow metadata. +- [`scripts/flow/inspect_repo_flow.py`](scripts/flow/inspect_repo_flow.py) - read-only-by-default inspector for repo flow metadata; see the [operator guide](docs/flow-inspector.md). - [`docs/autonomous-flow-platform.md`](docs/autonomous-flow-platform.md) - design rationale and rollout plan. - [`docs/omt-global-operating-map.md`](docs/omt-global-operating-map.md) - org-level ownership map. @@ -52,15 +52,20 @@ Open PR clearance has priority over starting new implementation work: ```sh python -m json.tool schemas/issue-contract.schema.json >/dev/null python -m json.tool schemas/pr-contract.schema.json >/dev/null -python -m py_compile scripts/flow/inspect_repo_flow.py +python -m py_compile scripts/flow/inspector_core.py scripts/flow/inspect_repo_flow.py +python -m unittest discover -s tests -v ``` -Run the repo CI gate before opening a PR: +Inspect a repository without mutating GitHub or local assignment state: ```sh -bash scripts/ci/run-fast-checks.sh +python scripts/flow/inspect_repo_flow.py --repo OMT-Global/flow --issues ``` +The inspector requires explicit mutation flags and explicit output/assignment +paths. Review the proposed writes in the default report before following the +[apply examples](docs/flow-inspector.md#apply-reviewed-writes). + ## Bootstrap Relationship `flow` defines the operating protocol. `bootstrap` projects the relevant repo-local pieces into OMT-Global repositories through `project.bootstrap.yaml`, managed templates, labels, workflows, and GitHub policy. Update policy here first, then let bootstrap reconcile downstream repos deliberately. diff --git a/docs/flow-inspector.md b/docs/flow-inspector.md new file mode 100644 index 0000000..3089124 --- /dev/null +++ b/docs/flow-inspector.md @@ -0,0 +1,85 @@ +# Flow inspector operator guide + +The Flow inspector classifies live GitHub issues and pull requests, chooses the +next actor, and reports proposed state, lane, and repair-assignment changes. Its +default mode is read-only: it does not edit GitHub labels or assignment state, +and it does not write a report unless an output path is supplied. + +## Inspect without mutation + +```sh +python scripts/flow/inspect_repo_flow.py \ + --repo OMT-Global/flow \ + --issues +``` + +Use `--json` for the complete report on stdout or configure an explicit, +atomic file output: + +```sh +python scripts/flow/inspect_repo_flow.py \ + --repo OMT-Global/flow \ + --issues \ + --json-out /tmp/flow-inspection.json +``` + +Each report says `mode: read-only` and includes `proposedWrites`. A check +rollup is green only when at least one check is present and every observed check +completed successfully. Empty, pending, neutral, skipped, cancelled, malformed, +or unknown evidence cannot produce a green classification. + +If the host has a maintenance gate, configure it explicitly: + +```sh +python scripts/flow/inspect_repo_flow.py \ + --repo OMT-Global/flow \ + --maintenance-gate /opt/omt/check_maintenance_gate.sh +``` + +No repository, workspace, report directory, assignment store, or maintenance +script path is built into the inspector. + +## Apply reviewed writes + +These commands mutate external or local state. First run the matching read-only +inspection and review `proposedWrites`. + +Reconcile GitHub state and lane labels: + +```sh +python scripts/flow/inspect_repo_flow.py \ + --repo OMT-Global/flow \ + --issues \ + --apply-labels \ + --label-limit 10 +``` + +Reconciliation adds the desired state/lane and removes every stale label in the +same mutually exclusive `state:` or `lane:` family. Other label families are +left untouched. + +Dispatch repair candidates to an explicit assignment store: + +```sh +python scripts/flow/inspect_repo_flow.py \ + --repo OMT-Global/flow \ + --dispatch-repairs \ + --dispatch-limit 5 \ + --assignments /var/lib/omt/pr-repair-assignments.json +``` + +Assignment updates hold an exclusive sidecar lock for the read/modify/write +transaction, write a temporary file in the destination directory, flush it, +and atomically replace the destination. Existing repository/PR assignments are +deduplicated while the lock is held. + +## Verification + +```sh +python -m py_compile scripts/flow/inspector_core.py scripts/flow/inspect_repo_flow.py +python -m unittest discover -s tests -p 'test_flow_inspector.py' -v +python scripts/flow/inspect_repo_flow.py --help +``` + +Never use `--apply-labels` or `--dispatch-repairs` for a smoke test. Live +mutation testing is a human decision point and must be explicitly authorized. diff --git a/scripts/flow/inspect_repo_flow.py b/scripts/flow/inspect_repo_flow.py index fe73b74..db1edef 100755 --- a/scripts/flow/inspect_repo_flow.py +++ b/scripts/flow/inspect_repo_flow.py @@ -1,464 +1,288 @@ #!/usr/bin/env python3 -"""Read-only OMT-Global flow inspector. - -Classifies GitHub issues and PRs into the flow protocol without mutating GitHub. -""" +"""Inspect Flow state safely; GitHub and assignment writes require explicit flags.""" from __future__ import annotations import argparse +import fcntl import json +import os import subprocess -import sys +import tempfile from collections import Counter from datetime import datetime, timezone from pathlib import Path from typing import Any -WORKSPACE = Path('/home/pheidon/.openclaw/workspace') -GATE = WORKSPACE / 'skills/openclaw-maintenance/scripts/check_maintenance_gate.sh' -DEFAULT_REPORT_DIR = WORKSPACE / 'state/flow-inspections' -DEFAULT_ASSIGNMENTS = WORKSPACE / 'state/pr-repair-assignments.json' - -STATE_LABEL_BY_FLOW = { - 'Intake': 'state:intake', - 'Ready for Planning': 'state:ready-for-planning', - 'Ready for Implementation': 'state:ready-for-implementation', - 'Implementing': 'state:implementing', - 'PR Open': 'state:needs-review', - 'Needs Review': 'state:needs-review', - 'Needs Repair': 'state:needs-repair', - 'Repairing': 'state:repairing', - 'Ready for Approval': 'state:ready-for-approval', - 'Approved / Waiting Checks': 'state:waiting-checks', - 'Auto-merge Armed': 'state:auto-merge-armed', - 'Blocked - Human': 'state:blocked-human', - 'Blocked - Infrastructure': 'state:blocked-infra', - 'Blocked - Scope': 'state:blocked-scope', - 'Paused': 'state:paused', -} - -LANE_LABEL_BY_ACTOR = { - 'Pheidon': 'lane:pheidon', - 'Apollo': 'lane:apollo', - 'Ares': 'lane:ares', - 'Daedalus': 'lane:daedalus', - 'Hephaestus': 'lane:hephaestus', - 'Hermes': 'lane:hermes', -} - -LABEL_SPECS = { - 'state:intake': ('d9d9d9', 'Captured but not yet planned.'), - 'state:ready-for-planning': ('ccebc5', 'Ready for Apollo/Pheidon planning refinement.'), - 'state:ready-for-implementation': ('bc80bd', 'Issue has enough contract to assign implementation.'), - 'state:implementing': ('80b1d3', 'Worker lane is actively implementing.'), - 'state:needs-review': ('ffffb3', 'PR needs review.'), - 'state:needs-repair': ('fb8072', 'PR or issue needs repair before it can advance.'), - 'state:repairing': ('fdb462', 'Repair is actively assigned.'), - 'state:ready-for-approval': ('b3de69', 'Pheidon/gate approval is the next action.'), - 'state:waiting-checks': ('ffffb3', 'Approved or ready but waiting on checks/merge queue.'), - 'state:auto-merge-armed': ('b3de69', 'Auto-merge is enabled and GitHub gates own completion.'), - 'state:blocked-human': ('e41a1c', 'Human decision required.'), - 'state:blocked-infra': ('984ea3', 'Blocked by tool, auth, runner, or infrastructure failure.'), - 'state:blocked-scope': ('ff7f00', 'Blocked by unclear scope or acceptance criteria.'), - 'state:paused': ('999999', 'Intentionally paused.'), - 'lane:pheidon': ('b3de69', 'Orchestration, gate, governance, and controller action.'), - 'lane:apollo': ('8dd3c7', 'Scope, backlog, synthesis, and issue-contract work.'), - 'lane:ares': ('fb8072', 'Validation, adversarial review, and test-pressure work.'), - 'lane:daedalus': ('80b1d3', 'Implementation and substantive code repair work.'), - 'lane:hephaestus': ('fdb462', 'CI, build, lockfile, mergeability, and artifact work.'), - 'lane:hermes': ('bebada', 'macOS/platform-native or special execution work.'), -} - -LANE_BY_AUTHOR = { - 'apollo-omt': 'Apollo', - 'ares-omt': 'Ares', - 'daedalus-omt': 'Daedalus', - 'hephaestus-omt': 'Hephaestus', - 'hermes-omt': 'Hermes', - 'pheidon-omt': 'Pheidon', - 'athena-omt': 'Ares', - 'athena': 'Ares', - 'chatgpt-codex-connector': 'Daedalus', -} +from inspector_core import ( + LABEL_SPECS, + classify_issue, + classify_pr, + default_assignments, + plan_label_changes, + plan_repair_dispatch, +) def run(cmd: list[str], *, check: bool = True) -> subprocess.CompletedProcess[str]: - cp = subprocess.run(cmd, text=True, capture_output=True) - if check and cp.returncode: - raise RuntimeError(cp.stderr.strip() or cp.stdout.strip() or f'command failed: {cmd!r}') - return cp + completed = subprocess.run(cmd, text=True, capture_output=True) + if check and completed.returncode: + raise RuntimeError(completed.stderr.strip() or completed.stdout.strip() or f"command failed: {cmd!r}") + return completed def gh_json(args: list[str]) -> Any: - return json.loads(run(['gh', *args]).stdout) - - -def maintenance_gate() -> None: - cp = run(['bash', str(GATE), 'flow-inspect', 'inspect_repo_flow', 'Maintenance active, queued flow inspection.'], check=False) - if cp.returncode: - print(cp.stdout or cp.stderr, end='') - sys.exit(10) - - -def now() -> str: - return datetime.now(timezone.utc).strftime('%Y-%m-%dT%H:%M:%SZ') - - -def label_names(item: dict[str, Any]) -> set[str]: - return {l.get('name', '') for l in item.get('labels') or []} - - -def first_label_value(labels: set[str], prefix: str) -> str | None: - for label in sorted(labels): - if label.startswith(prefix): - return label.split(':', 1)[1] - return None - - -def check_summary(rollup: list[dict[str, Any]]) -> dict[str, Any]: - failing = [] - pending = [] - passing = [] - for c in rollup or []: - name = c.get('name') or c.get('workflowName') or c.get('context') or 'check' - status = c.get('status') - conclusion = c.get('conclusion') - if status and status != 'COMPLETED': - pending.append(name) - elif conclusion in ('SUCCESS', 'SKIPPED', 'NEUTRAL'): - passing.append(name) - elif conclusion: - failing.append(name) - return { - 'green': not failing and not pending, - 'failing': failing, - 'pending': pending, - 'passing': passing, - } - + return json.loads(run(["gh", *args]).stdout) -def active_change_requests(pr: dict[str, Any]) -> list[dict[str, str]]: - out = [] - for review in pr.get('latestReviews') or []: - if review.get('state') == 'CHANGES_REQUESTED': - out.append({ - 'author': (review.get('author') or {}).get('login') or 'unknown', - 'submittedAt': review.get('submittedAt') or '', - 'body': (review.get('body') or '').strip()[:1000], - }) - return out - -def owner_from_author(pr: dict[str, Any]) -> str: - login = ((pr.get('author') or {}).get('login') or '').lower() - return LANE_BY_AUTHOR.get(login, 'Pheidon') - - -def repair_actor_for_pr(owner_lane: str, checks: dict[str, Any], merge_state: str) -> tuple[str, str]: - failing = {str(name).lower() for name in checks.get('failing') or []} - metadata_markers = ( - 'validate pr description', - 'ci gate', - 'validate secrets', - 'detect relevant changes', - 'attest', +def maintenance_gate(path: Path) -> None: + completed = run( + ["bash", str(path), "flow-inspect", "inspect_repo_flow", "Maintenance active, queued flow inspection."], + check=False, ) - if merge_state in ('BEHIND', 'DIRTY'): - return 'Hephaestus', f'Restore mergeability from {merge_state.lower()} state, rerun required checks, and push with --force-with-lease when rebasing.' - if any(any(marker in name for marker in metadata_markers) for name in failing): - return 'Hephaestus', 'Repair PR metadata, generated governance surfaces, or CI gate wiring, then rerun required checks.' - if owner_lane in ('Pheidon', 'Apollo'): - return 'Daedalus', 'Reproduce failing checks, make the minimal substantive repair, validate, and push.' - return owner_lane, 'Reproduce failing checks, make the minimal repair, validate, and push.' - - -def classify_pr(pr: dict[str, Any]) -> dict[str, Any]: - checks = check_summary(pr.get('statusCheckRollup') or []) - labels = label_names(pr) - merge_state = pr.get('mergeStateStatus') or 'UNKNOWN' - review = pr.get('reviewDecision') or 'UNKNOWN' - changes = active_change_requests(pr) - owner = first_label_value(labels, 'lane:') - owner_lane = owner.title() if owner else owner_from_author(pr) - next_actor = owner_lane - reasons: list[str] = [] - - if pr.get('isDraft'): - state = 'Paused' - next_action = 'Undraft or explicitly mark as paused/not planned.' - reasons.append('draft') - elif merge_state == 'CLEAN' and review == 'APPROVED' and checks['green']: - if pr.get('autoMergeRequest'): - state = 'Auto-merge Armed' - next_action = 'Wait for GitHub auto-merge or merge queue completion.' - reasons.append('clean_approved_green') - else: - state = 'Needs Repair' - next_actor = owner_lane - next_action = 'PR author should enable auto-merge where GitHub allows it, or record the unavailable/unsafe reason under Merge Automation.' - reasons.extend(['clean_approved_green', 'auto_merge_missing']) - elif changes or review == 'CHANGES_REQUESTED': - state = 'Needs Repair' - next_actor = owner_lane if owner_lane not in ('Pheidon', 'Apollo') else 'Daedalus' - next_action = 'Convert requested changes into a repair task and push an updated PR head.' - reasons.append('changes_requested') - elif merge_state in ('DIRTY', 'BEHIND'): - state = 'Needs Repair' - next_actor, next_action = repair_actor_for_pr(owner_lane, checks, merge_state) - reasons.append(f'merge_state:{merge_state.lower()}') - if checks['failing']: - reasons.append('failing_checks') - elif checks['failing']: - state = 'Needs Repair' - next_actor, next_action = repair_actor_for_pr(owner_lane, checks, merge_state) - reasons.append('failing_checks') - elif review == 'REVIEW_REQUIRED': - state = 'Needs Review' - next_actor = 'Ares' - next_action = 'Perform non-author review and either approve or file exact repair feedback.' - reasons.append('review_required') - elif checks['pending']: - state = 'Approved / Waiting Checks' if review == 'APPROVED' else 'PR Open' - next_action = 'Wait for checks; reclassify on completion.' - reasons.append('checks_pending') - elif merge_state == 'BLOCKED': - state = 'Blocked - Infrastructure' - next_actor = 'Pheidon' - next_action = 'Inspect branch protection/reviews/checks and convert blocker into explicit next action.' - reasons.append('blocked') - else: - state = 'PR Open' - next_actor = 'Pheidon' - next_action = 'Inspect manually; classifier did not find a confident next state.' - reasons.append('unclassified') - - return { - 'kind': 'pr', - 'number': pr['number'], - 'title': pr['title'], - 'url': pr['url'], - 'flowState': state, - 'ownerLane': owner_lane, - 'nextActor': next_actor, - 'nextAction': next_action, - 'reasons': reasons, - 'mergeStateStatus': merge_state, - 'reviewDecision': review, - 'checks': checks, - 'autoMergeArmed': bool(pr.get('autoMergeRequest')), - 'changeRequests': changes, - 'updatedAt': pr.get('updatedAt'), - } - - -def classify_issue(issue: dict[str, Any]) -> dict[str, Any]: - labels = label_names(issue) - state_label = first_label_value(labels, 'state:') - lane_label = first_label_value(labels, 'lane:') - autonomy = first_label_value(labels, 'autonomy:') - body = issue.get('body') or '' - - state_map = { - 'intake': 'Intake', - 'ready-for-planning': 'Ready for Planning', - 'ready-for-implementation': 'Ready for Implementation', - 'implementing': 'Implementing', - 'needs-review': 'Needs Review', - 'needs-repair': 'Needs Repair', - 'repairing': 'Repairing', - 'blocked-human': 'Blocked - Human', - 'blocked-infra': 'Blocked - Infrastructure', - 'blocked-scope': 'Blocked - Scope', - 'paused': 'Paused', - } - flow_state = state_map.get(state_label or '', '') - if not flow_state: - has_contract = 'Acceptance criteria' in body or 'Validation' in body or 'validation' in body.lower() - flow_state = 'Ready for Implementation' if has_contract else 'Intake' - - owner_lane = lane_label.title() if lane_label else 'Pheidon' - if flow_state == 'Intake': - next_actor = 'Apollo' - next_action = 'Refine into an executable issue contract with acceptance criteria and validation.' - elif flow_state == 'Ready for Implementation': - next_actor = owner_lane if owner_lane != 'Pheidon' else 'Daedalus' - next_action = 'Assign to worker WIP slot when PR queue allows.' - elif flow_state.startswith('Blocked'): - next_actor = 'Pheidon' - next_action = 'Resolve blocker or convert it into a flow-blocker issue.' - else: - next_actor = owner_lane - next_action = 'Continue according to current state.' - - return { - 'kind': 'issue', - 'number': issue['number'], - 'title': issue['title'], - 'url': issue['url'], - 'flowState': flow_state, - 'ownerLane': owner_lane, - 'nextActor': next_actor, - 'nextAction': next_action, - 'autonomy': autonomy, - 'labels': sorted(labels), - 'updatedAt': issue.get('updatedAt'), - } + if completed.returncode: + print(completed.stdout or completed.stderr, end="") + raise SystemExit(10) -def desired_labels(item: dict[str, Any]) -> list[str]: - labels = [] - state = STATE_LABEL_BY_FLOW.get(item.get('flowState', '')) - lane = LANE_LABEL_BY_ACTOR.get(item.get('nextActor', '')) or LANE_LABEL_BY_ACTOR.get(item.get('ownerLane', '')) - if state: - labels.append(state) - if lane: - labels.append(lane) - return labels +def now() -> str: + return datetime.now(timezone.utc).strftime("%Y-%m-%dT%H:%M:%SZ") def ensure_label(repo: str, name: str) -> None: - color, description = LABEL_SPECS.get(name, ('d9d9d9', 'OMT-Global flow label.')) - cp = run(['gh', 'label', 'create', name, '--repo', repo, '--color', color, '--description', description], check=False) - if cp.returncode and 'already exists' not in (cp.stderr + cp.stdout): - # If it exists with different metadata, keep going; label add is what matters. - probe = run(['gh', 'label', 'list', '--repo', repo, '--search', name, '--json', 'name'], check=False) + color, description = LABEL_SPECS.get(name, ("d9d9d9", "OMT-Global flow label.")) + completed = run( + ["gh", "label", "create", name, "--repo", repo, "--color", color, "--description", description], + check=False, + ) + if completed.returncode and "already exists" not in (completed.stderr + completed.stdout): + probe = run(["gh", "label", "list", "--repo", repo, "--search", name, "--json", "name"], check=False) if probe.returncode or name not in probe.stdout: - raise RuntimeError(cp.stderr.strip() or cp.stdout.strip()) + raise RuntimeError(completed.stderr.strip() or completed.stdout.strip()) -def apply_labels(repo: str, items: list[dict[str, Any]], *, max_items: int | None = None) -> dict[str, Any]: - applied = [] - selected = items if max_items is None else items[:max_items] - needed = sorted({label for item in selected for label in desired_labels(item)}) - for label in needed: +def apply_labels( + repo: str, + items: list[dict[str, Any]], + *, + max_items: int | None = None, +) -> dict[str, Any]: + changes = plan_label_changes(items, max_items=max_items) + for label in sorted({label for change in changes for label in change["add"]}): ensure_label(repo, label) - for item in selected: - labels = desired_labels(item) - if not labels: - continue - target = ['pr' if item['kind'] == 'pr' else 'issue', 'edit', str(item['number']), '--repo', repo, '--add-label', ','.join(labels)] - run(['gh', *target]) - applied.append({'kind': item['kind'], 'number': item['number'], 'labels': labels}) - return {'count': len(applied), 'items': applied} + for change in changes: + command = [ + "gh", + "pr" if change["kind"] == "pr" else "issue", + "edit", + str(change["number"]), + "--repo", + repo, + ] + if change["add"]: + command.extend(["--add-label", ",".join(change["add"])]) + if change["remove"]: + command.extend(["--remove-label", ",".join(change["remove"])]) + run(command) + return {"count": len(changes), "items": changes} def load_assignments(path: Path) -> dict[str, Any]: if path.exists(): - return json.loads(path.read_text()) - return {lane: {'current': None, 'queued': []} for lane in ('apollo', 'ares', 'daedalus', 'hephaestus', 'hermes')} - - -def dispatch_repairs(repo: str, prs: list[dict[str, Any]], path: Path, *, max_items: int) -> dict[str, Any]: - assignments = load_assignments(path) - existing = set() - for bucket in assignments.values(): - current = bucket.get('current') if isinstance(bucket, dict) else None - if current: - existing.add((current.get('repo'), current.get('number'))) - for queued in (bucket.get('queued') or []) if isinstance(bucket, dict) else []: - existing.add((queued.get('repo'), queued.get('number'))) - - added = [] - for item in prs: - if len(added) >= max_items: - break - if item.get('flowState') != 'Needs Repair': - continue - actor = item.get('nextActor') - lane = (actor or '').lower() - if lane not in ('apollo', 'ares', 'daedalus', 'hephaestus', 'hermes'): - lane = 'daedalus' - key = (repo, item['number']) - if key in existing: - continue - payload = { - 'repo': repo, - 'number': item['number'], - 'title': item['title'], - 'url': item['url'], - 'flowState': item['flowState'], - 'nextActor': item['nextActor'], - 'nextAction': item['nextAction'], - 'reasons': item.get('reasons', []), - 'mergeStateStatus': item.get('mergeStateStatus'), - 'reviewDecision': item.get('reviewDecision'), - 'failingChecks': item.get('checks', {}).get('failing', []), - 'pendingChecks': item.get('checks', {}).get('pending', []), - 'coordinationRoutine': 'flow_inspector_dispatch', - } - bucket = assignments.setdefault(lane, {'current': None, 'queued': []}) - if bucket.get('current') is None: - bucket['current'] = payload - else: - bucket.setdefault('queued', []).append(payload) - existing.add(key) - added.append({'lane': lane, 'number': item['number'], 'title': item['title']}) + parsed = json.loads(path.read_text()) + if not isinstance(parsed, dict): + raise ValueError(f"{path} must contain a JSON object") + return parsed + return default_assignments() + + +def atomic_write_json(path: Path, value: dict[str, Any]) -> None: + path.parent.mkdir(parents=True, exist_ok=True) + temporary_path: Path | None = None + try: + with tempfile.NamedTemporaryFile( + mode="w", + encoding="utf-8", + dir=path.parent, + prefix=f".{path.name}.", + suffix=".tmp", + delete=False, + ) as handle: + temporary_path = Path(handle.name) + json.dump(value, handle, indent=2, ensure_ascii=False) + handle.write("\n") + handle.flush() + os.fsync(handle.fileno()) + os.replace(temporary_path, path) + directory_fd = os.open(path.parent, os.O_RDONLY) + try: + os.fsync(directory_fd) + finally: + os.close(directory_fd) + finally: + if temporary_path is not None and temporary_path.exists(): + temporary_path.unlink() + + +def dispatch_repairs( + repo: str, + prs: list[dict[str, Any]], + path: Path, + *, + max_items: int, +) -> dict[str, Any]: path.parent.mkdir(parents=True, exist_ok=True) - path.write_text(json.dumps(assignments, indent=2) + '\n') - return {'count': len(added), 'path': str(path), 'items': added} - - -def main() -> None: - ap = argparse.ArgumentParser() - ap.add_argument('--repo', default='OMT-Global/axiom') - ap.add_argument('--limit', type=int, default=100) - ap.add_argument('--json-out', type=Path) - ap.add_argument('--issues', action='store_true', help='include open issues as well as PRs') - ap.add_argument('--apply-labels', action='store_true', help='write flow state/lane labels to GitHub') - ap.add_argument('--label-limit', type=int, help='maximum items to label when --apply-labels is set') - ap.add_argument('--dispatch-repairs', action='store_true', help='write Needs Repair PRs to local repair assignment state') - ap.add_argument('--dispatch-limit', type=int, default=10) - ap.add_argument('--assignments', type=Path, default=DEFAULT_ASSIGNMENTS) - args = ap.parse_args() - - maintenance_gate() - - prs = gh_json(['pr', 'list', '--repo', args.repo, '--state', 'open', '--limit', str(args.limit), '--json', - 'number,title,url,isDraft,author,headRefName,mergeStateStatus,reviewDecision,statusCheckRollup,autoMergeRequest,latestReviews,labels,updatedAt']) + lock_path = path.with_name(f"{path.name}.lock") + with lock_path.open("a+", encoding="utf-8") as lock: + fcntl.flock(lock.fileno(), fcntl.LOCK_EX) + try: + assignments = load_assignments(path) + updated, added = plan_repair_dispatch(repo, prs, assignments, max_items=max_items) + if added: + atomic_write_json(path, updated) + finally: + fcntl.flock(lock.fileno(), fcntl.LOCK_UN) + return {"count": len(added), "path": str(path), "items": added} + + +def parse_args(argv: list[str] | None = None) -> argparse.Namespace: + parser = argparse.ArgumentParser( + description=( + "Inspect open GitHub work and report proposed Flow state changes. " + "The default mode is read-only; mutation requires --apply-labels or --dispatch-repairs." + ) + ) + parser.add_argument("--repo", required=True, help="GitHub repository in OWNER/NAME form") + parser.add_argument("--limit", type=int, default=100) + parser.add_argument("--json", action="store_true", help="print the complete JSON report to stdout") + parser.add_argument("--json-out", type=Path, help="write the report atomically to this explicit path") + parser.add_argument("--issues", action="store_true", help="include open issues as well as pull requests") + parser.add_argument( + "--maintenance-gate", + type=Path, + help="run this explicitly configured maintenance-gate script before inspection", + ) + parser.add_argument( + "--apply-labels", + action="store_true", + help="MUTATING: reconcile Flow state/lane labels on GitHub", + ) + parser.add_argument("--label-limit", type=int, help="maximum items to mutate with --apply-labels") + parser.add_argument( + "--dispatch-repairs", + action="store_true", + help="MUTATING: write Needs Repair PRs to the configured assignment store", + ) + parser.add_argument("--dispatch-limit", type=int, default=10) + parser.add_argument( + "--assignments", + type=Path, + help="explicit repair assignment JSON path; required by --dispatch-repairs", + ) + args = parser.parse_args(argv) + if args.dispatch_repairs and args.assignments is None: + parser.error("--dispatch-repairs requires --assignments") + return args + + +def main(argv: list[str] | None = None) -> None: + args = parse_args(argv) + if args.maintenance_gate: + maintenance_gate(args.maintenance_gate) + + prs = gh_json( + [ + "pr", + "list", + "--repo", + args.repo, + "--state", + "open", + "--limit", + str(args.limit), + "--json", + "number,title,url,isDraft,author,headRefName,mergeStateStatus,reviewDecision,statusCheckRollup,autoMergeRequest,latestReviews,labels,updatedAt", + ] + ) pr_items = [classify_pr(pr) for pr in prs] issue_items: list[dict[str, Any]] = [] if args.issues: - issues = gh_json(['issue', 'list', '--repo', args.repo, '--state', 'open', '--limit', str(args.limit), '--json', - 'number,title,url,labels,assignees,updatedAt,body']) + issues = gh_json( + [ + "issue", + "list", + "--repo", + args.repo, + "--state", + "open", + "--limit", + str(args.limit), + "--json", + "number,title,url,labels,assignees,updatedAt,body", + ] + ) issue_items = [classify_issue(issue) for issue in issues] - counts = Counter(item['flowState'] for item in [*pr_items, *issue_items]) - sorted_prs = sorted(pr_items, key=lambda x: x['number'], reverse=True) - sorted_issues = sorted(issue_items, key=lambda x: x['number'], reverse=True) + sorted_prs = sorted(pr_items, key=lambda item: item["number"], reverse=True) + sorted_issues = sorted(issue_items, key=lambda item: item["number"], reverse=True) + all_items = [*sorted_prs, *sorted_issues] + counts = Counter(item["flowState"] for item in all_items) + proposed_assignments = load_assignments(args.assignments) if args.assignments else default_assignments() + _, proposed_dispatches = plan_repair_dispatch( + args.repo, + sorted_prs, + proposed_assignments, + max_items=args.dispatch_limit, + ) + proposed_writes = { + "labels": plan_label_changes(all_items, max_items=args.label_limit), + "dispatch": proposed_dispatches, + } + write_results = {} if args.apply_labels: - write_results['labels'] = apply_labels(args.repo, [*sorted_prs, *sorted_issues], max_items=args.label_limit) + write_results["labels"] = apply_labels(args.repo, all_items, max_items=args.label_limit) if args.dispatch_repairs: - write_results['dispatch'] = dispatch_repairs(args.repo, sorted_prs, args.assignments, max_items=args.dispatch_limit) + write_results["dispatch"] = dispatch_repairs( + args.repo, + sorted_prs, + args.assignments, + max_items=args.dispatch_limit, + ) report = { - 'repo': args.repo, - 'generatedAt': now(), - 'counts': dict(sorted(counts.items())), - 'writeResults': write_results, - 'prs': sorted_prs, - 'issues': sorted_issues, + "repo": args.repo, + "generatedAt": now(), + "mode": "apply" if write_results else "read-only", + "counts": dict(sorted(counts.items())), + "proposedWrites": proposed_writes, + "writeResults": write_results, + "prs": sorted_prs, + "issues": sorted_issues, } + if args.json_out: + atomic_write_json(args.json_out, report) - out = args.json_out - if out is None: - safe_repo = args.repo.replace('/', '-') - DEFAULT_REPORT_DIR.mkdir(parents=True, exist_ok=True) - out = DEFAULT_REPORT_DIR / f'{safe_repo}-{datetime.now(timezone.utc).strftime("%Y%m%dT%H%M%SZ")}.json' - out.parent.mkdir(parents=True, exist_ok=True) - out.write_text(json.dumps(report, indent=2, ensure_ascii=False) + '\n') + if args.json: + print(json.dumps(report, indent=2, ensure_ascii=False)) + return - print(f'Flow inspection for {args.repo}') + print(f"Flow inspection for {args.repo} (mode: {report['mode']})") for state, count in sorted(counts.items()): - print(f'- {state}: {count}') - print('\nTop PR actions:') - for item in report['prs'][:20]: - print(f"- PR #{item['number']}: {item['flowState']} -> {item['nextActor']}: {item['nextAction']} ({', '.join(item['reasons'])})") + print(f"- {state}: {count}") + print("\nTop PR actions:") + for item in sorted_prs[:20]: + print( + f"- PR #{item['number']}: {item['flowState']} -> {item['nextActor']}: " + f"{item['nextAction']} ({', '.join(item['reasons'])})" + ) + print("\nProposed writes (not applied unless a MUTATING flag was supplied):") + print(json.dumps(proposed_writes, indent=2)) if write_results: - print('\nWrites:') + print("\nApplied writes:") print(json.dumps(write_results, indent=2)) - print(f'\nwrote {out}') + if args.json_out: + print(f"\nwrote {args.json_out}") -if __name__ == '__main__': +if __name__ == "__main__": main() diff --git a/scripts/flow/inspector_core.py b/scripts/flow/inspector_core.py new file mode 100644 index 0000000..e4e9b43 --- /dev/null +++ b/scripts/flow/inspector_core.py @@ -0,0 +1,368 @@ +"""Pure classification and mutation-planning logic for the Flow inspector.""" +from __future__ import annotations + +from copy import deepcopy +from typing import Any + + +STATE_LABEL_BY_FLOW = { + "Intake": "state:intake", + "Ready for Planning": "state:ready-for-planning", + "Ready for Implementation": "state:ready-for-implementation", + "Implementing": "state:implementing", + "PR Open": "state:needs-review", + "Needs Review": "state:needs-review", + "Needs Repair": "state:needs-repair", + "Repairing": "state:repairing", + "Ready for Approval": "state:ready-for-approval", + "Approved / Waiting Checks": "state:waiting-checks", + "Auto-merge Armed": "state:auto-merge-armed", + "Blocked - Human": "state:blocked-human", + "Blocked - Infrastructure": "state:blocked-infra", + "Blocked - Scope": "state:blocked-scope", + "Paused": "state:paused", +} + +LANE_LABEL_BY_ACTOR = { + "Pheidon": "lane:pheidon", + "Apollo": "lane:apollo", + "Ares": "lane:ares", + "Daedalus": "lane:daedalus", + "Hephaestus": "lane:hephaestus", + "Hermes": "lane:hermes", +} + +LABEL_SPECS = { + "state:intake": ("d9d9d9", "Captured but not yet planned."), + "state:ready-for-planning": ("ccebc5", "Ready for Apollo/Pheidon planning refinement."), + "state:ready-for-implementation": ("bc80bd", "Issue has enough contract to assign implementation."), + "state:implementing": ("80b1d3", "Worker lane is actively implementing."), + "state:needs-review": ("ffffb3", "PR needs review."), + "state:needs-repair": ("fb8072", "PR or issue needs repair before it can advance."), + "state:repairing": ("fdb462", "Repair is actively assigned."), + "state:ready-for-approval": ("b3de69", "Pheidon/gate approval is the next action."), + "state:waiting-checks": ("ffffb3", "Approved or ready but waiting on checks/merge queue."), + "state:auto-merge-armed": ("b3de69", "Auto-merge is enabled and GitHub gates own completion."), + "state:blocked-human": ("e41a1c", "Human decision required."), + "state:blocked-infra": ("984ea3", "Blocked by tool, auth, runner, or infrastructure failure."), + "state:blocked-scope": ("ff7f00", "Blocked by unclear scope or acceptance criteria."), + "state:paused": ("999999", "Intentionally paused."), + "lane:pheidon": ("b3de69", "Orchestration, gate, governance, and controller action."), + "lane:apollo": ("8dd3c7", "Scope, backlog, synthesis, and issue-contract work."), + "lane:ares": ("fb8072", "Validation, adversarial review, and test-pressure work."), + "lane:daedalus": ("80b1d3", "Implementation and substantive code repair work."), + "lane:hephaestus": ("fdb462", "CI, build, lockfile, mergeability, and artifact work."), + "lane:hermes": ("bebada", "macOS/platform-native or special execution work."), +} + +LANE_BY_AUTHOR = { + "apollo-omt": "Apollo", + "ares-omt": "Ares", + "daedalus-omt": "Daedalus", + "hephaestus-omt": "Hephaestus", + "hermes-omt": "Hermes", + "pheidon-omt": "Pheidon", + "athena-omt": "Ares", + "athena": "Ares", + "chatgpt-codex-connector": "Daedalus", +} + +FAILURE_CONCLUSIONS = { + "ACTION_REQUIRED", + "CANCELLED", + "FAILURE", + "STARTUP_FAILURE", + "STALE", + "TIMED_OUT", +} + + +def label_names(item: dict[str, Any]) -> set[str]: + return {label.get("name", "") for label in item.get("labels") or [] if label.get("name")} + + +def first_label_value(labels: set[str], prefix: str) -> str | None: + for label in sorted(labels): + if label.startswith(prefix): + return label.split(":", 1)[1] + return None + + +def check_summary(rollup: list[dict[str, Any]] | None) -> dict[str, Any]: + """Summarize checks and fail closed unless every observed check succeeded.""" + failing: list[str] = [] + pending: list[str] = [] + passing: list[str] = [] + indeterminate: list[str] = [] + checks = rollup or [] + for check in checks: + name = check.get("name") or check.get("workflowName") or check.get("context") or "check" + status = str(check.get("status") or "UNKNOWN").upper() + conclusion = str(check.get("conclusion") or "UNKNOWN").upper() + if status != "COMPLETED": + pending.append(name) + elif conclusion == "SUCCESS": + passing.append(name) + elif conclusion in FAILURE_CONCLUSIONS: + failing.append(name) + else: + indeterminate.append(name) + return { + "green": bool(checks) and bool(passing) and not failing and not pending and not indeterminate, + "failing": sorted(failing), + "pending": sorted(pending), + "passing": sorted(passing), + "indeterminate": sorted(indeterminate), + "observed": len(checks), + } + + +def active_change_requests(pr: dict[str, Any]) -> list[dict[str, str]]: + requests = [] + for review in pr.get("latestReviews") or []: + if review.get("state") == "CHANGES_REQUESTED": + requests.append( + { + "author": (review.get("author") or {}).get("login") or "unknown", + "submittedAt": review.get("submittedAt") or "", + "body": (review.get("body") or "").strip()[:1000], + } + ) + return requests + + +def owner_from_author(pr: dict[str, Any]) -> str: + login = ((pr.get("author") or {}).get("login") or "").lower() + return LANE_BY_AUTHOR.get(login, "Pheidon") + + +def repair_actor_for_pr(owner_lane: str, checks: dict[str, Any], merge_state: str) -> tuple[str, str]: + failing = {str(name).lower() for name in checks.get("failing") or []} + metadata_markers = ( + "validate pr description", + "ci gate", + "validate secrets", + "detect relevant changes", + "attest", + ) + if merge_state in ("BEHIND", "DIRTY"): + return "Hephaestus", f"Restore mergeability from {merge_state.lower()} state, rerun required checks, and push with --force-with-lease when rebasing." + if any(any(marker in name for marker in metadata_markers) for name in failing): + return "Hephaestus", "Repair PR metadata, generated governance surfaces, or CI gate wiring, then rerun required checks." + if owner_lane in ("Pheidon", "Apollo"): + return "Daedalus", "Reproduce failing checks, make the minimal substantive repair, validate, and push." + return owner_lane, "Reproduce failing checks, make the minimal repair, validate, and push." + + +def classify_pr(pr: dict[str, Any]) -> dict[str, Any]: + checks = check_summary(pr.get("statusCheckRollup")) + labels = label_names(pr) + merge_state = pr.get("mergeStateStatus") or "UNKNOWN" + review = pr.get("reviewDecision") or "UNKNOWN" + changes = active_change_requests(pr) + owner = first_label_value(labels, "lane:") + owner_lane = owner.title() if owner else owner_from_author(pr) + next_actor = owner_lane + reasons: list[str] = [] + + if pr.get("isDraft"): + state = "Paused" + next_action = "Undraft or explicitly mark as paused/not planned." + reasons.append("draft") + elif changes or review == "CHANGES_REQUESTED": + state = "Needs Repair" + next_actor = owner_lane if owner_lane not in ("Pheidon", "Apollo") else "Daedalus" + next_action = "Convert requested changes into a repair task and push an updated PR head." + reasons.append("changes_requested") + elif merge_state in ("DIRTY", "BEHIND"): + state = "Needs Repair" + next_actor, next_action = repair_actor_for_pr(owner_lane, checks, merge_state) + reasons.append(f"merge_state:{merge_state.lower()}") + if checks["failing"]: + reasons.append("failing_checks") + elif checks["failing"]: + state = "Needs Repair" + next_actor, next_action = repair_actor_for_pr(owner_lane, checks, merge_state) + reasons.append("failing_checks") + elif review == "REVIEW_REQUIRED": + state = "Needs Review" + next_actor = "Ares" + next_action = "Perform non-author review and either approve or file exact repair feedback." + reasons.append("review_required") + elif checks["pending"] or checks["indeterminate"] or not checks["observed"]: + state = "Approved / Waiting Checks" if review == "APPROVED" else "PR Open" + next_action = "Wait for complete required-check evidence; reclassify only after every observed check succeeds." + reasons.append("checks_pending" if checks["pending"] else "checks_indeterminate") + elif merge_state == "BLOCKED": + state = "Blocked - Infrastructure" + next_actor = "Pheidon" + next_action = "Inspect branch protection, reviews, and checks and convert the blocker into an explicit next action." + reasons.append("blocked") + elif merge_state == "CLEAN" and review == "APPROVED" and checks["green"]: + if pr.get("autoMergeRequest"): + state = "Auto-merge Armed" + next_action = "Wait for GitHub auto-merge or merge queue completion." + reasons.append("clean_approved_green") + else: + state = "Needs Repair" + next_action = "PR author should enable auto-merge where GitHub allows it, or record the unavailable or unsafe reason under Merge Automation." + reasons.extend(["clean_approved_green", "auto_merge_missing"]) + else: + state = "PR Open" + next_actor = "Pheidon" + next_action = "Inspect manually; classifier did not find a confident next state." + reasons.append("unclassified") + + return { + "kind": "pr", + "number": pr["number"], + "title": pr["title"], + "url": pr["url"], + "flowState": state, + "ownerLane": owner_lane, + "nextActor": next_actor, + "nextAction": next_action, + "reasons": reasons, + "mergeStateStatus": merge_state, + "reviewDecision": review, + "checks": checks, + "autoMergeArmed": bool(pr.get("autoMergeRequest")), + "changeRequests": changes, + "labels": sorted(labels), + "updatedAt": pr.get("updatedAt"), + } + + +def classify_issue(issue: dict[str, Any]) -> dict[str, Any]: + labels = label_names(issue) + state_label = first_label_value(labels, "state:") + lane_label = first_label_value(labels, "lane:") + autonomy = first_label_value(labels, "autonomy:") + body = issue.get("body") or "" + state_map = { + "intake": "Intake", + "ready-for-planning": "Ready for Planning", + "ready-for-implementation": "Ready for Implementation", + "implementing": "Implementing", + "needs-review": "Needs Review", + "needs-repair": "Needs Repair", + "repairing": "Repairing", + "blocked-human": "Blocked - Human", + "blocked-infra": "Blocked - Infrastructure", + "blocked-scope": "Blocked - Scope", + "paused": "Paused", + } + flow_state = state_map.get(state_label or "", "") + if not flow_state: + has_contract = "acceptance criteria" in body.lower() or "validation" in body.lower() + flow_state = "Ready for Implementation" if has_contract else "Intake" + + owner_lane = lane_label.title() if lane_label else "Pheidon" + if flow_state == "Intake": + next_actor = "Apollo" + next_action = "Refine into an executable issue contract with acceptance criteria and validation." + elif flow_state == "Ready for Implementation": + next_actor = owner_lane if owner_lane != "Pheidon" else "Daedalus" + next_action = "Assign to worker WIP slot when PR queue allows." + elif flow_state.startswith("Blocked"): + next_actor = "Pheidon" + next_action = "Resolve blocker or convert it into a flow-blocker issue." + else: + next_actor = owner_lane + next_action = "Continue according to current state." + + return { + "kind": "issue", + "number": issue["number"], + "title": issue["title"], + "url": issue["url"], + "flowState": flow_state, + "ownerLane": owner_lane, + "nextActor": next_actor, + "nextAction": next_action, + "autonomy": autonomy, + "labels": sorted(labels), + "updatedAt": issue.get("updatedAt"), + } + + +def desired_labels(item: dict[str, Any]) -> list[str]: + labels = [] + state = STATE_LABEL_BY_FLOW.get(item.get("flowState", "")) + lane = LANE_LABEL_BY_ACTOR.get(item.get("nextActor", "")) or LANE_LABEL_BY_ACTOR.get(item.get("ownerLane", "")) + if state: + labels.append(state) + if lane: + labels.append(lane) + return labels + + +def plan_label_changes(items: list[dict[str, Any]], *, max_items: int | None = None) -> list[dict[str, Any]]: + selected = items if max_items is None else items[:max_items] + changes = [] + for item in selected: + current = set(item.get("labels") or []) + desired = set(desired_labels(item)) + stale = {label for label in current if label.startswith(("state:", "lane:")) and label not in desired} + add = sorted(desired - current) + remove = sorted(stale) + if add or remove: + changes.append({"kind": item["kind"], "number": item["number"], "add": add, "remove": remove}) + return changes + + +def default_assignments() -> dict[str, Any]: + return {lane: {"current": None, "queued": []} for lane in ("apollo", "ares", "daedalus", "hephaestus", "hermes")} + + +def plan_repair_dispatch( + repo: str, + prs: list[dict[str, Any]], + assignments: dict[str, Any], + *, + max_items: int, +) -> tuple[dict[str, Any], list[dict[str, Any]]]: + planned = deepcopy(assignments) + existing = set() + for bucket in planned.values(): + current = bucket.get("current") if isinstance(bucket, dict) else None + if current: + existing.add((current.get("repo"), current.get("number"))) + for queued in (bucket.get("queued") or []) if isinstance(bucket, dict) else []: + existing.add((queued.get("repo"), queued.get("number"))) + + added = [] + for item in prs: + if len(added) >= max_items: + break + if item.get("flowState") != "Needs Repair": + continue + lane = str(item.get("nextActor") or "").lower() + if lane not in ("apollo", "ares", "daedalus", "hephaestus", "hermes"): + lane = "daedalus" + key = (repo, item["number"]) + if key in existing: + continue + payload = { + "repo": repo, + "number": item["number"], + "title": item["title"], + "url": item["url"], + "flowState": item["flowState"], + "nextActor": item["nextActor"], + "nextAction": item["nextAction"], + "reasons": item.get("reasons", []), + "mergeStateStatus": item.get("mergeStateStatus"), + "reviewDecision": item.get("reviewDecision"), + "failingChecks": item.get("checks", {}).get("failing", []), + "pendingChecks": item.get("checks", {}).get("pending", []), + "coordinationRoutine": "flow_inspector_dispatch", + } + bucket = planned.setdefault(lane, {"current": None, "queued": []}) + if bucket.get("current") is None: + bucket["current"] = payload + else: + bucket.setdefault("queued", []).append(payload) + existing.add(key) + added.append({"lane": lane, "number": item["number"], "title": item["title"]}) + return planned, added diff --git a/tests/test_flow_inspector.py b/tests/test_flow_inspector.py new file mode 100644 index 0000000..0bffc2a --- /dev/null +++ b/tests/test_flow_inspector.py @@ -0,0 +1,311 @@ +from __future__ import annotations + +import importlib.util +import io +import json +import subprocess +import sys +import tempfile +import unittest +from contextlib import redirect_stderr, redirect_stdout +from pathlib import Path +from unittest import mock + + +ROOT = Path(__file__).resolve().parents[1] +FLOW_SCRIPTS = ROOT / "scripts" / "flow" + + +def load_module(name: str, path: Path): + sys.path.insert(0, str(FLOW_SCRIPTS)) + try: + spec = importlib.util.spec_from_file_location(name, path) + if spec is None or spec.loader is None: + raise RuntimeError(f"Unable to load {path}") + module = importlib.util.module_from_spec(spec) + spec.loader.exec_module(module) + return module + finally: + sys.path.remove(str(FLOW_SCRIPTS)) + + +CORE = load_module("inspector_core", FLOW_SCRIPTS / "inspector_core.py") +CLI = load_module("inspect_repo_flow", FLOW_SCRIPTS / "inspect_repo_flow.py") + + +def check(*, status: str = "COMPLETED", conclusion: str | None = "SUCCESS", name: str = "CI Gate"): + return {"name": name, "status": status, "conclusion": conclusion} + + +def pull_request(**overrides): + value = { + "number": 24, + "title": "Harden inspector", + "url": "https://example.invalid/pull/24", + "isDraft": False, + "author": {"login": "daedalus-omt"}, + "mergeStateStatus": "CLEAN", + "reviewDecision": "APPROVED", + "statusCheckRollup": [check()], + "autoMergeRequest": {"enabledAt": "2026-07-14T00:00:00Z"}, + "latestReviews": [], + "labels": [{"name": "lane:daedalus"}], + "updatedAt": "2026-07-14T00:00:00Z", + } + value.update(overrides) + return value + + +class CheckSummaryTests(unittest.TestCase): + def test_only_completed_success_is_green(self) -> None: + cases = [ + ("empty", [], False, "indeterminate"), + ("pending", [check(status="IN_PROGRESS", conclusion=None)], False, "pending"), + ("passing", [check()], True, "passing"), + ("failing", [check(conclusion="FAILURE")], False, "failing"), + ("neutral", [check(conclusion="NEUTRAL")], False, "indeterminate"), + ("skipped", [check(conclusion="SKIPPED")], False, "indeterminate"), + ("cancelled", [check(conclusion="CANCELLED")], False, "failing"), + ("unknown", [check(conclusion="MYSTERY")], False, "indeterminate"), + ] + for name, rollup, expected_green, bucket in cases: + with self.subTest(name=name): + summary = CORE.check_summary(rollup) + self.assertEqual(summary["green"], expected_green) + if rollup: + self.assertEqual(summary[bucket], ["CI Gate"]) + else: + self.assertEqual(summary["observed"], 0) + + def test_one_indeterminate_check_prevents_other_successes_from_being_green(self) -> None: + summary = CORE.check_summary([check(), check(name="Optional", conclusion="SKIPPED")]) + self.assertFalse(summary["green"]) + self.assertEqual(summary["passing"], ["CI Gate"]) + self.assertEqual(summary["indeterminate"], ["Optional"]) + + +class PullRequestClassificationTests(unittest.TestCase): + def test_pr_state_matrix(self) -> None: + cases = [ + ("draft", {"isDraft": True}, "Paused", "draft"), + ( + "changes requested", + {"reviewDecision": "CHANGES_REQUESTED"}, + "Needs Repair", + "changes_requested", + ), + ("dirty", {"mergeStateStatus": "DIRTY"}, "Needs Repair", "merge_state:dirty"), + ("behind", {"mergeStateStatus": "BEHIND"}, "Needs Repair", "merge_state:behind"), + ( + "failing", + {"statusCheckRollup": [check(conclusion="FAILURE")]}, + "Needs Repair", + "failing_checks", + ), + ( + "review required", + {"reviewDecision": "REVIEW_REQUIRED"}, + "Needs Review", + "review_required", + ), + ( + "pending", + {"statusCheckRollup": [check(status="QUEUED", conclusion=None)]}, + "Approved / Waiting Checks", + "checks_pending", + ), + ( + "empty", + {"statusCheckRollup": []}, + "Approved / Waiting Checks", + "checks_indeterminate", + ), + ( + "blocked", + {"mergeStateStatus": "BLOCKED"}, + "Blocked - Infrastructure", + "blocked", + ), + ( + "approved without auto merge", + {"autoMergeRequest": None}, + "Needs Repair", + "auto_merge_missing", + ), + ("auto merge", {}, "Auto-merge Armed", "clean_approved_green"), + ] + for name, overrides, expected_state, expected_reason in cases: + with self.subTest(name=name): + result = CORE.classify_pr(pull_request(**overrides)) + self.assertEqual(result["flowState"], expected_state) + self.assertIn(expected_reason, result["reasons"]) + + def test_latest_change_request_is_active_repair_evidence(self) -> None: + result = CORE.classify_pr( + pull_request( + latestReviews=[ + { + "state": "CHANGES_REQUESTED", + "author": {"login": "reviewer"}, + "submittedAt": "2026-07-14T00:00:00Z", + "body": "Please fix this", + } + ] + ) + ) + self.assertEqual(result["flowState"], "Needs Repair") + self.assertEqual(result["changeRequests"][0]["author"], "reviewer") + + +class IssueClassificationTests(unittest.TestCase): + def test_explicit_intake_routes_to_planning(self) -> None: + result = CORE.classify_issue( + { + "number": 24, + "title": "Inspector", + "url": "https://example.invalid/issues/24", + "body": "Complete acceptance criteria", + "labels": [{"name": "state:intake"}, {"name": "lane:daedalus"}], + } + ) + self.assertEqual(result["flowState"], "Intake") + self.assertEqual(result["nextActor"], "Apollo") + + def test_unlabelled_complete_contract_is_ready_for_implementation(self) -> None: + result = CORE.classify_issue( + { + "number": 25, + "title": "Drift", + "url": "https://example.invalid/issues/25", + "body": "## Acceptance criteria\n- complete", + "labels": [], + } + ) + self.assertEqual(result["flowState"], "Ready for Implementation") + self.assertEqual(result["nextActor"], "Daedalus") + + def test_blocked_issue_routes_to_pheidon(self) -> None: + result = CORE.classify_issue( + { + "number": 26, + "title": "Decision", + "url": "https://example.invalid/issues/26", + "body": "", + "labels": [{"name": "state:blocked-human"}, {"name": "lane:apollo"}], + } + ) + self.assertEqual(result["flowState"], "Blocked - Human") + self.assertEqual(result["nextActor"], "Pheidon") + + +class MutationPlanningTests(unittest.TestCase): + def test_label_plan_reconciles_mutually_exclusive_families(self) -> None: + item = CORE.classify_pr( + pull_request( + reviewDecision="REVIEW_REQUIRED", + labels=[ + {"name": "state:implementing"}, + {"name": "state:paused"}, + {"name": "lane:daedalus"}, + {"name": "kind:bug"}, + ], + ) + ) + changes = CORE.plan_label_changes([item]) + self.assertEqual( + changes, + [ + { + "kind": "pr", + "number": 24, + "add": ["lane:ares", "state:needs-review"], + "remove": ["lane:daedalus", "state:implementing", "state:paused"], + } + ], + ) + + def test_apply_labels_executes_exact_add_and_remove_mutation(self) -> None: + item = CORE.classify_pr( + pull_request( + reviewDecision="REVIEW_REQUIRED", + labels=[{"name": "state:implementing"}, {"name": "lane:daedalus"}], + ) + ) + completed = subprocess.CompletedProcess([], 0, "", "") + with mock.patch.object(CLI, "ensure_label") as ensure, mock.patch.object(CLI, "run", return_value=completed) as run: + result = CLI.apply_labels("OMT-Global/flow", [item]) + self.assertEqual(result["count"], 1) + self.assertEqual( + sorted(call.args[1] for call in ensure.call_args_list), + ["lane:ares", "state:needs-review"], + ) + run.assert_called_once_with( + [ + "gh", + "pr", + "edit", + "24", + "--repo", + "OMT-Global/flow", + "--add-label", + "lane:ares,state:needs-review", + "--remove-label", + "lane:daedalus,state:implementing", + ] + ) + + def test_repair_dispatch_deduplicates_existing_assignment(self) -> None: + item = CORE.classify_pr(pull_request(autoMergeRequest=None)) + initial = CORE.default_assignments() + updated, first = CORE.plan_repair_dispatch("OMT-Global/flow", [item], initial, max_items=10) + _, second = CORE.plan_repair_dispatch("OMT-Global/flow", [item], updated, max_items=10) + self.assertEqual(first, [{"lane": "daedalus", "number": 24, "title": "Harden inspector"}]) + self.assertEqual(second, []) + + def test_assignment_write_uses_exclusive_lock_and_atomic_replace(self) -> None: + item = CORE.classify_pr(pull_request(autoMergeRequest=None)) + with tempfile.TemporaryDirectory() as directory: + path = Path(directory) / "assignments.json" + with mock.patch.object(CLI.fcntl, "flock", wraps=CLI.fcntl.flock) as flock: + result = CLI.dispatch_repairs("OMT-Global/flow", [item], path, max_items=10) + self.assertEqual(result["count"], 1) + self.assertEqual(json.loads(path.read_text())["daedalus"]["current"]["number"], 24) + self.assertTrue(any(call.args[1] == CLI.fcntl.LOCK_EX for call in flock.call_args_list)) + self.assertTrue(any(call.args[1] == CLI.fcntl.LOCK_UN for call in flock.call_args_list)) + self.assertEqual(list(path.parent.glob(".assignments.json.*.tmp")), []) + + +class CliSafetyTests(unittest.TestCase): + def test_repo_is_required(self) -> None: + with redirect_stderr(io.StringIO()), self.assertRaises(SystemExit): + CLI.parse_args([]) + + def test_default_mode_has_no_mutation_flags_or_implicit_paths(self) -> None: + args = CLI.parse_args(["--repo", "OMT-Global/flow"]) + self.assertFalse(args.apply_labels) + self.assertFalse(args.dispatch_repairs) + self.assertIsNone(args.assignments) + self.assertIsNone(args.json_out) + self.assertIsNone(args.maintenance_gate) + + def test_dispatch_requires_explicit_assignment_path(self) -> None: + with redirect_stderr(io.StringIO()), self.assertRaises(SystemExit): + CLI.parse_args(["--repo", "OMT-Global/flow", "--dispatch-repairs"]) + + def test_default_main_reports_without_calling_mutators(self) -> None: + output = io.StringIO() + with ( + mock.patch.object(CLI, "gh_json", side_effect=[[], []]), + mock.patch.object(CLI, "apply_labels") as apply_labels, + mock.patch.object(CLI, "dispatch_repairs") as dispatch_repairs, + redirect_stdout(output), + ): + CLI.main(["--repo", "OMT-Global/flow", "--issues"]) + apply_labels.assert_not_called() + dispatch_repairs.assert_not_called() + self.assertIn("mode: read-only", output.getvalue()) + self.assertIn("Proposed writes", output.getvalue()) + + +if __name__ == "__main__": + unittest.main()