diff --git a/src/code_mower/controller.py b/src/code_mower/controller.py index b60dd560..4061429e 100644 --- a/src/code_mower/controller.py +++ b/src/code_mower/controller.py @@ -385,6 +385,10 @@ def _pr_decision( except ContractError: valid_lineage = False base["lineage"] = {key: lineage.get(key) for key in ("status", "reason", "contributors", "current_writer")} + if lineage.get("status") == "unmanaged": + return {**base, "lane_id": "", "decision_state": "unmanaged", + "next_action": "not a Code Mower PR", "next_detail": "lineage unmanaged: " + str(lineage.get("reason", "no_code_mower_provenance")), + "stop_condition": "unmanaged", "owner_action_kind": "", "merge_method": ""} if not valid_lineage: return {**base, "lane_id": "", "decision_state": "owner_action", "next_action": "owner action required", "next_detail": "lineage " + str(lineage.get("status", "unknown")), @@ -554,7 +558,14 @@ def _pr_decision( def _select_pr(prs: Sequence[Mapping[str, Any]]) -> Mapping[str, Any] | None: - return sorted(prs, key=_pr_priority)[0] if prs else None + managed = [ + pr for pr in prs + if not ( + isinstance(pr.get("lineage"), Mapping) + and pr["lineage"].get("status") == "unmanaged" + ) + ] + return sorted(managed, key=_pr_priority)[0] if managed else None def _queue_metrics( diff --git a/src/code_mower/lane_status.py b/src/code_mower/lane_status.py index 04c308ab..403db5da 100644 --- a/src/code_mower/lane_status.py +++ b/src/code_mower/lane_status.py @@ -235,6 +235,138 @@ def _checks(raw: Any) -> list[dict[str, str]]: return (major or checks)[:8] +def _has_code_mower_check_claim(raw: Any) -> bool: + """Inspect every readable raw check identity for Code Mower provenance. + + The operator-facing check projection is intentionally bounded. Provenance + classification must not inherit that display limit, because a real Code + Mower check can appear after any number of unrelated checks. + """ + if not isinstance(raw, list): + return False + for check in raw: + if not isinstance(check, Mapping): + continue + for field in ("name", "context", "workflowName"): + value = check.get(field) + if not isinstance(value, str): + continue + if _is_code_mower_check_identity(value): + return True + app = check.get("app") + if isinstance(app, Mapping): + for field in ("slug", "name"): + value = app.get(field) + if not isinstance(value, str): + continue + if _normalized_app_identity(value) == "code-mower": + return True + return False + + +def _is_code_mower_check_identity(value: str) -> bool: + """Match the normalized Code Mower check namespace at an exact boundary.""" + + normalized = re.sub(r"\s+", " ", value.strip().casefold()).replace("_", "-") + if normalized.startswith("code mower"): + normalized = "code-mower" + normalized.removeprefix("code mower") + return ( + normalized == "code-mower" + or normalized.startswith("code-mower/") + or normalized.startswith("code-mower ") + ) + + +def _normalized_app_identity(value: str) -> str: + return re.sub(r"[^a-z0-9]+", "-", value.strip().casefold()).strip("-") + + +def _status_check_rollup_is_readable(raw: Any) -> bool: + """Validate the GitHub union records used as check identities. + + `gh pr --json statusCheckRollup` returns CheckRun and StatusContext union + variants. Tests and older gh versions may omit ``__typename``, so the + primary ``name``/``context`` field also identifies the variant. Optional + fields may be absent or null, but a present value must retain its source + type; stringifying malformed identity data would make an incomplete list + look like trustworthy evidence that no Code Mower check exists. + """ + if not isinstance(raw, list): + return False + for check in raw: + if not isinstance(check, Mapping): + return False + + typename = check.get("__typename") + if typename is not None and typename not in {"CheckRun", "StatusContext"}: + return False + + for field in ("name", "context", "workflowName"): + if field in check and check[field] is not None and not isinstance(check[field], str): + return False + + name = check.get("name") + context = check.get("context") + has_name = isinstance(name, str) and bool(name.strip()) + has_context = isinstance(context, str) and bool(context.strip()) + if typename == "CheckRun" and (not has_name or has_context): + return False + if typename == "StatusContext" and (not has_context or has_name): + return False + if typename is None and has_name == has_context: + return False + variant = typename or ("CheckRun" if has_name else "StatusContext") + + # Reject known fields from the other side of the GraphQL union. Empty + # nullable compatibility values carry no identity, but two populated + # variant shapes must never be guessed into one. + cross_variant_fields = ( + ("context", "targetUrl", "state") + if variant == "CheckRun" + else ("name", "workflowName", "detailsUrl", "conclusion", "status", "completedAt") + ) + if any(check.get(field) not in (None, "") for field in cross_variant_fields): + return False + + for field in ( + "detailsUrl", "targetUrl", "startedAt", "createdAt", "completedAt", + "conclusion", "state", "status", + ): + if field in check and check[field] is not None and not isinstance(check[field], str): + return False + + app = check.get("app") + if app is not None: + if variant != "CheckRun": + return False + if not isinstance(app, Mapping): + return False + recognized_identity = False + normalized_text_identities = [] + for field in ("slug", "name"): + if field in app: + if not isinstance(app[field], str) or not app[field].strip(): + return False + normalized_identity = _normalized_app_identity(app[field]) + if not normalized_identity: + return False + normalized_text_identities.append(normalized_identity) + recognized_identity = True + if len(set(normalized_text_identities)) > 1: + return False + if "databaseId" in app: + database_id = app["databaseId"] + if not ( + (isinstance(database_id, str) and bool(database_id.strip())) + or (type(database_id) is int and database_id > 0) + ): + return False + recognized_identity = True + if not recognized_identity: + return False + return True + + def _has_state(checks: Sequence[Mapping[str, str]], states: set[str]) -> bool: return any(check.get("state", "") in states for check in checks) @@ -324,6 +456,93 @@ def _author(pr: Mapping[str, Any]) -> str: return _text(author.get("login")) if isinstance(author, Mapping) else _text(author) +def _extract_code_mower_labels(lineage_config: Mapping[str, Any] | None) -> set[str]: + """Extract all configured Code Mower labels from policy. + + Returns builder labels, dispatch labels, and audit labels derived from + validated policy. Empty when lineage_config is None or lanes are missing. + """ + if lineage_config is None: + return set() + + labels = set() + + builder_identity = lineage_config.get("builder_identity", {}) + if isinstance(builder_identity, Mapping): + builder_labels = builder_identity.get("labels", {}) + if isinstance(builder_labels, Mapping): + labels.update(_text(label).lower() for label in builder_labels.keys()) + + for label, lane in builder_labels.items(): + label_lower = _text(label).lower() + if label_lower.startswith("builder:"): + label_suffix = label_lower.removeprefix("builder:") + if label_suffix: + labels.add(f"dispatched:{label_suffix}") + lane_lower = _text(lane).lower() + labels.add(f"dispatched:{lane_lower}") + + lanes = lineage_config.get("lanes", {}) + if isinstance(lanes, Mapping): + for lane_data in lanes.values(): + if not isinstance(lane_data, Mapping): + continue + + lane_labels = lane_data.get("labels", {}) + if isinstance(lane_labels, Mapping): + for label_type in ("needs", "done", "blocked"): + label = lane_labels.get(label_type, "") + if label: + labels.add(_text(label).lower()) + + return labels + + +def _has_code_mower_claim( + *, + labels: Sequence[str], + author: str, + branch: str, + has_code_mower_check: bool, + identity: Any, + has_lineage_markers: bool, + lineage_config: Mapping[str, Any] | None = None, +) -> bool: + """Check if a PR has any Code Mower provenance claim. + + Returns True if the PR has configured builder/dispatch/audit labels, a mapped + author, a configured branch prefix, Code Mower checks, or readable lineage + markers. Returns False for ordinary PRs with no Code Mower involvement. + """ + if has_lineage_markers: + return True + + label_set = set(_text(label).lower() for label in labels) + + configured_labels = _extract_code_mower_labels(lineage_config) + if label_set & configured_labels: + return True + + if has_code_mower_check: + return True + + if identity: + identity_authors = getattr(identity, "authors", ()) + if identity_authors: + author_lower = _text(author).lower() + configured_authors = {_text(account).lower() for account, _ in identity_authors} + if author_lower in configured_authors: + return True + + identity_prefixes = getattr(identity, "branch_prefixes", ()) + if identity_prefixes: + branch_lower = _text(branch).lower() + if any(branch_lower.startswith(_text(prefix).lower()) for prefix, _ in identity_prefixes): + return True + + return False + + def _summarize_pr( repo: str, pr: Mapping[str, Any], @@ -432,18 +651,37 @@ def _remote( "next_action": "pass --config code-mower.yml to evaluate lineage", } continue + identity = None + has_lineage_markers = False + prerequisites_validated = False + history_validated = False + raw_labels = raw_pr.get("labels") + label_names = [item.get("name", "") for item in raw_labels if isinstance(item, Mapping)] if isinstance(raw_labels, list) else [] + raw_author = raw_pr.get("author") + author_login = raw_author.get("login", "") if isinstance(raw_author, Mapping) else "" + raw_branch = raw_pr.get("headRefName") + branch = raw_branch if isinstance(raw_branch, str) else "" + raw_checks = raw_pr.get("statusCheckRollup") + checks_readable = _status_check_rollup_is_readable(raw_checks) + has_code_mower_check = checks_readable and _has_code_mower_check_claim(raw_checks) + try: if policy_config.validate_config(lineage_config): raise ContractError("Trusted validated status policy required") identity = lineage_identity(lineage_config) authority = lineage_authorities(lineage_config) - target = Target(repo, raw_pr.get("number"), raw_pr.get("headRefName"), raw_pr.get("headRefOid")) - raw_labels = raw_pr.get("labels") - raw_author = raw_pr.get("author") if (not isinstance(raw_labels, list) - or any(not isinstance(item, Mapping) or not isinstance(item.get("name"), str) for item in raw_labels) - or not isinstance(raw_author, Mapping) or not isinstance(raw_author.get("login"), str)): - raise ContractError("Exact readable labels and author required") + or any(not isinstance(item, Mapping) + or not isinstance(item.get("name"), str) + or not item["name"].strip() for item in raw_labels) + or not isinstance(raw_author, Mapping) + or not isinstance(raw_author.get("login"), str) + or not raw_author["login"].strip() + or not isinstance(raw_branch, str) + or not checks_readable): + raise ContractError("Exact readable labels, author, branch, and checks required") + target = Target(repo, raw_pr.get("number"), raw_branch, raw_pr.get("headRefOid")) + prerequisites_validated = True def page(number, size, target=target): nonlocal budget if budget <= 0: @@ -451,8 +689,15 @@ def page(number, size, target=target): budget -= 1 return gh_json_runner(["api", f"repos/{target.repo}/issues/{target.pr_number}/comments?per_page={size}&page={number}"]) history = lineage_history(page) - _, decision = lineage_decision(target, identity, authority, history, - author=raw_author["login"], labels=[item["name"] for item in raw_labels]) + history_validated = True + + for comment in history.comments: + if comment.account in authority.accounts and "CODE_MOWER_BUILDER_LINEAGE" in comment.body: + has_lineage_markers = True + break + + chain, decision = lineage_decision(target, identity, authority, history, + author=author_login, labels=label_names) pr["lineage"] = lineage_projection(decision) pr["lineage"]["repo"] = target.repo pr["lineage"]["pr_number"] = target.pr_number @@ -462,21 +707,66 @@ def page(number, size, target=target): str(lane.get("author_lane") or lane.get("trailer_lane") or lane.get("provider") or key) for key, lane in lanes.items() if isinstance(lane, Mapping) and admit(decision, str(lane.get("author_lane") or lane.get("trailer_lane") or lane.get("provider") or key))}) + if pr["lineage"]["status"] == "ready" and pr["lineage"]["reason"] == "no_identity" and not pr["lineage"]["contributors"]: + has_claim = _has_code_mower_claim( + labels=label_names, + author=author_login, + branch=branch, + has_code_mower_check=has_code_mower_check, + identity=identity, + has_lineage_markers=has_lineage_markers, + lineage_config=lineage_config, + ) + if not has_claim: + pr["lineage"] = {"status": "unmanaged", "reason": "no_code_mower_provenance", + "current_writer": None, "contributors": [], "admitted_reviewers": []} except LaneStatusUnavailable: - pr["lineage"] = { - "status": "unavailable", - "reason": "lineage_unreadable", - "current_writer": None, - "contributors": [], - "admitted_reviewers": [], - "next_action": "restore readable lineage metadata and rerun status", - } + has_claim = _has_code_mower_claim( + labels=label_names, + author=author_login, + branch=branch, + has_code_mower_check=has_code_mower_check, + identity=identity, + has_lineage_markers=has_lineage_markers, + lineage_config=lineage_config, + ) + if has_claim: + pr["lineage"] = { + "status": "unavailable", + "reason": "lineage_unreadable", + "current_writer": None, + "contributors": [], + "admitted_reviewers": [], + "next_action": "restore readable lineage metadata and rerun status", + } + else: + pr["lineage"] = {"status": "unmanaged", "reason": "no_code_mower_provenance", + "current_writer": None, "contributors": [], "admitted_reviewers": []} except (ValueError, KeyError, TypeError, RuntimeError): - pr["lineage"] = {"status": "unknown", "reason": "lineage_unreadable", - "current_writer": None, "contributors": [], "admitted_reviewers": []} + if not prerequisites_validated or not history_validated: + pr["lineage"] = {"status": "unknown", "reason": "lineage_unreadable", + "current_writer": None, "contributors": [], "admitted_reviewers": []} + else: + has_claim = _has_code_mower_claim( + labels=label_names, + author=author_login, + branch=branch, + has_code_mower_check=has_code_mower_check, + identity=identity, + has_lineage_markers=has_lineage_markers, + lineage_config=lineage_config, + ) + if has_claim: + pr["lineage"] = {"status": "unknown", "reason": "lineage_unreadable", + "current_writer": None, "contributors": [], "admitted_reviewers": []} + else: + pr["lineage"] = {"status": "unmanaged", "reason": "no_code_mower_provenance", + "current_writer": None, "contributors": [], "admitted_reviewers": []} if pr["lineage"]["status"] == "unavailable": pr["next_action"] = str(pr["lineage"]["next_action"]) pr["next_detail"] = "lineage unavailable: " + pr["lineage"]["reason"] + elif pr["lineage"]["status"] == "unmanaged": + pass elif pr["lineage"]["status"] != "ready": pr["next_action"] = "owner action required" pr["next_detail"] = "lineage " + pr["lineage"]["status"] + ": " + pr["lineage"]["reason"] diff --git a/tests/test_controller.py b/tests/test_controller.py index d89ab3a7..c4ca5a11 100644 --- a/tests/test_controller.py +++ b/tests/test_controller.py @@ -207,6 +207,38 @@ def test_ready_issue_selects_one_builder_dispatch_without_mutation() -> None: assert report["decision"]["would_mutate"] is False +def test_unmanaged_pr_does_not_preempt_ready_issue_dispatch() -> None: + unmanaged = _pr(number=41) + unmanaged["lineage"] = { + "status": "unmanaged", + "reason": "no_code_mower_provenance", + "current_writer": None, + "contributors": [], + "admitted_reviewers": [], + } + + report = _evaluate( + [unmanaged], + ready_issues=[ + { + "number": 7, + "url": "https://github.com/owner/repo/issues/7", + "author": "owner", + "updated_at": NOW, + "labels": ["tier:R", "builder:codex"], + "builder_lane": "codex", + "assigned": False, + "dispatched": False, + "owner_action": False, + } + ], + ) + + assert report["decision"]["decision_state"] == "dispatch_builder" + assert report["decision"]["issue_number"] == 7 + assert report["queue"]["metrics"]["open_pr_count"] == 1 + + def test_blocked_audit_stops_controller() -> None: report = _evaluate([_pr(blocked=["claude-audit-blocked"])]) diff --git a/tests/test_lane_status.py b/tests/test_lane_status.py index 672c980e..6ceb6f24 100644 --- a/tests/test_lane_status.py +++ b/tests/test_lane_status.py @@ -864,3 +864,1278 @@ def command_runner(args: list[str]) -> subprocess.CompletedProcess[str]: inventory = lane_status.local_listener_inventory(command_runner) self.assertEqual((inventory["available"], inventory["listeners"]), (False, [])) + + def test_pr_with_no_code_mower_markers_is_unmanaged(self) -> None: + def gh_json(args: list[str]) -> object: + if args[:2] == ["pr", "list"]: + return [ + { + "number": 42, + "title": "Regular human PR", + "url": "https://github.com/owner/repo/pull/42", + "headRefName": "feature/my-work", + "headRefOid": "abcdef0123456789abcdef0123456789abcdef01", + "author": {"login": "human-contributor"}, + "isDraft": False, + "mergeStateStatus": "CLEAN", + "updatedAt": NOW.isoformat().replace("+00:00", "Z"), + "labels": [], + "statusCheckRollup": [], + } + ] + if args[:2] == ["run", "list"]: + return [] + if args[0] == "api" and "/comments?" in args[1]: + return [] + raise lane_status.LaneStatusUnavailable("unexpected gh call") + + report = lane_status.collect_status( + lineage_config=policy({}), + repo="owner/repo", + gh_json_runner=gh_json, + command_runner=lambda _args: _completed(""), + now=NOW, + ) + + pr = report["remote"]["pull_requests"][0] + self.assertEqual(pr["lineage"]["status"], "unmanaged") + self.assertEqual(pr["lineage"]["reason"], "no_code_mower_provenance") + self.assertNotEqual(pr["next_action"], "owner action required") + + def test_dependabot_pr_is_unmanaged(self) -> None: + def gh_json(args: list[str]) -> object: + if args[:2] == ["pr", "list"]: + return [ + { + "number": 99, + "title": "Bump dependency version", + "url": "https://github.com/owner/repo/pull/99", + "headRefName": "dependabot/npm_and_yarn/deps-1234", + "headRefOid": "abcdef0123456789abcdef0123456789abcdef01", + "author": {"login": "dependabot[bot]"}, + "isDraft": False, + "mergeStateStatus": "CLEAN", + "updatedAt": NOW.isoformat().replace("+00:00", "Z"), + "labels": [{"name": "dependencies"}], + "statusCheckRollup": [], + } + ] + if args[:2] == ["run", "list"]: + return [] + if args[0] == "api" and "/comments?" in args[1]: + return [] + raise lane_status.LaneStatusUnavailable("unexpected gh call") + + report = lane_status.collect_status( + lineage_config=policy({}), + repo="owner/repo", + gh_json_runner=gh_json, + command_runner=lambda _args: _completed(""), + now=NOW, + ) + + pr = report["remote"]["pull_requests"][0] + self.assertEqual(pr["lineage"]["status"], "unmanaged") + self.assertEqual(pr["lineage"]["reason"], "no_code_mower_provenance") + self.assertNotEqual(pr["next_action"], "owner action required") + + def test_ordinary_pr_with_unavailable_history_is_unmanaged(self) -> None: + def gh_json(args: list[str]) -> object: + if args[:2] == ["pr", "list"]: + return [ + { + "number": 77, + "title": "Regular PR with network issues", + "url": "https://github.com/owner/repo/pull/77", + "headRefName": "feature/work", + "headRefOid": "abcdef0123456789abcdef0123456789abcdef01", + "author": {"login": "contributor"}, + "isDraft": False, + "mergeStateStatus": "CLEAN", + "updatedAt": NOW.isoformat().replace("+00:00", "Z"), + "labels": [], + "statusCheckRollup": [], + } + ] + if args[:2] == ["run", "list"]: + return [] + if args[0] == "api" and "/comments?" in args[1]: + raise lane_status.LaneStatusUnavailable("Network timeout") + raise lane_status.LaneStatusUnavailable("unexpected gh call") + + report = lane_status.collect_status( + lineage_config=policy({}), + repo="owner/repo", + gh_json_runner=gh_json, + command_runner=lambda _args: _completed(""), + now=NOW, + ) + + pr = report["remote"]["pull_requests"][0] + self.assertEqual(pr["lineage"]["status"], "unmanaged") + self.assertEqual(pr["lineage"]["reason"], "no_code_mower_provenance") + self.assertNotEqual(pr["next_action"], "owner action required") + + def test_claimed_pr_with_unavailable_history_is_actionable(self) -> None: + def gh_json(args: list[str]) -> object: + if args[:2] == ["pr", "list"]: + return [ + { + "number": 88, + "title": "Code Mower PR with network issues", + "url": "https://github.com/owner/repo/pull/88", + "headRefName": "codex/work", + "headRefOid": "abcdef0123456789abcdef0123456789abcdef01", + "author": {"login": "source-bot"}, + "isDraft": False, + "mergeStateStatus": "CLEAN", + "updatedAt": NOW.isoformat().replace("+00:00", "Z"), + "labels": [{"name": "builder:codex"}], + "statusCheckRollup": [], + } + ] + if args[:2] == ["run", "list"]: + return [] + if args[0] == "api" and "/comments?" in args[1]: + raise lane_status.LaneStatusUnavailable("Network timeout") + raise lane_status.LaneStatusUnavailable("unexpected gh call") + + report = lane_status.collect_status( + lineage_config=policy({}), + repo="owner/repo", + gh_json_runner=gh_json, + command_runner=lambda _args: _completed(""), + now=NOW, + ) + + pr = report["remote"]["pull_requests"][0] + self.assertEqual(pr["lineage"]["status"], "unavailable") + self.assertEqual(pr["lineage"]["reason"], "lineage_unreadable") + self.assertEqual(pr["next_action"], "restore readable lineage metadata and rerun status") + + def test_pr_with_generic_package_check_is_unmanaged(self) -> None: + def gh_json(args: list[str]) -> object: + if args[:2] == ["pr", "list"]: + return [ + { + "number": 111, + "title": "Normal PR with package check", + "url": "https://github.com/owner/repo/pull/111", + "headRefName": "fix/bug", + "headRefOid": "abcdef0123456789abcdef0123456789abcdef01", + "author": {"login": "developer"}, + "isDraft": False, + "mergeStateStatus": "CLEAN", + "updatedAt": NOW.isoformat().replace("+00:00", "Z"), + "labels": [], + "statusCheckRollup": [ + { + "__typename": "CheckRun", + "name": "package / build", + "conclusion": "SUCCESS", + } + ], + } + ] + if args[:2] == ["run", "list"]: + return [] + if args[0] == "api" and "/comments?" in args[1]: + return [] + raise lane_status.LaneStatusUnavailable("unexpected gh call") + + report = lane_status.collect_status( + lineage_config=policy({}), + repo="owner/repo", + gh_json_runner=gh_json, + command_runner=lambda _args: _completed(""), + now=NOW, + ) + + pr = report["remote"]["pull_requests"][0] + self.assertEqual(pr["lineage"]["status"], "unmanaged") + self.assertEqual(pr["lineage"]["reason"], "no_code_mower_provenance") + + def test_pr_with_configured_dispatch_alias_is_managed(self) -> None: + def gh_json(args: list[str]) -> object: + if args[:2] == ["pr", "list"]: + return [ + { + "number": 222, + "title": "Grok bot PR", + "url": "https://github.com/owner/repo/pull/222", + "headRefName": "grok/work", + "headRefOid": "abcdef0123456789abcdef0123456789abcdef01", + "author": {"login": "grok-bot[bot]"}, + "isDraft": False, + "mergeStateStatus": "CLEAN", + "updatedAt": NOW.isoformat().replace("+00:00", "Z"), + "labels": [{"name": "builder:grok-bot"}], + "statusCheckRollup": [], + } + ] + if args[:2] == ["run", "list"]: + return [] + if args[0] == "api" and "/comments?" in args[1]: + raise RuntimeError("Simulated error") + raise lane_status.LaneStatusUnavailable("unexpected gh call") + + config_with_grok = policy({}) + config_with_grok["builder_identity"]["labels"]["builder:grok-bot"] = "cursor" + + report = lane_status.collect_status( + lineage_config=config_with_grok, + repo="owner/repo", + gh_json_runner=gh_json, + command_runner=lambda _args: _completed(""), + now=NOW, + ) + + pr = report["remote"]["pull_requests"][0] + self.assertEqual(pr["lineage"]["status"], "unknown") + self.assertEqual(pr["lineage"]["reason"], "lineage_unreadable") + + def test_pr_with_builder_label_but_unreadable_lineage_is_actionable(self) -> None: + def gh_json(args: list[str]) -> object: + if args[:2] == ["pr", "list"]: + return [ + { + "number": 55, + "title": "Malformed Code Mower PR", + "url": "https://github.com/owner/repo/pull/55", + "headRefName": "codex/work", + "headRefOid": "abcdef0123456789abcdef0123456789abcdef01", + "author": {"login": "source-bot"}, + "isDraft": False, + "mergeStateStatus": "CLEAN", + "updatedAt": NOW.isoformat().replace("+00:00", "Z"), + "labels": [{"name": "builder:codex"}], + "statusCheckRollup": [], + } + ] + if args[:2] == ["run", "list"]: + return [] + if args[0] == "api" and "/comments?" in args[1]: + raise RuntimeError("Simulated history fetch failure") + raise lane_status.LaneStatusUnavailable("unexpected gh call") + + report = lane_status.collect_status( + lineage_config=policy({}), + repo="owner/repo", + gh_json_runner=gh_json, + command_runner=lambda _args: _completed(""), + now=NOW, + ) + + pr = report["remote"]["pull_requests"][0] + self.assertEqual(pr["lineage"]["status"], "unknown") + self.assertEqual(pr["lineage"]["reason"], "lineage_unreadable") + self.assertEqual(pr["next_action"], "owner action required") + + def test_pr_with_dispatch_label_is_managed(self) -> None: + def gh_json(args: list[str]) -> object: + if args[:2] == ["pr", "list"]: + return [ + { + "number": 333, + "title": "Dispatched work", + "url": "https://github.com/owner/repo/pull/333", + "headRefName": "feature/work", + "headRefOid": "abcdef0123456789abcdef0123456789abcdef01", + "author": {"login": "human-contributor"}, + "isDraft": False, + "mergeStateStatus": "CLEAN", + "updatedAt": NOW.isoformat().replace("+00:00", "Z"), + "labels": [{"name": "dispatched:codex"}], + "statusCheckRollup": [], + } + ] + if args[:2] == ["run", "list"]: + return [] + if args[0] == "api" and "/comments?" in args[1]: + raise RuntimeError("Simulated error") + raise lane_status.LaneStatusUnavailable("unexpected gh call") + + config = policy() + from code_mower import config as policy_config + self.assertEqual(policy_config.validate_config(config), []) + + report = lane_status.collect_status( + lineage_config=config, + repo="owner/repo", + gh_json_runner=gh_json, + command_runner=lambda _args: _completed(""), + now=NOW, + ) + + pr = report["remote"]["pull_requests"][0] + self.assertEqual(pr["lineage"]["status"], "unknown") + self.assertEqual(pr["lineage"]["reason"], "lineage_unreadable") + + def test_pr_with_dispatch_alias_is_managed(self) -> None: + def gh_json(args: list[str]) -> object: + if args[:2] == ["pr", "list"]: + return [ + { + "number": 444, + "title": "Legacy dispatch alias", + "url": "https://github.com/owner/repo/pull/444", + "headRefName": "feature/legacy", + "headRefOid": "abcdef0123456789abcdef0123456789abcdef01", + "author": {"login": "human"}, + "isDraft": False, + "mergeStateStatus": "CLEAN", + "updatedAt": NOW.isoformat().replace("+00:00", "Z"), + "labels": [{"name": "dispatched:grok-bot"}], + "statusCheckRollup": [], + } + ] + if args[:2] == ["run", "list"]: + return [] + if args[0] == "api" and "/comments?" in args[1]: + raise RuntimeError("Simulated error") + raise lane_status.LaneStatusUnavailable("unexpected gh call") + + config = policy() + config["builder_identity"]["labels"]["builder:grok-bot"] = "cursor" + from code_mower import config as policy_config + self.assertEqual(policy_config.validate_config(config), []) + + report = lane_status.collect_status( + lineage_config=config, + repo="owner/repo", + gh_json_runner=gh_json, + command_runner=lambda _args: _completed(""), + now=NOW, + ) + + pr = report["remote"]["pull_requests"][0] + self.assertEqual(pr["lineage"]["status"], "unknown") + self.assertEqual(pr["lineage"]["reason"], "lineage_unreadable") + + def test_pr_with_audit_need_label_is_managed(self) -> None: + def gh_json(args: list[str]) -> object: + if args[:2] == ["pr", "list"]: + return [ + { + "number": 555, + "title": "Needs audit", + "url": "https://github.com/owner/repo/pull/555", + "headRefName": "feature/needs-audit", + "headRefOid": "abcdef0123456789abcdef0123456789abcdef01", + "author": {"login": "contributor"}, + "isDraft": False, + "mergeStateStatus": "CLEAN", + "updatedAt": NOW.isoformat().replace("+00:00", "Z"), + "labels": [{"name": "needs-codex-audit"}], + "statusCheckRollup": [], + } + ] + if args[:2] == ["run", "list"]: + return [] + if args[0] == "api" and "/comments?" in args[1]: + raise RuntimeError("Simulated error") + raise lane_status.LaneStatusUnavailable("unexpected gh call") + + config_with_audit = policy() + config_with_audit["lanes"] = { + "codex": { + "type": "audit", + "driver": "local_cli", + "provider": "codex", + "labels": { + "needs": "needs-codex-audit", + "done": "codex-audit-done", + "blocked": "codex-audit-blocked" + } + } + } + from code_mower import config as policy_config + self.assertEqual(policy_config.validate_config(config_with_audit), []) + + report = lane_status.collect_status( + lineage_config=config_with_audit, + repo="owner/repo", + gh_json_runner=gh_json, + command_runner=lambda _args: _completed(""), + now=NOW, + ) + + pr = report["remote"]["pull_requests"][0] + self.assertEqual(pr["lineage"]["status"], "unknown") + self.assertEqual(pr["lineage"]["reason"], "lineage_unreadable") + + def test_pr_with_generic_audit_label_is_unmanaged(self) -> None: + def gh_json(args: list[str]) -> object: + if args[:2] == ["pr", "list"]: + return [ + { + "number": 666, + "title": "Security audit needed", + "url": "https://github.com/owner/repo/pull/666", + "headRefName": "feature/security", + "headRefOid": "abcdef0123456789abcdef0123456789abcdef01", + "author": {"login": "developer"}, + "isDraft": False, + "mergeStateStatus": "CLEAN", + "updatedAt": NOW.isoformat().replace("+00:00", "Z"), + "labels": [{"name": "needs-security-audit"}], + "statusCheckRollup": [], + } + ] + if args[:2] == ["run", "list"]: + return [] + if args[0] == "api" and "/comments?" in args[1]: + return [] + raise lane_status.LaneStatusUnavailable("unexpected gh call") + + config_with_codex_audit = policy() + config_with_codex_audit["lanes"] = { + "codex": { + "type": "audit", + "driver": "local_cli", + "provider": "codex", + "labels": { + "needs": "needs-codex-audit", + "done": "codex-audit-done", + "blocked": "codex-audit-blocked" + } + } + } + from code_mower import config as policy_config + self.assertEqual(policy_config.validate_config(config_with_codex_audit), []) + + report = lane_status.collect_status( + lineage_config=config_with_codex_audit, + repo="owner/repo", + gh_json_runner=gh_json, + command_runner=lambda _args: _completed(""), + now=NOW, + ) + + pr = report["remote"]["pull_requests"][0] + self.assertEqual(pr["lineage"]["status"], "unmanaged") + self.assertEqual(pr["lineage"]["reason"], "no_code_mower_provenance") + + def test_configured_author_recognized_when_exclusion_disabled(self) -> None: + def gh_json(args: list[str]) -> object: + if args[:2] == ["pr", "list"]: + return [ + { + "number": 777, + "title": "Bot PR with exclusion disabled", + "url": "https://github.com/owner/repo/pull/777", + "headRefName": "bot/work", + "headRefOid": "abcdef0123456789abcdef0123456789abcdef01", + "author": {"login": "source-bot"}, + "isDraft": False, + "mergeStateStatus": "CLEAN", + "updatedAt": NOW.isoformat().replace("+00:00", "Z"), + "labels": [], + "statusCheckRollup": [], + } + ] + if args[:2] == ["run", "list"]: + return [] + if args[0] == "api" and "/comments?" in args[1]: + raise RuntimeError("Simulated error") + raise lane_status.LaneStatusUnavailable("unexpected gh call") + + config_no_exclusion = policy() + config_no_exclusion["merge_authority_excludes_author"] = False + + report = lane_status.collect_status( + lineage_config=config_no_exclusion, + repo="owner/repo", + gh_json_runner=gh_json, + command_runner=lambda _args: _completed(""), + now=NOW, + ) + + pr = report["remote"]["pull_requests"][0] + self.assertEqual(pr["lineage"]["status"], "unknown") + self.assertEqual(pr["lineage"]["reason"], "lineage_unreadable") + + def test_configured_prefix_recognized_when_exclusion_disabled(self) -> None: + def gh_json(args: list[str]) -> object: + if args[:2] == ["pr", "list"]: + return [ + { + "number": 888, + "title": "Branch prefix with exclusion disabled", + "url": "https://github.com/owner/repo/pull/888", + "headRefName": "codex/prefix-work", + "headRefOid": "abcdef0123456789abcdef0123456789abcdef01", + "author": {"login": "human"}, + "isDraft": False, + "mergeStateStatus": "CLEAN", + "updatedAt": NOW.isoformat().replace("+00:00", "Z"), + "labels": [], + "statusCheckRollup": [], + } + ] + if args[:2] == ["run", "list"]: + return [] + if args[0] == "api" and "/comments?" in args[1]: + raise RuntimeError("Simulated error") + raise lane_status.LaneStatusUnavailable("unexpected gh call") + + config_no_exclusion = policy() + config_no_exclusion["merge_authority_excludes_author"] = False + + report = lane_status.collect_status( + lineage_config=config_no_exclusion, + repo="owner/repo", + gh_json_runner=gh_json, + command_runner=lambda _args: _completed(""), + now=NOW, + ) + + pr = report["remote"]["pull_requests"][0] + self.assertEqual(pr["lineage"]["status"], "unknown") + self.assertEqual(pr["lineage"]["reason"], "lineage_unreadable") + + def test_non_builder_identity_label_does_not_generate_dispatch_claim(self) -> None: + def gh_json(args: list[str]) -> object: + if args[:2] == ["pr", "list"]: + return [ + { + "number": 999, + "title": "Custom identity PR", + "url": "https://github.com/owner/repo/pull/999", + "headRefName": "feature/custom", + "headRefOid": "abcdef0123456789abcdef0123456789abcdef01", + "author": {"login": "developer"}, + "isDraft": False, + "mergeStateStatus": "CLEAN", + "updatedAt": NOW.isoformat().replace("+00:00", "Z"), + "labels": [{"name": "dispatched:custom-bot"}], + "statusCheckRollup": [], + } + ] + if args[:2] == ["run", "list"]: + return [] + if args[0] == "api" and "/comments?" in args[1]: + return [] + raise lane_status.LaneStatusUnavailable("unexpected gh call") + + config = policy() + config["builder_identity"]["labels"]["identity:custom-bot"] = "custom" + from code_mower import config as policy_config + self.assertEqual(policy_config.validate_config(config), []) + + report = lane_status.collect_status( + lineage_config=config, + repo="owner/repo", + gh_json_runner=gh_json, + command_runner=lambda _args: _completed(""), + now=NOW, + ) + + pr = report["remote"]["pull_requests"][0] + self.assertEqual(pr["lineage"]["status"], "unmanaged") + self.assertEqual(pr["lineage"]["reason"], "no_code_mower_provenance") + + def test_pr_with_malformed_lineage_marker_remains_actionable(self) -> None: + def gh_json(args: list[str]) -> object: + if args[:2] == ["pr", "list"]: + return [ + { + "number": 888, + "title": "PR with malformed lineage", + "url": "https://github.com/owner/repo/pull/888", + "headRefName": "feature/malformed", + "headRefOid": "badc0ffeebadc0ffeebadc0ffeebadc0ffeebadc", + "author": {"login": "developer"}, + "isDraft": False, + "mergeStateStatus": "CLEAN", + "updatedAt": NOW.isoformat().replace("+00:00", "Z"), + "labels": [], + "statusCheckRollup": [], + } + ] + if args[:2] == ["run", "list"]: + return [] + if args[0] == "api" and "/comments?" in args[1]: + return [ + { + "user": {"login": "lineage-publisher[bot]"}, + "body": "", + "created_at": NOW.isoformat(), + } + ] + raise lane_status.LaneStatusUnavailable("unexpected gh call") + + config = policy() + from code_mower import config as policy_config + self.assertEqual(policy_config.validate_config(config), []) + + report = lane_status.collect_status( + lineage_config=config, + repo="owner/repo", + gh_json_runner=gh_json, + command_runner=lambda _args: _completed(""), + now=NOW, + ) + + pr = report["remote"]["pull_requests"][0] + self.assertEqual(pr["lineage"]["status"], "unknown") + self.assertEqual(pr["lineage"]["reason"], "lineage_unreadable") + + def test_pr_with_conflicting_lineage_marker_remains_actionable(self) -> None: + def gh_json(args: list[str]) -> object: + if args[:2] == ["pr", "list"]: + return [ + { + "number": 777, + "title": "PR with conflicting lineage", + "url": "https://github.com/owner/repo/pull/777", + "headRefName": "feature/conflict", + "headRefOid": "deadbeefdeadbeefdeadbeefdeadbeefdeadbeef", + "author": {"login": "developer"}, + "isDraft": False, + "mergeStateStatus": "CLEAN", + "updatedAt": NOW.isoformat().replace("+00:00", "Z"), + "labels": [], + "statusCheckRollup": [], + } + ] + if args[:2] == ["run", "list"]: + return [] + if args[0] == "api" and "/comments?" in args[1]: + return [ + { + "user": {"login": "lineage-publisher[bot]"}, + "body": '', + "created_at": NOW.isoformat(), + } + ] + raise lane_status.LaneStatusUnavailable("unexpected gh call") + + config = policy() + from code_mower import config as policy_config + self.assertEqual(policy_config.validate_config(config), []) + + report = lane_status.collect_status( + lineage_config=config, + repo="owner/repo", + gh_json_runner=gh_json, + command_runner=lambda _args: _completed(""), + now=NOW, + ) + + pr = report["remote"]["pull_requests"][0] + self.assertEqual(pr["lineage"]["status"], "unknown") + self.assertEqual(pr["lineage"]["reason"], "lineage_unreadable") + + def test_pr_with_untrusted_lineage_marker_remains_unmanaged(self) -> None: + def gh_json(args: list[str]) -> object: + if args[:2] == ["pr", "list"]: + return [ + { + "number": 666, + "title": "PR with untrusted marker", + "url": "https://github.com/owner/repo/pull/666", + "headRefName": "feature/untrusted", + "headRefOid": "abcdef0123456789abcdef0123456789abcdef99", + "author": {"login": "random-user"}, + "isDraft": False, + "mergeStateStatus": "CLEAN", + "updatedAt": NOW.isoformat().replace("+00:00", "Z"), + "labels": [], + "statusCheckRollup": [], + } + ] + if args[:2] == ["run", "list"]: + return [] + if args[0] == "api" and "/comments?" in args[1]: + return [ + { + "user": {"login": "untrusted-user"}, + "body": "", + "created_at": NOW.isoformat(), + } + ] + raise lane_status.LaneStatusUnavailable("unexpected gh call") + + config = policy() + from code_mower import config as policy_config + self.assertEqual(policy_config.validate_config(config), []) + + report = lane_status.collect_status( + lineage_config=config, + repo="owner/repo", + gh_json_runner=gh_json, + command_runner=lambda _args: _completed(""), + now=NOW, + ) + + pr = report["remote"]["pull_requests"][0] + self.assertEqual(pr["lineage"]["status"], "unmanaged") + self.assertEqual(pr["lineage"]["reason"], "no_code_mower_provenance") + + def test_pr_with_invalid_policy_remains_actionable(self) -> None: + def gh_json(args: list[str]) -> object: + if args[:2] == ["pr", "list"]: + return [ + { + "number": 555, + "title": "PR with invalid policy", + "url": "https://github.com/owner/repo/pull/555", + "headRefName": "feature/test", + "headRefOid": "abcdef0123456789abcdef0123456789abcdef01", + "author": {"login": "developer"}, + "isDraft": False, + "mergeStateStatus": "CLEAN", + "updatedAt": NOW.isoformat().replace("+00:00", "Z"), + "labels": [], + "statusCheckRollup": [], + } + ] + if args[:2] == ["run", "list"]: + return [] + if args[0] == "api" and "/comments?" in args[1]: + return [] + raise lane_status.LaneStatusUnavailable("unexpected gh call") + + invalid_config = {"version": "invalid", "project": {}} + + report = lane_status.collect_status( + lineage_config=invalid_config, + repo="owner/repo", + gh_json_runner=gh_json, + command_runner=lambda _args: _completed(""), + now=NOW, + ) + + pr = report["remote"]["pull_requests"][0] + self.assertEqual(pr["lineage"]["status"], "unknown") + self.assertEqual(pr["lineage"]["reason"], "lineage_unreadable") + + def test_pr_with_malformed_labels_remains_actionable(self) -> None: + def gh_json(args: list[str]) -> object: + if args[:2] == ["pr", "list"]: + return [ + { + "number": 444, + "title": "PR with malformed labels", + "url": "https://github.com/owner/repo/pull/444", + "headRefName": "feature/test", + "headRefOid": "abcdef0123456789abcdef0123456789abcdef01", + "author": {"login": "developer"}, + "isDraft": False, + "mergeStateStatus": "CLEAN", + "updatedAt": NOW.isoformat().replace("+00:00", "Z"), + "labels": ["not-a-dict", {"wrong": "structure"}], + "statusCheckRollup": [], + } + ] + if args[:2] == ["run", "list"]: + return [] + if args[0] == "api" and "/comments?" in args[1]: + return [] + raise lane_status.LaneStatusUnavailable("unexpected gh call") + + config = policy() + from code_mower import config as policy_config + self.assertEqual(policy_config.validate_config(config), []) + + report = lane_status.collect_status( + lineage_config=config, + repo="owner/repo", + gh_json_runner=gh_json, + command_runner=lambda _args: _completed(""), + now=NOW, + ) + + pr = report["remote"]["pull_requests"][0] + self.assertEqual(pr["lineage"]["status"], "unknown") + self.assertEqual(pr["lineage"]["reason"], "lineage_unreadable") + + def test_pr_with_malformed_author_remains_actionable(self) -> None: + def gh_json(args: list[str]) -> object: + if args[:2] == ["pr", "list"]: + return [ + { + "number": 333, + "title": "PR with malformed author", + "url": "https://github.com/owner/repo/pull/333", + "headRefName": "feature/test", + "headRefOid": "abcdef0123456789abcdef0123456789abcdef01", + "author": {"login": 12345}, + "isDraft": False, + "mergeStateStatus": "CLEAN", + "updatedAt": NOW.isoformat().replace("+00:00", "Z"), + "labels": [], + "statusCheckRollup": [], + } + ] + if args[:2] == ["run", "list"]: + return [] + if args[0] == "api" and "/comments?" in args[1]: + return [] + raise lane_status.LaneStatusUnavailable("unexpected gh call") + + config = policy() + from code_mower import config as policy_config + self.assertEqual(policy_config.validate_config(config), []) + + report = lane_status.collect_status( + lineage_config=config, + repo="owner/repo", + gh_json_runner=gh_json, + command_runner=lambda _args: _completed(""), + now=NOW, + ) + + pr = report["remote"]["pull_requests"][0] + self.assertEqual(pr["lineage"]["status"], "unknown") + self.assertEqual(pr["lineage"]["reason"], "lineage_unreadable") + + def test_pr_with_malformed_visible_metadata_remains_actionable(self) -> None: + base_pr = { + "number": 334, + "title": "PR with malformed identity metadata", + "url": "https://github.com/owner/repo/pull/334", + "headRefName": "feature/test", + "headRefOid": "abcdef0123456789abcdef0123456789abcdef01", + "author": {"login": "developer"}, + "isDraft": False, + "mergeStateStatus": "CLEAN", + "updatedAt": NOW.isoformat().replace("+00:00", "Z"), + "labels": [], + "statusCheckRollup": [], + } + + malformed_cases = ( + ("labels collection", {"labels": {"name": "not-a-list"}}), + ("empty label", {"labels": [{"name": " "}]}), + ("author login", {"author": {"login": " "}}), + ("branch", {"headRefName": 123}), + ("head SHA", {"headRefOid": 123}), + ("PR number", {"number": "334"}), + ) + for case, overrides in malformed_cases: + with self.subTest(case=case): + raw_pr = {**base_pr, **overrides} + + def gh_json(args: list[str], raw_pr: dict[str, object] = raw_pr) -> object: + if args[:2] == ["pr", "list"]: + return [raw_pr] + if args[:2] == ["run", "list"]: + return [] + if args[0] == "api" and "/comments?" in args[1]: + return [] + raise lane_status.LaneStatusUnavailable("unexpected gh call") + + report = lane_status.collect_status( + lineage_config=policy(), + repo="owner/repo", + gh_json_runner=gh_json, + command_runner=lambda _args: _completed(""), + now=NOW, + ) + + pr = report["remote"]["pull_requests"][0] + self.assertEqual(pr["lineage"]["status"], "unknown") + self.assertEqual(pr["lineage"]["reason"], "lineage_unreadable") + self.assertEqual(pr["next_action"], "owner action required") + + def test_code_mower_claim_uses_all_raw_checks_not_bounded_projection(self) -> None: + unrelated_checks = [ + {"__typename": "CheckRun", "name": f"package / shard-{index}", "conclusion": "SUCCESS"} + for index in range(8) + ] + + def gh_json(args: list[str]) -> object: + if args[:2] == ["pr", "list"]: + return [ + { + "number": 335, + "title": "Claim after display limit", + "url": "https://github.com/owner/repo/pull/335", + "headRefName": "feature/test", + "headRefOid": "abcdef0123456789abcdef0123456789abcdef01", + "author": {"login": "developer"}, + "isDraft": False, + "mergeStateStatus": "CLEAN", + "updatedAt": NOW.isoformat().replace("+00:00", "Z"), + "labels": [], + "statusCheckRollup": [ + *unrelated_checks, + {"__typename": "CheckRun", "name": "code-mower/gate", "conclusion": "SUCCESS"}, + ], + } + ] + if args[:2] == ["run", "list"]: + return [] + if args[0] == "api" and "/comments?" in args[1]: + raise lane_status.LaneStatusUnavailable("history unavailable") + raise lane_status.LaneStatusUnavailable("unexpected gh call") + + report = lane_status.collect_status( + lineage_config=policy(), + repo="owner/repo", + gh_json_runner=gh_json, + command_runner=lambda _args: _completed(""), + now=NOW, + ) + + pr = report["remote"]["pull_requests"][0] + self.assertEqual(len(pr["checks"]), 8) + self.assertNotIn("code-mower/gate", {check["name"] for check in pr["checks"]}) + self.assertEqual(pr["lineage"]["status"], "unavailable") + self.assertEqual(pr["lineage"]["reason"], "lineage_unreadable") + + def test_exact_check_identity_namespace_controls_provenance_claim(self) -> None: + cases = ( + ("name", "code-mower/gate", True), + ("context", " CODE_MOWER/GATE ", True), + ("workflowName", "Code Mower CI", True), + ("workflowName", "code_mower local audit request", True), + ("name", "code-mower", True), + ("name", "not-code-mower/gate", False), + ("context", "code-mower-simulator", False), + ("workflowName", "third-party code-mower compatibility", False), + ("name", "code_mower_simulator", False), + ("context", "code-mowerish/gate", False), + ("workflowName", "Not Code Mower CI", False), + ) + + for field, value, expected in cases: + with self.subTest(field=field, value=value): + self.assertEqual( + lane_status._has_code_mower_check_claim([{field: value}]), + expected, + ) + + def test_check_identity_lookalikes_remain_unmanaged_for_all_history_states(self) -> None: + lookalikes = ( + {"__typename": "CheckRun", "name": "not-code-mower/gate"}, + {"__typename": "StatusContext", "context": "code-mower-simulator"}, + { + "__typename": "CheckRun", + "name": "package", + "workflowName": "third-party code-mower compatibility", + }, + ) + + for raw_check in lookalikes: + for history_available in (True, False): + with self.subTest(raw_check=raw_check, history_available=history_available): + def gh_json( + args: list[str], + raw_check: dict[str, object] = raw_check, + history_available: bool = history_available, + ) -> object: + if args[:2] == ["pr", "list"]: + return [ + { + "number": 340, + "title": "Ordinary PR with unrelated check identity", + "url": "https://github.com/owner/repo/pull/340", + "headRefName": "feature/test", + "headRefOid": "abcdef0123456789abcdef0123456789abcdef01", + "author": {"login": "developer"}, + "isDraft": False, + "mergeStateStatus": "CLEAN", + "updatedAt": NOW.isoformat().replace("+00:00", "Z"), + "labels": [], + "statusCheckRollup": [raw_check], + } + ] + if args[:2] == ["run", "list"]: + return [] + if args[0] == "api" and "/comments?" in args[1]: + if history_available: + return [] + raise lane_status.LaneStatusUnavailable("history unavailable") + raise lane_status.LaneStatusUnavailable("unexpected gh call") + + report = lane_status.collect_status( + lineage_config=policy(), + repo="owner/repo", + gh_json_runner=gh_json, + command_runner=lambda _args: _completed(""), + now=NOW, + ) + + pr = report["remote"]["pull_requests"][0] + self.assertEqual(pr["lineage"]["status"], "unmanaged") + self.assertEqual(pr["lineage"]["reason"], "no_code_mower_provenance") + + def test_malformed_check_collection_or_identity_remains_actionable(self) -> None: + malformed_rollups = ( + {"name": "code-mower/gate"}, + ["not-a-check-mapping"], + [{"__typename": "CheckRun", "name": 123, "conclusion": "SUCCESS"}], + [{"__typename": "StatusContext", "context": 123, "state": "SUCCESS"}], + [{"__typename": "UnknownCheck", "name": "package", "conclusion": "SUCCESS"}], + [{"__typename": "CheckRun", "name": "package", "app": "github-actions"}], + [{"name": "package", "context": "external-ci"}], + [{"__typename": "CheckRun", "name": "package", "app": {}}], + [{"__typename": "CheckRun", "name": "package", "app": {"slug": ""}}], + [{"__typename": "CheckRun", "name": "package", "app": {"owner": 42}}], + ) + + for raw_checks in malformed_rollups: + with self.subTest(raw_checks=raw_checks): + def gh_json(args: list[str], raw_checks: object = raw_checks) -> object: + if args[:2] == ["pr", "list"]: + return [ + { + "number": 336, + "title": "PR with malformed checks", + "url": "https://github.com/owner/repo/pull/336", + "headRefName": "feature/test", + "headRefOid": "abcdef0123456789abcdef0123456789abcdef01", + "author": {"login": "developer"}, + "isDraft": False, + "mergeStateStatus": "CLEAN", + "updatedAt": NOW.isoformat().replace("+00:00", "Z"), + "labels": [], + "statusCheckRollup": raw_checks, + } + ] + if args[:2] == ["run", "list"]: + return [] + if args[0] == "api" and "/comments?" in args[1]: + return [] + raise lane_status.LaneStatusUnavailable("unexpected gh call") + + report = lane_status.collect_status( + lineage_config=policy(), + repo="owner/repo", + gh_json_runner=gh_json, + command_runner=lambda _args: _completed(""), + now=NOW, + ) + + pr = report["remote"]["pull_requests"][0] + self.assertEqual(pr["lineage"]["status"], "unknown") + self.assertEqual(pr["lineage"]["reason"], "lineage_unreadable") + self.assertEqual(pr["next_action"], "owner action required") + + def test_valid_empty_and_unrelated_check_variants_remain_unmanaged(self) -> None: + valid_rollups = ( + [], + [ + { + "__typename": "CheckRun", + "name": "package", + "workflowName": "quality", + "detailsUrl": "https://github.com/owner/repo/actions/runs/1", + "startedAt": "2026-09-01T11:40:00Z", + "completedAt": "2026-09-01T11:41:00Z", + "status": "COMPLETED", + "conclusion": "SUCCESS", + "app": {"slug": "github-actions", "name": "GitHub Actions", "databaseId": 15368}, + }, + { + "__typename": "StatusContext", + "context": "external-ci", + "targetUrl": "https://ci.example.test/build/1", + "startedAt": "2026-09-01T11:40:00Z", + "state": "SUCCESS", + }, + ], + ) + + for raw_checks in valid_rollups: + with self.subTest(raw_checks=raw_checks): + def gh_json(args: list[str], raw_checks: list[dict[str, object]] = raw_checks) -> object: + if args[:2] == ["pr", "list"]: + return [ + { + "number": 337, + "title": "Ordinary PR with readable checks", + "url": "https://github.com/owner/repo/pull/337", + "headRefName": "feature/test", + "headRefOid": "abcdef0123456789abcdef0123456789abcdef01", + "author": {"login": "developer"}, + "isDraft": False, + "mergeStateStatus": "CLEAN", + "updatedAt": NOW.isoformat().replace("+00:00", "Z"), + "labels": [], + "statusCheckRollup": raw_checks, + } + ] + if args[:2] == ["run", "list"]: + return [] + if args[0] == "api" and "/comments?" in args[1]: + return [] + raise lane_status.LaneStatusUnavailable("unexpected gh call") + + report = lane_status.collect_status( + lineage_config=policy(), + repo="owner/repo", + gh_json_runner=gh_json, + command_runner=lambda _args: _completed(""), + now=NOW, + ) + + pr = report["remote"]["pull_requests"][0] + self.assertEqual(pr["lineage"]["status"], "unmanaged") + self.assertEqual(pr["lineage"]["reason"], "no_code_mower_provenance") + + def test_status_check_union_and_app_identity_matrix(self) -> None: + valid_rollups = ( + [], + [{"__typename": "CheckRun", "name": "package"}], + [{"__typename": "StatusContext", "context": "external-ci"}], + [{"name": "package"}], + [{"context": "external-ci"}], + [{"__typename": "CheckRun", "name": "gate", "app": {"slug": "code-mower"}}], + [{"__typename": "CheckRun", "name": "gate", "app": {"name": "Code Mower"}}], + [{ + "__typename": "CheckRun", + "name": "package", + "app": {"slug": "github-actions", "name": "GitHub Actions"}, + }], + [{"__typename": "CheckRun", "name": "package", "app": {"databaseId": 15368}}], + ) + invalid_rollups = ( + None, + {}, + [None], + [{"name": "package", "context": "external-ci"}], + [{"__typename": "CheckRun", "name": "package", "context": "external-ci"}], + [{"__typename": "CheckRun", "context": "external-ci"}], + [{"__typename": "CheckRun", "name": "package", "targetUrl": "https://ci.example.test"}], + [{"__typename": "CheckRun", "name": "package", "state": "SUCCESS"}], + [{"__typename": "StatusContext", "name": "package"}], + [{"__typename": "StatusContext", "name": "package", "context": "external-ci"}], + [{"__typename": "StatusContext", "context": "external-ci", "workflowName": "quality"}], + [{"__typename": "StatusContext", "context": "external-ci", "detailsUrl": "https://ci.example.test"}], + [{"__typename": "StatusContext", "context": "external-ci", "app": {"slug": "code-mower"}}], + [{"context": "external-ci", "workflowName": "quality"}], + [{"__typename": "Other", "name": "package"}], + [{"__typename": "CheckRun", "name": "package", "app": {}}], + [{"__typename": "CheckRun", "name": "package", "app": {"owner": 42}}], + [{"__typename": "CheckRun", "name": "package", "app": {"slug": ""}}], + [{"__typename": "CheckRun", "name": "package", "app": {"slug": "---"}}], + [{"__typename": "CheckRun", "name": "package", "app": {"name": None}}], + [{"__typename": "CheckRun", "name": "package", "app": {"databaseId": 0}}], + [{"__typename": "CheckRun", "name": "package", "app": {"databaseId": False}}], + [{ + "__typename": "CheckRun", + "name": "gate", + "app": {"slug": "github-actions", "name": "Code Mower"}, + }], + ) + + for rollup in valid_rollups: + with self.subTest(valid=rollup): + self.assertTrue(lane_status._status_check_rollup_is_readable(rollup)) + for rollup in invalid_rollups: + with self.subTest(invalid=rollup): + self.assertFalse(lane_status._status_check_rollup_is_readable(rollup)) + + def test_exact_app_identity_controls_provenance_claim(self) -> None: + cases = ( + ({"slug": "code-mower"}, "unavailable"), + ({"name": "Code Mower"}, "unavailable"), + ({"slug": "CODE_MOWER", "name": "Code Mower"}, "unavailable"), + ({"slug": "code.mower"}, "unavailable"), + ({"slug": "not-code-mower"}, "unmanaged"), + ({"slug": "code-mower-simulator"}, "unmanaged"), + ({"name": "Third Party Code Mower"}, "unmanaged"), + ({"slug": "github-actions", "name": "GitHub Actions"}, "unmanaged"), + ({"slug": "github-actions", "name": "Code Mower"}, "unknown"), + ({"slug": "code-mower", "name": "Code Mower Simulator"}, "unknown"), + ) + + for app, expected_status in cases: + with self.subTest(app=app): + def gh_json(args: list[str], app: dict[str, object] = app) -> object: + if args[:2] == ["pr", "list"]: + return [ + { + "number": 338, + "title": "App identity check", + "url": "https://github.com/owner/repo/pull/338", + "headRefName": "feature/test", + "headRefOid": "abcdef0123456789abcdef0123456789abcdef01", + "author": {"login": "developer"}, + "isDraft": False, + "mergeStateStatus": "CLEAN", + "updatedAt": NOW.isoformat().replace("+00:00", "Z"), + "labels": [], + "statusCheckRollup": [ + { + "__typename": "CheckRun", + "name": "gate", + "app": app, + "conclusion": "SUCCESS", + } + ], + } + ] + if args[:2] == ["run", "list"]: + return [] + if args[0] == "api" and "/comments?" in args[1]: + raise lane_status.LaneStatusUnavailable("history unavailable") + raise lane_status.LaneStatusUnavailable("unexpected gh call") + + report = lane_status.collect_status( + lineage_config=policy(), + repo="owner/repo", + gh_json_runner=gh_json, + command_runner=lambda _args: _completed(""), + now=NOW, + ) + + pr = report["remote"]["pull_requests"][0] + self.assertEqual(pr["lineage"]["status"], expected_status) + self.assertEqual(pr["lineage"]["reason"], "lineage_unreadable" if expected_status != "unmanaged" else "no_code_mower_provenance") + + def test_malformed_comment_history_stays_unknown_and_actionable(self) -> None: + malformed_pages = ( + {"comments": []}, + [None], + [{"body": None, "user": {"login": "developer"}}], + [{"body": "", "user": "developer"}], + [{"body": "", "user": {"login": 123}}], + [{ + "body": "", + "user": {"login": "developer"}, + "author": {"login": "other-developer"}, + }], + ) + + for malformed_page in malformed_pages: + with self.subTest(malformed_page=malformed_page): + def gh_json(args: list[str], malformed_page: object = malformed_page) -> object: + if args[:2] == ["pr", "list"]: + return [ + { + "number": 339, + "title": "Malformed comment history", + "url": "https://github.com/owner/repo/pull/339", + "headRefName": "feature/test", + "headRefOid": "abcdef0123456789abcdef0123456789abcdef01", + "author": {"login": "developer"}, + "isDraft": False, + "mergeStateStatus": "CLEAN", + "updatedAt": NOW.isoformat().replace("+00:00", "Z"), + "labels": [], + "statusCheckRollup": [], + } + ] + if args[:2] == ["run", "list"]: + return [] + if args[0] == "api" and "/comments?" in args[1]: + return malformed_page + raise lane_status.LaneStatusUnavailable("unexpected gh call") + + report = lane_status.collect_status( + lineage_config=policy(), + repo="owner/repo", + gh_json_runner=gh_json, + command_runner=lambda _args: _completed(""), + now=NOW, + ) + + pr = report["remote"]["pull_requests"][0] + self.assertEqual(pr["lineage"]["status"], "unknown") + self.assertEqual(pr["lineage"]["reason"], "lineage_unreadable") + self.assertEqual(pr["next_action"], "owner action required") diff --git a/tests/test_lineage_consumer_projection.py b/tests/test_lineage_consumer_projection.py index 8a3926e8..ec98c92d 100644 --- a/tests/test_lineage_consumer_projection.py +++ b/tests/test_lineage_consumer_projection.py @@ -108,7 +108,8 @@ def test_status_global_budget_retains_unknown_targets(self): def gh(args): if args[:2] == ['pr', 'list']: return [{'number': i, 'headRefName': 'codex/topic', 'headRefOid': HEAD, - 'author': {'login': 'human'}, 'labels': [{'name': 'builder:codex'}]} for i in range(1, 11)] + 'author': {'login': 'human'}, 'labels': [{'name': 'builder:codex'}], + 'statusCheckRollup': []} for i in range(1, 11)] if args[0] == 'api': calls.append(args) return [{}]*100