diff --git a/code-mower-package-manifest.json b/code-mower-package-manifest.json index 1a8fc6f6..9ab45d8d 100644 --- a/code-mower-package-manifest.json +++ b/code-mower-package-manifest.json @@ -1447,6 +1447,11 @@ "source": "src/code_mower/remote_session_cli.py", "target": "src/code_mower/remote_session_cli.py" }, + { + "kind": "core", + "source": "src/code_mower/review_authority.py", + "target": "src/code_mower/review_authority.py" + }, { "kind": "core", "source": "src/code_mower/reviewer_metrics.py", diff --git a/docs/orchestrator-prompt-pack.md b/docs/orchestrator-prompt-pack.md index b82d4915..93654457 100644 --- a/docs/orchestrator-prompt-pack.md +++ b/docs/orchestrator-prompt-pack.md @@ -213,6 +213,47 @@ Grok Build, Gemini CLI, Hermes CLI, or Devin, stay informational unless the repository has already promoted that lane under docs/lane-promotion-policy.md. ``` +## Optional Devin Setup Prompt + +Use this only when the owner has asked for Devin. The default adoption is +Claude + Codex, and this prompt adds no Devin work to a repository that did not +select it. Role qualification and session leases stay as described in +docs/participant-qualification.md; selecting a transport grants no review or +merge authority. + +```text +The owner selected Devin for OWNER/REPO. + +Report the current posture first: run code-mower doctor CONFIG --profile PROFILE +--devin and read the selected transport, readiness, and next actions. Use the +same CONFIG and PROFILE the repository actually uses; a bare run inspects the +packaged starter instead. A checkout that has no code-mower.yml has no path to +name: run code-mower doctor --packaged-starter --profile PROFILE --devin, which +selects the maintained packaged starter wherever this installation keeps it, and +use --packaged-starter in place of CONFIG in the commands below. That selector +ignores cwd-local config files and keeps the profile you name, so it reports the +same posture from any directory; --easy does not, because it is a first-run +profile alias whose starter fallback depends on what the working directory +contains. Never substitute the starter for a repository configuration to shorten +a command; it inspects a different posture. + +To change transports, preview the selection with code-mower init CONFIG +--profile PROFILE --set-transport devin=devin_api_v3 --dry-run, then stage it +with --apply --output-dir .code-mower.generated. Staging writes only that review +tree: the active posture keeps reporting the installed configuration until the +generated files are reviewed and installed through the normal setup PR. Rerun +the doctor command above afterwards to confirm the switch. + +If the repository still carries .github/workflows/devin-audit-bridge.yml or +.github/workflows/devin-audit-labeler.yml, report them as superseded by the +maintained Sessions API v3 transport and propose removing exactly those files in +the same reviewed PR. Do not delete or rewrite repository-owned workflow files +yourself. + +Do not set up owner credentials, do not start paid sessions, and do not treat an +ordinary CLI login as campaign readiness. +``` + ## Owner Status Prompt Use this when asking an orchestrator for a compact operating snapshot. diff --git a/src/code_mower/claude_audit_pr.py b/src/code_mower/claude_audit_pr.py index ed753441..db497c52 100644 --- a/src/code_mower/claude_audit_pr.py +++ b/src/code_mower/claude_audit_pr.py @@ -26,6 +26,7 @@ from typing import Any, Dict, List, Optional, Tuple from code_mower import context_audit, context_delivery, context_review +from code_mower.review_authority import AuthorityRequest as ReviewAuthorityRequest if __package__ in {None, "", "tools"}: try: @@ -248,6 +249,10 @@ class ClaudeAuditConfig: include_decision_context: bool = True decision_authorities: Tuple[str, ...] = () merge_authority: bool = True + # When set, `merge_authority` above is only the fail-closed value used until + # `audit_pr()` resolves the posture against the base revision it fetched. + # Direct callers that decided authority themselves leave this unset. + authority_request: Optional[ReviewAuthorityRequest] = None calibration_badge: str = "" context_revision: Optional[str] = None context_state_dir: Optional[Path] = None @@ -297,6 +302,11 @@ class DiffContext: full_diff_bytes: int included_diff_bytes: int adaptive_expanded: bool = False + # The commit the base ref resolved to once fetched, which is the revision + # this diff was taken against. Defaulted so historical fixtures and direct + # constructions keep working; an empty value means "not recorded", not + # "the working tree". + fetched_base_ref: str = "" def __iter__(self): """Preserve the historical `(stat, diff, truncated)` unpacking API.""" @@ -876,6 +886,52 @@ def _run_git_limited( return _decode_limited_diff(chunks, truncated=truncated), observed_bytes, truncated +def _resolve_fetched_authority( + config: ClaudeAuditConfig, local_repo: Path, base_revision: str +) -> ClaudeAuditConfig: + """Return `config` with the posture resolved against the fetched base. + + The wrapper cannot decide authority before the audit runs: the local base ref + may be stale or missing until the diff context fetches it, so a posture + computed then would describe the policy the fetch replaced — keeping + merge-authority wording through a repository demotion, or reporting an + unavailable base that is simply not fetched yet. The review diffs against the + fetched revision, so the rendered header is resolved against that same one. + + Callers that decided authority themselves pass no request and are returned + unchanged, which keeps direct `ClaudeAuditConfig` use and fixtures working. + """ + request = config.authority_request + if request is None: + return config + posture = request.resolve( + repo_root=local_repo, base_ref=base_revision or config.base_ref + ) + print( + " review authority: " + f"{posture['label']} ({posture['policy_source']}/{posture['reason']})", + file=sys.stderr, + flush=True, + ) + return replace(config, merge_authority=posture["merge_authority"]) + + +class _FetchedHeadMismatchWithBase(FetchedHeadMismatch): + """A force-push race that still knows which base revision was fetched. + + The base is fetched before the head mismatch is detected, so the stale + notice can be rendered against that same revision. Subclassing keeps every + existing `except FetchedHeadMismatch` handler and the shared exception + contract unchanged. + """ + + def __init__( + self, expected_sha: str, actual_sha: str, fetched_base_ref: str + ) -> None: + super().__init__(expected_sha, actual_sha) + self.fetched_base_ref = fetched_base_ref + + def _build_diff_context( local_repo: Path, pr_number: int, @@ -905,7 +961,9 @@ def _build_diff_context( expected_head_sha=expected_head_sha, ) if fetched_head_ref.lower() != expected_head_sha.lower(): - raise FetchedHeadMismatch(expected_head_sha, fetched_head_ref) + raise _FetchedHeadMismatchWithBase( + expected_head_sha, fetched_head_ref, fetched_base_ref + ) diff_range = f"{fetched_base_ref}...{fetched_head_ref}" stat = _run_git(local_repo, ["diff", "--stat", "--find-renames", diff_range]) changed_files_text = _run_git( @@ -935,6 +993,7 @@ def _build_diff_context( full_diff_bytes=full_diff_bytes, included_diff_bytes=len(included_diff.encode("utf-8")), adaptive_expanded=adaptive_expanded, + fetched_base_ref=fetched_base_ref, ) @@ -1349,6 +1408,13 @@ def audit_pr(config: ClaudeAuditConfig, repo: str, pr_number: int) -> ClaudeAudi config.max_diff_hard_limit_bytes, ) except FetchedHeadMismatch as exc: + # The base was fetched before the head mismatch was detected, so the + # stale notice still renders the posture of the refreshed base. + config = replace( + config, + base_ref=getattr(exc, "fetched_base_ref", "") or config.base_ref, + ) + config = _resolve_fetched_authority(config, local_repo, config.base_ref) actions_run_id = os.environ.get("GITHUB_RUN_ID") or None print( f" force-push race: fetched head {exc.actual_sha[:8]} does not " @@ -1447,6 +1513,17 @@ def audit_pr(config: ClaudeAuditConfig, repo: str, pr_number: int) -> ClaudeAudi ) return result + # The diff was taken against the revision the base ref resolved to once + # fetched. Pin it onto the config, because `base_ref` was a mutable name + # until here and everything below reads it -- the rendered posture, the + # trusted-ref lookups, the review doctrine load and the review prompt's own + # base. Carrying the SHA means the comment and the diff describe the one + # revision this audit fetched even if the tracking ref moves afterwards. A + # context that recorded no revision keeps the name it was given. + config = replace( + config, base_ref=diff_context.fetched_base_ref or config.base_ref + ) + config = _resolve_fetched_authority(config, local_repo, config.base_ref) print( f" diff budget: {diff_context.diagnostics()}", file=sys.stderr, @@ -1943,7 +2020,14 @@ def _parse_args(argv: Optional[List[str]] = None) -> argparse.Namespace: or _env_flag("CLAUDE_AUDIT_NO_SPEND_CAPTURE"), help="do not append this audit run to reviewer-spend.json", ) - posture_default = _env_flag_default("CLAUDE_AUDIT_MERGE_AUTHORITY", True) + # Unset means "render the posture this repository actually configures". + # An explicit flag or env override stays authoritative, so an operator can + # still state a posture for a checkout that configures no lanes. + posture_default = ( + _env_flag_default("CLAUDE_AUDIT_MERGE_AUTHORITY", True) + if os.environ.get("CLAUDE_AUDIT_MERGE_AUTHORITY") is not None + else None + ) ap.add_argument( "--merge-authority", dest="merge_authority", @@ -1957,6 +2041,14 @@ def _parse_args(argv: Optional[List[str]] = None) -> argparse.Namespace: action="store_false", help="Render audit comments as informational-only lane comments.", ) + ap.add_argument( + "--code-mower-config", + default=os.environ.get("CODE_MOWER_CONFIG") or None, + help=( + "repository configuration whose review lane decides the rendered " + "posture; defaults to code-mower.yml in the audited checkout" + ), + ) ap.add_argument( "--calibration-badge", default=os.environ.get("CLAUDE_AUDIT_CALIBRATION_BADGE", ""), @@ -2019,6 +2111,16 @@ def main(argv: Optional[List[str]] = None) -> int: try: repo_paths = _parse_repo_paths(args.repo_paths) _validate_repo_path_for_wrapper(repo_paths, args.repo) + # The posture is resolved inside audit_pr, against the base revision the + # diff is taken from; the local base ref here may still be stale or + # missing. Only an explicitly selected configuration is checked now, so + # an operator typo fails before any network work rather than mid-audit. + authority_request = ReviewAuthorityRequest( + product="claude", + config_path=args.code_mower_config, + override=args.merge_authority, + ) + authority_request.validate() config = ClaudeAuditConfig( github_token=token, repo_paths=repo_paths, @@ -2041,7 +2143,10 @@ def main(argv: Optional[List[str]] = None) -> int: max_plan_context_bytes=args.max_plan_context_bytes, max_plan_context_file_bytes=args.max_plan_context_file_bytes, include_decision_context=not args.no_decision_context, - merge_authority=args.merge_authority, + # Fail closed until the fetched base decides: nothing rendered before + # that resolution may claim authority this run has not verified. + merge_authority=False, + authority_request=authority_request, calibration_badge=args.calibration_badge, ) result = audit_pr(config, args.repo, args.pr) diff --git a/src/code_mower/cli.py b/src/code_mower/cli.py index 1f119e05..5377b404 100644 --- a/src/code_mower/cli.py +++ b/src/code_mower/cli.py @@ -516,6 +516,7 @@ def _init_main(argv: list[str]) -> int: if ( argv[:1] == ["auth"] or _has_flag(argv, "--easy") + or _has_flag(argv, "--packaged-starter") or _has_positional_config(argv, options_with_values) ): return code_mower_init.main(argv) diff --git a/src/code_mower/codex_audit_pr.py b/src/code_mower/codex_audit_pr.py index f19007f3..d278d15c 100644 --- a/src/code_mower/codex_audit_pr.py +++ b/src/code_mower/codex_audit_pr.py @@ -70,6 +70,7 @@ from code_mower import context_audit, context_delivery, context_review from code_mower.context_contract import ContextError +from code_mower.review_authority import AuthorityRequest as ReviewAuthorityRequest import argparse import json @@ -288,6 +289,11 @@ class AuditConfig: # reference provider catalog's Codex audit posture; pass --informational # when replaying or calibrating a lane that is not a repository gate. merge_authority: bool = True + # When set, `merge_authority` above is only the fail-closed value used until + # `audit_pr()` resolves the posture against the base revision it fetches. + # Direct callers that decided authority themselves leave this unset and keep + # whatever `merge_authority` they passed. + authority_request: Optional["ReviewAuthorityRequest"] = None # Optional human-facing calibration status. This must never decide merge # authority; it only renders as a separate badge line in the comment. calibration_badge: str = "" @@ -865,6 +871,52 @@ def _fetch_base_ref(local_repo: Path, base_ref: str) -> None: _shared_fetch_base_ref(local_repo, base_ref) +def _pinned_base_revision(local_repo: Path, base_ref: str) -> str: + """Return the commit `base_ref` names right after it was fetched. + + Pinning the snapshot immediately after the fetch keeps everything derived + from it describing one revision even if the tracking ref moves later in the + run. When the ref cannot be resolved the name is returned unchanged, so a + base that is genuinely unavailable stays unavailable rather than being + quietly replaced by something that resolves. + """ + try: + pinned = _run_git_text( + local_repo, ["rev-parse", "--verify", f"{base_ref}^{{commit}}"], timeout=10 + ).strip() + except (OSError, ValueError, subprocess.SubprocessError): + return base_ref + return pinned or base_ref + + +def _resolve_fetched_authority( + config: "AuditConfig", local_repo: Path, base_revision: str +) -> "AuditConfig": + """Return `config` with the posture resolved against the fetched base. + + The wrapper cannot decide authority before the audit runs: the local base ref + may be stale or missing until `audit_pr` fetches it, so a posture computed + then would describe the policy the fetch replaced — keeping merge-authority + wording through a repository demotion, or reporting an unavailable base that + is simply not fetched yet. The review compares against the fetched revision, + so the rendered header is resolved against that same revision here. + + Callers that decided authority themselves pass no request and are returned + unchanged, which keeps direct `AuditConfig` use and recorded fixtures working. + """ + request = config.authority_request + if request is None: + return config + posture = request.resolve(repo_root=local_repo, base_ref=base_revision) + print( + " review authority: " + f"{posture['label']} ({posture['policy_source']}/{posture['reason']})", + file=sys.stderr, + flush=True, + ) + return replace(config, merge_authority=posture["merge_authority"]) + + def _run_git_text(local_repo: Path, args: List[str], *, timeout: int = 60) -> str: return _shared_run_git_text(local_repo, args, timeout=timeout) @@ -1871,6 +1923,19 @@ def audit_pr(config: AuditConfig, repo: str, pr_number: int) -> AuditResult: # before running the review. Stale base = wrong diff = wrong review. _fetch_pr_head(local_repo, pr_number, head_sha_start) _fetch_base_ref(local_repo, config.base_ref) + # Pin the fetched snapshot onto the config itself. `base_ref` was a mutable + # name until here, and everything below reads it -- the rendered posture, the + # trusted-ref lookups, the review context diagnostics and the review's own + # `--base`. Carrying the SHA instead means they all describe the one revision + # this audit fetched, even if the tracking ref moves later in the run. A ref + # that will not resolve keeps its name, so an unavailable base stays + # unavailable rather than being quietly replaced by something that resolves. + config = replace( + config, base_ref=_pinned_base_revision(local_repo, config.base_ref) + ) + # The base is now the revision this review compares against, so the posture + # the comment renders is resolved here rather than before the fetch. + config = _resolve_fetched_authority(config, local_repo, config.base_ref) decision_authorities = _decision_authorities_for_repo( local_repo, config.decision_authorities, @@ -2473,7 +2538,14 @@ def _parse_args(argv: Optional[List[str]] = None) -> argparse.Namespace: or _env_flag("CODEX_AUDIT_NO_SPEND_CAPTURE"), help="do not append this audit run to reviewer-spend.json", ) - posture_default = _env_flag_default("CODEX_AUDIT_MERGE_AUTHORITY", True) + # Unset means "render the posture this repository actually configures". + # An explicit flag or env override stays authoritative, so an operator can + # still state a posture for a checkout that configures no lanes. + posture_default = ( + _env_flag_default("CODEX_AUDIT_MERGE_AUTHORITY", True) + if os.environ.get("CODEX_AUDIT_MERGE_AUTHORITY") is not None + else None + ) ap.add_argument( "--merge-authority", dest="merge_authority", @@ -2487,6 +2559,14 @@ def _parse_args(argv: Optional[List[str]] = None) -> argparse.Namespace: action="store_false", help="Render audit comments as informational-only lane comments.", ) + ap.add_argument( + "--code-mower-config", + default=os.environ.get("CODE_MOWER_CONFIG") or None, + help=( + "repository configuration whose review lane decides the rendered " + "posture; defaults to code-mower.yml in the audited checkout" + ), + ) ap.add_argument( "--calibration-badge", default=os.environ.get("CODEX_AUDIT_CALIBRATION_BADGE", ""), @@ -2616,6 +2696,16 @@ def main(argv: Optional[List[str]] = None) -> int: try: repo_paths = _parse_repo_paths(args.repo_paths) _validate_repo_path_for_wrapper(repo_paths, args.repo) + # The posture is resolved inside audit_pr, against the base revision this + # audit fetches; the local base ref here may still be stale or missing. + # Only an explicitly selected configuration is checked now, so an + # operator typo fails before any network work rather than mid-audit. + authority_request = ReviewAuthorityRequest( + product="codex", + config_path=args.code_mower_config, + override=args.merge_authority, + ) + authority_request.validate() except ValueError as exc: print(f"error: {exc}", file=sys.stderr) return 1 @@ -2662,7 +2752,10 @@ def main(argv: Optional[List[str]] = None) -> int: max_plan_context_bytes=args.max_plan_context_bytes, max_plan_context_file_bytes=args.max_plan_context_file_bytes, include_decision_context=not args.no_decision_context, - merge_authority=args.merge_authority, + # Fail closed until the fetched base decides: nothing rendered before + # that resolution may claim authority this run has not verified. + merge_authority=False, + authority_request=authority_request, calibration_badge=args.calibration_badge, ) diff --git a/src/code_mower/devin_readiness.py b/src/code_mower/devin_readiness.py index 9d2ee0ec..dfce2384 100644 --- a/src/code_mower/devin_readiness.py +++ b/src/code_mower/devin_readiness.py @@ -70,6 +70,14 @@ # transport. Rewriting the participant list instead would delete every unrelated # participant and profile lane the repository selected. TRANSPORT_OPTION = "--set-transport" +# The packaged starter configuration is selected by its own explicit selector, +# not by a path: the path it resolves to lives inside the installation that ran +# the check. `--easy` is not that selector -- it is a first-run profile alias +# whose starter fallback depends on what the working directory contains, and +# doctor and init fall back to different files -- so the supported selector that +# names the maintained package resource directly is rendered instead. +PACKAGED_STARTER_SOURCE = "packaged_starter" +PACKAGED_STARTER_OPTION = "--packaged-starter" SELECTABLE_TRANSPORTS = (LOCAL_TRANSPORT, HOSTED_TRANSPORT) CANONICAL_LANES = frozenset( entry.review_lane @@ -82,6 +90,13 @@ # through the normal setup PR. A test pins this directory to init's own default. GENERATED_OUTPUT_DIR = ".code-mower.generated" +# Installing the reviewed tree writes the repository's own configuration, and +# that installed file -- not the package resource a starter finding was read +# from -- is what the run afterwards uses. Verification therefore selects this +# path, because the packaged starter is never rewritten by an install and would +# keep reporting its unchanged transport. A test pins it to init's own default. +INSTALLED_CONFIG_PATH = "code-mower.yml" + # The public declaration fields a lane names for each transport. A repository that # named its own Devin lanes edits these fields itself: no generated command can # retarget a lane it cannot name without rebuilding the profile around it. @@ -247,6 +262,10 @@ class _Pin: config_path: str = "" profile: str = "" repo_slug: str = "" + # How the caller selected that configuration. The packaged starter is + # resolved to an installation-specific path at runtime, so a command pinned + # to it would only run on the machine that generated the finding. + config_source: str = "" # A generated transport switch can only replace canonical Devin lanes; a # custom-named lane is edited by its owner instead of rewritten. targeted_switch: bool = True @@ -256,6 +275,7 @@ def doctor(self, *, devin: bool = False, flags: tuple[str, ...] = ()) -> str: return doctor_command( config_path=self.config_path, profile=self.profile, + config_source=self.config_source, devin=devin, flags=flags, ) @@ -264,6 +284,7 @@ def readiness(self, *, repo_slug: str | None = None) -> str: return readiness_command( config_path=self.config_path, profile=self.profile, + config_source=self.config_source, repo_slug=self.repo_slug if repo_slug is None else repo_slug, ) @@ -272,6 +293,7 @@ def select(self, transport: str) -> str: transport, config_path=self.config_path, profile=self.profile, + config_source=self.config_source, targeted=self.targeted_switch, lanes=self.custom_lanes, ) @@ -748,13 +770,32 @@ def _unselected_findings(pin: _Pin) -> tuple[ReadinessFinding, ...]: ) -def _pinned(command: str, *, config_path: str, profile: str) -> str: +def _portable_starter(config_source: str, profile: str) -> bool: + """Return true when the starter selector names this posture without a path. + + The packaged starter has no repository path: it is resolved inside whichever + installation ran the check, so pinning it renders a command only that machine + can run. ``--packaged-starter`` selects exactly that maintained resource and + keeps whichever ``--profile`` the finding describes, so it stands in for the + path at every profile rather than only the recommended one. + """ + return config_source == PACKAGED_STARTER_SOURCE + + +def _pinned( + command: str, *, config_path: str, profile: str, config_source: str = "" +) -> str: """Return `command` scoped to exactly one configuration and profile. Both inputs are shell-quoted because a configuration path and a profile name - may contain spaces, and an unquoted command would inspect something else. + may contain spaces, and an unquoted command would inspect something else. The + packaged starter is named by its supported selector instead of a path, + because its resolved path belongs to one installation; the profile is still + pinned, because the selector does not choose one. """ - if config_path: + if _portable_starter(config_source, profile): + command += f" {PACKAGED_STARTER_OPTION}" + elif config_path: command += f" {shlex.quote(config_path)}" if profile: command += f" --profile {shlex.quote(profile)}" @@ -773,6 +814,7 @@ def doctor_command( *, config_path: str = "", profile: str = "", + config_source: str = "", repo_slug: str = "", devin: bool = False, flags: tuple[str, ...] = (), @@ -783,13 +825,20 @@ def doctor_command( because an unpinned check can read another profile's Devin lane, and a bare rerun can report a different posture than the finding that asked for it. A caller without those inputs gets guidance to reuse its own instead of a - command that silently inspects something else. + command that silently inspects something else. `config_source` distinguishes + a repository configuration, whose path is portable, from the packaged + starter, whose path is not. """ shown = "code-mower doctor" + (" --devin" if devin else "") shown += "".join(f" {flag}" for flag in flags) if not profile: return _unpinned_guidance(shown) - command = _pinned("code-mower doctor", config_path=config_path, profile=profile) + command = _pinned( + "code-mower doctor", + config_path=config_path, + profile=profile, + config_source=config_source, + ) if devin: command += " --devin" command += "".join(f" {flag}" for flag in flags) @@ -802,11 +851,37 @@ def readiness_command( *, config_path: str = "", profile: str = "", + config_source: str = "", repo_slug: str = "", ) -> str: """Return the `--devin` doctor command that inspects exactly this posture.""" return doctor_command( - config_path=config_path, profile=profile, repo_slug=repo_slug, devin=True + config_path=config_path, + profile=profile, + config_source=config_source, + repo_slug=repo_slug, + devin=True, + ) + + +def _installed_readiness_command( + *, config_path: str, profile: str, config_source: str +) -> str: + """Return the readiness command that confirms an installed switch took effect. + + Preview and staging select the configuration the finding was read from, but + the check that confirms the transport actually changed has to read the + configuration the repository runs afterwards. An install never rewrites the + packaged starter, so keeping its selector here would re-report the unchanged + starter transport rather than the installed one; the installed repository + configuration is named instead, at the same profile the finding describes. A + finding already sourced from a repository configuration is installed over + that same file, so its own path stays the right thing to verify. + """ + if _portable_starter(config_source, profile): + return readiness_command(config_path=INSTALLED_CONFIG_PATH, profile=profile) + return readiness_command( + config_path=config_path, profile=profile, config_source=config_source ) @@ -815,6 +890,7 @@ def select_transport_command( *, config_path: str = "", profile: str = "", + config_source: str = "", targeted: bool = True, lanes: tuple[str, ...] = (), ) -> str: @@ -826,12 +902,24 @@ def select_transport_command( nothing else, so it names the product's transport rather than a participant list, which would drop every unrelated participant and profile lane. + A finding against the packaged starter has no repository path to pin, so the + steps name that posture with its supported `--packaged-starter` selector + instead of the path it happened to resolve to inside this installation, and + still pin the profile the finding describes. A repository configuration is + never replaced by the starter to shorten a command. + `init` never rewrites the configuration it read: `--apply` stages a reviewable generated tree, so the steps are a dry-run preview, an apply into an explicit output directory, an install of the reviewed output through the normal setup PR, and only then a rerun of readiness. Claiming the posture switched because files were staged would misreport the active configuration. + That final check reads the *installed* configuration at the same profile. A + starter-sourced finding selects the package resource to preview and stage + from, but an install writes the repository's own configuration and leaves the + starter untouched, so verifying through the starter selector would report the + unchanged starter transport instead of the switch. + When the profile's Devin lanes are custom-named, no generated command can retarget them, so bounded manual guidance names those lanes instead. The Code Mower `--profile` selects the configuration profile and is never the credential @@ -841,18 +929,30 @@ def select_transport_command( raise ConfigError("Devin transport must be devin_cli or devin_api_v3") if not targeted: return custom_lane_guidance( - transport, config_path=config_path, profile=profile, lanes=lanes + transport, + config_path=config_path, + profile=profile, + config_source=config_source, + lanes=lanes, ) selection = f"{TRANSPORT_OPTION} devin={transport}" if not profile: return _unpinned_guidance(f"code-mower init {selection}") - pinned = _pinned("code-mower init", config_path=config_path, profile=profile) + pinned = _pinned( + "code-mower init", + config_path=config_path, + profile=profile, + config_source=config_source, + ) staged = shlex.quote(GENERATED_OUTPUT_DIR) + verify = _installed_readiness_command( + config_path=config_path, profile=profile, config_source=config_source + ) return ( f"preview it with `{pinned} {selection} --dry-run`, stage it with `{pinned} " f"{selection} --apply --output-dir {staged}`, then review the generated " "configuration and support files and install them through the normal setup PR " - f"before rerunning {doctor_command(config_path=config_path, profile=profile, devin=True)}" + f"before confirming the installed configuration with {verify}" "; staging writes only that review tree, so the active posture keeps reporting " "the installed configuration until the generated one replaces it, and the " "saved selection is repository-wide, so every profile selecting Devin moves " @@ -865,6 +965,7 @@ def custom_lane_guidance( *, config_path: str = "", profile: str = "", + config_source: str = "", lanes: tuple[str, ...] = (), ) -> str: """Return bounded manual guidance for retargeting custom-named Devin lanes. @@ -899,7 +1000,11 @@ def custom_lane_guidance( if lanes else "this profile's custom-named Devin lanes" ) - where = f" in {shlex.quote(config_path)}" if config_path else "" + if _portable_starter(config_source, profile): + # The starter has no repository path to edit; name the posture instead. + where = f" in the packaged starter configuration ({PACKAGED_STARTER_OPTION})" + else: + where = f" in {shlex.quote(config_path)}" if config_path else "" if profile: where += f" under profile {shlex.quote(profile)}" action = ( @@ -922,8 +1027,8 @@ def custom_lane_guidance( return f"{action}, then rerun {_unpinned_guidance('code-mower doctor --devin')}" return ( f"{action}, then rerun " - f"{doctor_command(config_path=config_path, profile=profile, devin=True)}; no " - "generated command can retarget a lane this repository named" + f"{doctor_command(config_path=config_path, profile=profile, config_source=config_source, devin=True)}" + "; no generated command can retarget a lane this repository named" ) @@ -932,13 +1037,16 @@ def setup_instructions( *, config_path: str = "", profile: str = "", + config_source: str = "", repo_slug: str = "", ) -> tuple[str, ...]: """Return host guidance for the selected optional Devin posture.""" if transport not in TRANSPORTS: raise ConfigError("Devin transport must be devin_cli or devin_api_v3") if transport == LOCAL_TRANSPORT: - check = readiness_command(config_path=config_path, profile=profile) + check = readiness_command( + config_path=config_path, profile=profile, config_source=config_source + ) authentication = ( "Devin executes locally through devin_cli: its ambient Devin Desktop/CLI " "login is the only authentication, and hosted service-user credentials do " @@ -946,7 +1054,10 @@ def setup_instructions( ) else: check = readiness_command( - config_path=config_path, profile=profile, repo_slug=repo_slug or "OWNER/REPO" + config_path=config_path, + profile=profile, + config_source=config_source, + repo_slug=repo_slug or "OWNER/REPO", ) authentication = ( "Devin executes hosted through devin_api_v3: it needs dedicated service-user " @@ -974,6 +1085,7 @@ def devin_readiness( config_profile: str | None = "recommended", config_dir: Path | None = None, config_path: str = "", + config_source: str = "", lane_config: Mapping[str, Any] | None = None, lane_id: str = "", lane_configs: tuple[tuple[str, Mapping[str, Any] | None], ...] = (), @@ -1011,6 +1123,7 @@ def devin_readiness( pin = _Pin( config_path=config_path, profile=config_profile or "", + config_source=config_source, repo_slug=repo_slug, targeted_switch=not custom_lanes, custom_lanes=custom_lanes, diff --git a/src/code_mower/doctor.py b/src/code_mower/doctor.py index 0fb301ae..b482233b 100644 --- a/src/code_mower/doctor.py +++ b/src/code_mower/doctor.py @@ -41,6 +41,7 @@ _check_cloud_token_surface = _doctor_checks.check_cloud_token_surface _evaluate_json_probe = _doctor_checks.evaluate_json_probe _local_cli_probe_remediation = _doctor_checks.local_cli_probe_remediation +render_doctor_summary = _doctor_checks.render_doctor_summary render_doctor_text = _doctor_checks.render_doctor_text resolve_doctor_config_path = _doctor_checks.resolve_doctor_config_path resolve_doctor_config_path_for_script = _doctor_checks.resolve_doctor_config_path_for_script @@ -73,10 +74,19 @@ def _doctor_config_source_label( config_path: Path, easy: bool, cwd: Path | None = None, + packaged_starter: bool = False, ) -> str: - """Classify the config source for adoption-facing doctor output.""" + """Classify the config source for adoption-facing doctor output. + + ``--packaged-starter`` names the maintained package resource directly, so it + classifies as the starter without consulting cwd-local files. ``--easy`` is + only a first-run profile alias whose starter fallback depends on what the + working directory happens to contain, so it stays a separate question. + """ cwd = cwd or Path.cwd() + if packaged_starter: + return "packaged_starter" if config_arg != "code-mower.yml": return "explicit_config" if config_path.name == "code-mower.example.yml" and easy: @@ -132,7 +142,9 @@ def main(argv: Sequence[str] | None = None) -> int: parser = argparse.ArgumentParser(description=__doc__) parser.add_argument('--context-online', action='store_true', help='Deliberately verify selected context authorization; never searches') parser.add_argument('--context-state-dir', type=Path, help='Private context store outside repositories') - parser.add_argument("config", nargs="?", default="code-mower.yml") + # Defaulted after parsing so that an explicit positional selection stays + # distinguishable from the default even when it names the same file. + parser.add_argument("config", nargs="?", default=None) parser.add_argument( "--provider-templates", default=code_mower_package.DEFAULT_PROVIDER_TEMPLATES, @@ -146,6 +158,15 @@ def main(argv: Sequence[str] | None = None) -> int: "absent, use the packaged example config" ), ) + parser.add_argument( + "--packaged-starter", + action="store_true", + help=( + "check the maintained packaged starter config, wherever this " + "installation keeps it; keeps the selected --profile and ignores " + "cwd-local config files" + ), + ) parser.add_argument( "--v05", action="store_true", @@ -301,6 +322,21 @@ def main(argv: Sequence[str] | None = None) -> int: "--operational-evidence", type=Path, help="Include closed local acceptance observations; never polls, retries or uploads", ) + detail_group = parser.add_mutually_exclusive_group() + detail_group.add_argument( + "--concise", + action="store_true", + help=( + "render a posture-scoped summary: every check still runs, but text " + "output leads with active failures and owner actions and counts the " + "remaining warnings by group" + ), + ) + detail_group.add_argument( + "--advanced", + action="store_true", + help="render every check in the text output (the default detail level)", + ) parser.add_argument("--strict", action="store_true") parser.add_argument("--json", action="store_true") args = parser.parse_args(argv) @@ -316,6 +352,18 @@ def main(argv: Sequence[str] | None = None) -> int: _apply_first_run_defaults(args) if args.easy and args.profile is None: args.profile = "recommended" + explicit_config = args.config is not None + if args.config is None: + args.config = "code-mower.yml" + if args.packaged_starter and explicit_config: + # Two contradictory explicit selections. Ignoring either one would check a + # configuration the caller did not ask for, so neither is guessed at. + print( + "error: --packaged-starter selects the maintained packaged starter; " + f"remove it or the explicit config {args.config!r}", + file=sys.stderr, + ) + return 1 try: repo_slug = "" @@ -327,7 +375,11 @@ def main(argv: Sequence[str] | None = None) -> int: repo_slug = detect_repo_slug(Path.cwd()) repo_source = "git_remote" if repo_slug else "" provider_templates_path = resolve_doctor_provider_templates_path(args.provider_templates) - config_path = resolve_doctor_config_path(args.config, easy=args.easy) + config_path = ( + code_mower_package.packaged_starter_config_path() + if args.packaged_starter + else resolve_doctor_config_path(args.config, easy=args.easy) + ) report = run_doctor( config_path=config_path, provider_templates_path=provider_templates_path, @@ -338,6 +390,7 @@ def main(argv: Sequence[str] | None = None) -> int: config_arg=args.config, config_path=config_path, easy=args.easy, + packaged_starter=args.packaged_starter, ), adoption=args.adoption, adoption_posture=args.adoption_posture, @@ -385,8 +438,14 @@ def main(argv: Sequence[str] | None = None) -> int: ),) report = replace(report, checks=report.checks + evidence_checks) + # JSON keeps the complete report, and the explicit advanced and campaign + # modes keep the full text view: a concise run only changes what a default + # text run reads first. + concise = args.concise and not args.advanced and not args.campaign if args.json: print(json.dumps(report.as_dict(), indent=2, sort_keys=True)) + elif concise: + print(render_doctor_summary(report), end="") else: print(render_doctor_text(report), end="") if report.failures: diff --git a/src/code_mower/doctor_checks/__init__.py b/src/code_mower/doctor_checks/__init__.py index 5cbef065..3fbe945b 100644 --- a/src/code_mower/doctor_checks/__init__.py +++ b/src/code_mower/doctor_checks/__init__.py @@ -63,7 +63,7 @@ DoctorReport, is_promotion_todo_check, ) -from .output import doctor_output_group, render_doctor_text +from .output import doctor_output_group, render_doctor_summary, render_doctor_text from .presets import ( apply_first_run_defaults, resolve_doctor_config_path, @@ -190,6 +190,7 @@ "local_cli_probe_remediation", "normalize_repo_slug", "provider_template_coverage", + "render_doctor_summary", "render_doctor_text", "resolve_doctor_config_path", "resolve_doctor_config_path_for_script", diff --git a/src/code_mower/doctor_checks/devin.py b/src/code_mower/doctor_checks/devin.py index 382f1ca6..232d8e8e 100644 --- a/src/code_mower/doctor_checks/devin.py +++ b/src/code_mower/doctor_checks/devin.py @@ -132,6 +132,7 @@ def check_devin_readiness( provider_config_dir: Path | None = None, config_profile: str | None = "recommended", config_path: str = "", + config_source: str = "", effective_lanes: Iterable[tuple[str, Mapping[str, Any]]] = (), adoption_posture: str = "reviewer-gate", ) -> list[DoctorCheck]: @@ -160,6 +161,7 @@ def check_devin_readiness( config_profile=config_profile, config_dir=provider_config_dir, config_path=config_path, + config_source=config_source, lane_configs=selected_lanes, adoption_posture=adoption_posture, include_unselected=include_unselected, diff --git a/src/code_mower/doctor_checks/output.py b/src/code_mower/doctor_checks/output.py index 68fc26a4..342771d7 100644 --- a/src/code_mower/doctor_checks/output.py +++ b/src/code_mower/doctor_checks/output.py @@ -74,6 +74,80 @@ def _adoption_posture_hint(check: DoctorCheck) -> bool: return check.name == "doctor.adoption.posture_hint" +def render_doctor_summary(report: DoctorReport) -> str: + """Render a posture-scoped summary of a doctor report. + + Every check still ran and every check is still in the JSON report: this view + only chooses what a first run reads first. Active failures and owner actions + are shown in full, because they are the only entries that ask for an action; + remaining warnings are counted by group so an intended posture's optional + providers do not bury them. The full text view stays one flag away and is + named here rather than assumed. + """ + lines = [ + "Code Mower doctor (concise)", + f"Status: {report.status}", + f"Config: {report.config_path}", + ] + if report.profile: + lines.append(f"Profile: {report.profile}") + lines.append(f"Checks: {_format_status_summary(report)}") + lines.append("") + for check in report.checks: + if _adoption_posture_hint(check): + lines.append( + f"Adoption posture: {check.status.upper()} {check.name}: {check.message}" + ) + if check.remediation: + lines.append(f" remediation: {check.remediation}") + lines.append("") + break + if not report.checks: + lines.append("No checks ran.") + return "\n".join(lines) + "\n" + + prioritized = [ + check + for check in report.checks + if not _adoption_posture_hint(check) + and (check.status == STATUS_FAIL or is_owner_action_check(check)) + ] + if prioritized: + lines.append("Active failures and owner actions") + for check in prioritized: + lines.extend(_format_check(check)) + lines.append("") + else: + lines.append("No active failures or owner actions.") + lines.append("") + + remaining: list[str] = [] + for group_id, checks in _group_checks(report.checks).items(): + counts = [] + promotion_todos = sum(1 for check in checks if is_promotion_todo_check(check)) + warnings = ( + sum(1 for check in checks if check.status == STATUS_WARN) + - sum(1 for check in checks if is_owner_action_check(check)) + - promotion_todos + ) + if promotion_todos: + counts.append(f"{promotion_todos} promotion todos") + if warnings: + counts.append(f"{warnings} warnings") + if counts: + label = GROUP_LABELS.get(group_id, group_id.title()) + remaining.append(f"- {label}: {', '.join(counts)}") + if remaining: + lines.append("Remaining detail by group") + lines.extend(remaining) + lines.append("") + lines.append( + "Full detail: rerun the same command with --advanced for every check, or " + "with --json for the complete report." + ) + return "\n".join(lines) + "\n" + + def render_doctor_text(report: DoctorReport) -> str: """Render a doctor report for terminal output.""" lines = [ diff --git a/src/code_mower/doctor_checks/runner.py b/src/code_mower/doctor_checks/runner.py index 5a31ae79..48d79dd4 100644 --- a/src/code_mower/doctor_checks/runner.py +++ b/src/code_mower/doctor_checks/runner.py @@ -320,6 +320,7 @@ def run_doctor( provider_config_dir=provider_config_dir, config_profile=profile, config_path=str(config_path), + config_source=config_source, effective_lanes=effective_lanes, adoption_posture=adoption_posture, ) diff --git a/src/code_mower/init.py b/src/code_mower/init.py index 0e008068..ba970690 100644 --- a/src/code_mower/init.py +++ b/src/code_mower/init.py @@ -20,6 +20,7 @@ sys.path.insert(0, str(Path(__file__).resolve().parents[1])) from code_mower import branch_policy +from code_mower import package as code_mower_package from code_mower import participants as code_mower_participants from code_mower.package_rendering import _render_provider_catalog @@ -3311,7 +3312,9 @@ def main(argv: list[str] | None = None) -> int: return _auth_main(argv[1:]) parser = argparse.ArgumentParser(description=__doc__) - parser.add_argument("config", nargs="?", default="code-mower.example.yml") + # Defaulted after parsing so that an explicit positional selection stays + # distinguishable from the default even when it names the same file. + parser.add_argument("config", nargs="?", default=None) parser.add_argument("--profile", default="recommended") parser.add_argument( "--with", dest="participants", metavar="PARTICIPANTS", @@ -3355,6 +3358,15 @@ def main(argv: list[str] | None = None) -> int: "with --apply to write generated output instead" ), ) + parser.add_argument( + "--packaged-starter", + action="store_true", + help=( + "render from the maintained packaged starter config, wherever this " + "installation keeps it; keeps the selected --profile and ignores " + "cwd-local config files" + ), + ) parser.add_argument("--dry-run", action="store_true", help="render the init plan") parser.add_argument("--apply", action="store_true", help="write generated files to --output-dir") tracker_group = parser.add_mutually_exclusive_group() @@ -3416,10 +3428,23 @@ def main(argv: list[str] | None = None) -> int: parser.add_argument("--json", action="store_true", help="emit dry-run plan as JSON") args = parser.parse_args(argv) + explicit_config = args.config is not None + if args.config is None: + args.config = PACKAGED_STARTER_CONFIG_NAME + if args.easy: args.profile = "recommended" if not args.dry_run and not args.apply: args.dry_run = True + if args.packaged_starter and explicit_config: + # Two contradictory explicit selections. Ignoring either one would render + # from a configuration the caller did not ask for, so neither is guessed. + print( + "error: --packaged-starter selects the maintained packaged starter; " + f"remove it or the explicit config {args.config!r}", + file=sys.stderr, + ) + return 1 if args.builders and not args.dry_run and not args.apply: args.dry_run = True if (args.interactive or args.participants is not None or args.set_transport is not None @@ -3443,11 +3468,17 @@ def main(argv: list[str] | None = None) -> int: return 1 try: - config_source = _resolve_config_path(args.config) - # An explicitly supplied local file keeps its identity even when its - # basename matches the starter; only a resolved packaged fallback - # counts as the packaged starter. - packaged_fallback = config_source != Path(args.config) + if args.packaged_starter: + # The explicit selector names the maintained package resource, so no + # cwd-local file named like the starter can stand in for it. + config_source = code_mower_package.packaged_starter_config_path() + packaged_fallback = True + else: + config_source = _resolve_config_path(args.config) + # An explicitly supplied local file keeps its identity even when its + # basename matches the starter; only a resolved packaged fallback + # counts as the packaged starter. + packaged_fallback = config_source != Path(args.config) rendered_config_path = ( str(config_source) if packaged_fallback else args.config ) diff --git a/src/code_mower/migration.py b/src/code_mower/migration.py index 1f0a6346..7e771fde 100644 --- a/src/code_mower/migration.py +++ b/src/code_mower/migration.py @@ -269,6 +269,23 @@ "tools/run_codex_audit_pr.sh", } ) +# The legacy Devin issue-comment bridge: a workflow that watched issue comments, +# its labeler, and the script they dispatched. The maintained Devin transport is +# the Sessions API v3 transport selected in the repository configuration, so +# these files are superseded rather than drifted. They are named exactly, never +# matched by glob, and never removed automatically: they are repository-owned +# workflow files whose deletion belongs in a reviewed PR. +SUPERSEDED_DEVIN_BRIDGE_PATHS = ( + ".github/workflows/devin-audit-bridge.yml", + ".github/workflows/devin-audit-labeler.yml", + "tools/devin_audit_bridge.py", +) +SUPERSEDED_DEVIN_BRIDGE_TRANSPORT = "devin_api_v3" +# The supported selection option. A test pins this to the option that +# devin_readiness owns, so the reported migration command cannot drift from the +# real flag while this module keeps its standalone-importable shape. +DEVIN_TRANSPORT_OPTION = "--set-transport" + STANDALONE_PIN_RELATIVE_PATH = "tools/code_mower_standalone_pin.env" STANDALONE_PIN_REF_KEY = "CODE_MOWER_STANDALONE_REF" STANDALONE_PIN_PLACEHOLDER_FRAGMENT = "pin-a-reviewed-code-mower" @@ -366,12 +383,70 @@ def _standalone_pin_drift_summary( } +def _superseded_devin_bridge_summary( + repo_path: Path, *, files: Sequence[Mapping[str, Any]] +) -> dict[str, Any]: + """Report legacy Devin issue-comment bridge files as a superseded transport. + + These files are not drift against the generated output: they are an earlier + transport that the maintained Sessions API v3 transport replaced. Reporting + them as ordinary `repo-only` entries invites deleting whatever the generator + no longer writes, so the finding names the migration instead and bounds it to + the exact files it observed. Nothing is deleted or rewritten here. + """ + classifications = { + str(item.get("path")): item for item in files if item.get("path") is not None + } + present: list[dict[str, Any]] = [] + for path in SUPERSEDED_DEVIN_BRIDGE_PATHS: + if not (repo_path / path).is_file(): + continue + item = classifications.get(path) or {} + present.append({"path": path, "tracked": bool(item.get("tracked"))}) + summary: dict[str, Any] = { + "transport": SUPERSEDED_DEVIN_BRIDGE_TRANSPORT, + "paths": [item["path"] for item in present], + "files": present, + } + if not present: + return { + **summary, + "status": "skip", + "reason": "no_superseded_bridge_files", + "next_action": "no legacy Devin issue-comment bridge files found", + } + pair = { + ".github/workflows/devin-audit-bridge.yml", + ".github/workflows/devin-audit-labeler.yml", + } + named = ", ".join(item["path"] for item in present) + return { + **summary, + "status": "warn", + "reason": ( + "superseded_bridge_pair" + if pair <= set(summary["paths"]) + else "superseded_bridge_files" + ), + "next_action": ( + "the legacy Devin issue-comment bridge and labeler are superseded by the " + f"maintained {SUPERSEDED_DEVIN_BRIDGE_TRANSPORT} Sessions API transport: " + "preview the transport selection for this configuration and profile with " + f"`code-mower init --profile {DEVIN_TRANSPORT_OPTION} " + f"devin={SUPERSEDED_DEVIN_BRIDGE_TRANSPORT} --dry-run`, then remove exactly " + f"{named} in the same reviewed PR; setup-drift never deletes or rewrites " + "repository-owned workflow files" + ), + } + + def _setup_drift_next_action( *, changed_count: int, standalone_pin: Mapping[str, Any], builder_hint: Mapping[str, Any], repo_path_hint: Mapping[str, Any] | None = None, + superseded_bridge: Mapping[str, Any] | None = None, ) -> str: file_action = "review differs, new, repo-only, and missing-from-output entries before copying generated setup" pin_warn = standalone_pin.get("status") == "warn" @@ -380,6 +455,8 @@ def _setup_drift_next_action( extras: list[str] = [] if repo_path_warn: extras.append(str(repo_path_hint["next_action"])) + if superseded_bridge and superseded_bridge.get("status") == "warn": + extras.append(str(superseded_bridge["next_action"])) if pin_warn: extras.append(str(standalone_pin["next_action"])) if builder_warn: @@ -617,6 +694,8 @@ def _is_setup_candidate_path(path: str) -> bool: return True if normalized in SETUP_DRIFT_BUILDER_PATHS: return True + if normalized in SUPERSEDED_DEVIN_BRIDGE_PATHS: + return True if normalized.startswith("docs/lanes/"): return True if normalized.startswith("tools/lane_configs/"): @@ -779,6 +858,8 @@ def render_setup_drift_report( changed_count = sum(counts[name] for name in SETUP_DRIFT_CLASSIFICATIONS if name != "same") standalone_pin = _standalone_pin_drift_summary(repo_path) standalone_pin_warn = standalone_pin["status"] == "warn" + superseded_bridge = _superseded_devin_bridge_summary(repo_path, files=files) + superseded_bridge_warn = superseded_bridge["status"] == "warn" builder_hint = _setup_drift_builder_hint( files=files, builders_supplied=bool(builders), @@ -799,6 +880,7 @@ def render_setup_drift_report( and not standalone_pin_warn and not builder_hint_warn and not repo_path_warn + and not superseded_bridge_warn ) else "warn" ), @@ -814,6 +896,7 @@ def render_setup_drift_report( "changed_count": changed_count, "repo_path_hint": repo_path_hint, "standalone_pin": standalone_pin, + "superseded_bridge": superseded_bridge, "builder_hint": builder_hint, "files": files, "next_action": _setup_drift_next_action( @@ -821,6 +904,7 @@ def render_setup_drift_report( standalone_pin=standalone_pin, builder_hint=builder_hint, repo_path_hint=repo_path_hint, + superseded_bridge=superseded_bridge, ), } @@ -872,6 +956,17 @@ def render_setup_drift_text(payload: dict[str, Any], *, limit: int = 50) -> str: "", ] ) + superseded_bridge = payload.get("superseded_bridge") or {} + if superseded_bridge and superseded_bridge.get("status") == "warn": + lines.extend( + [ + f"Superseded transport: WARN {superseded_bridge['reason']} " + f"superseded_by={superseded_bridge['transport']} " + f"files={', '.join(superseded_bridge['paths'])}", + f"Superseded transport next: {superseded_bridge['next_action']}", + "", + ] + ) repo_path_hint = payload.get("repo_path_hint") or {} if repo_path_hint and repo_path_hint.get("status") == "warn": lines.extend( diff --git a/src/code_mower/package.py b/src/code_mower/package.py index ab90edab..b86b7912 100644 --- a/src/code_mower/package.py +++ b/src/code_mower/package.py @@ -152,6 +152,32 @@ def _running_code_mower_version(repo_root: Path | None = None) -> str: return "0.0.0" +PACKAGED_STARTER_CONFIG_NAME = "code-mower.example.yml" + + +def packaged_starter_config_path() -> Path: + """Return the maintained packaged starter configuration. + + The starter is a package resource, so it is located relative to the installed + module and never relative to the working directory: a cwd-local file named + like the starter must not be able to redirect an explicit packaged-starter + selection into a different configuration. A source checkout keeps its + repository copy next to the package, which is the same maintained file. + """ + module_dir = Path(__file__).resolve().parent + candidates = ( + module_dir / "templates" / PACKAGED_STARTER_CONFIG_NAME, + module_dir.parents[1] / PACKAGED_STARTER_CONFIG_NAME, + ) + for candidate in candidates: + if candidate.is_file(): + return candidate + raise ConfigError( + "packaged starter configuration is missing from this installation; " + "reinstall code-mower or pass an explicit config path" + ) + + def resolve_package_config_path(path_text: str, *, explicit: bool = False) -> Path: path = Path(path_text) if explicit or path_text != DEFAULT_PACKAGE_CONFIG or path.is_absolute(): diff --git a/src/code_mower/package_manifest.py b/src/code_mower/package_manifest.py index 5368e15d..d9387afb 100644 --- a/src/code_mower/package_manifest.py +++ b/src/code_mower/package_manifest.py @@ -313,6 +313,7 @@ ("src/code_mower/participants.py", "src/code_mower/participants.py", "core"), ("src/code_mower/role_eligibility.py", "src/code_mower/role_eligibility.py", "core"), ("src/code_mower/role_eligibility.schema.json", "src/code_mower/role_eligibility.schema.json", "schema"), + ("src/code_mower/review_authority.py", "src/code_mower/review_authority.py", "core"), ("src/code_mower/provider_capabilities.py", "src/code_mower/provider_capabilities.py", "core"), ("src/code_mower/provider_capabilities.schema.json", "src/code_mower/provider_capabilities.schema.json", "schema"), ("src/code_mower/session.py", "src/code_mower/session.py", "core"), diff --git a/src/code_mower/review_authority.py b/src/code_mower/review_authority.py new file mode 100644 index 00000000..15c5dc15 --- /dev/null +++ b/src/code_mower/review_authority.py @@ -0,0 +1,305 @@ +"""Effective configured review authority for rendered status and audit comments. + +Authority is never conferred by installation, a lane label, or a provider +identity. It is the maintained role decision in :mod:`role_eligibility`, +narrowed by the review lane declaration this run actually selected, so rendered +status must describe that computed result instead of a wrapper default. No new +policy is decided here: the lane declaration and ``decide_role`` are read, and +only the wording is produced. + +Historical comments keep the wording they recorded when they were posted. This +module answers a different question -- what the current configured posture is -- +so a past ``merge-authority lane`` header stays readable evidence without +becoming a claim about the posture of this run. +""" + +from __future__ import annotations + +from dataclasses import dataclass +from pathlib import Path +from typing import Any, Mapping + +from .participants import PARTICIPANTS, reference_review_config +from .provider_capabilities import normalize_lane +from .provider_registry import REFERENCE_PROVIDERS +from .role_eligibility import decide_role +from .yaml_subset import ConfigError + +SCHEMA = "code_mower.reviewAuthority.v1" + +# Comment and session wording. The audit header wording is unchanged so that +# existing comment parsers and recorded fixtures keep matching. +MERGE_AUTHORITY_LABEL = "merge-authority lane" +INFORMATIONAL_LABEL = "informational only" +SESSION_MERGE_AUTHORITY_LABEL = "merge-authority lane" +SESSION_INFORMATIONAL_LABEL = "informational lane" + +REPOSITORY_CONFIG_FILENAME = "code-mower.yml" +# Implicit discovery reads the repository's active policy, which lives on the +# trusted base ref rather than in the PR-head checkout an audit runs against. +DEFAULT_BASE_REF = "origin/main" +# The trusted base could not be read at all, which is different from a base that +# verifiably tracks no configuration. An unknown policy supports no authority +# claim, so this source always renders informational with an actionable reason. +TRUSTED_BASE_UNAVAILABLE = "trusted_base_unavailable" +TRUSTED_BASE_UNAVAILABLE_ACTION = ( + "fetch the base ref this audit compares against (for example " + "`git fetch origin main`) or pass --code-mower-config with the repository " + "configuration to report, then rerun the audit" +) + + +@dataclass(frozen=True) +class AuthorityRequest: + """The wrapper inputs that resolve a posture once the base ref is fetched. + + A wrapper knows its product, its operator override, and any explicitly + selected configuration before it starts, but it does not yet know the base + revision the review will compare against: the local base ref can be stale or + entirely missing until the audit fetches it. Resolving before that fetch + reports the policy the fetch is about to replace, or reports the base as + unavailable when it is merely not fetched yet -- while the review itself uses + the refreshed revision. Carrying the inputs rather than an answer lets the + one resolver in this module run against the revision the review actually + used, so the rendered header and the review describe the same base. + """ + + product: str + config_path: str | Path | None = None + override: bool | None = None + + def validate(self) -> None: + """Fail now on an explicitly selected configuration that cannot be read. + + Resolution happens mid-audit, after network work has already started, so + an operator typo in ``--code-mower-config`` is surfaced up front instead. + Nothing about the base ref is consulted here: it is not fetched yet, and + an unfetched base is not an error. + """ + if self.config_path is not None: + resolve_repository_config(config_path=self.config_path) + + def resolve(self, *, repo_root: str | Path | None, base_ref: str) -> dict[str, Any]: + """Return the posture for `base_ref`, the revision this audit fetched.""" + return effective_merge_authority( + self.product, + config_path=self.config_path, + repo_root=repo_root, + base_ref=base_ref, + override=self.override, + ) + + +def authority_label(payload: Mapping[str, Any], *, session: bool = False) -> str: + """Render the posture wording for a resolved authority payload.""" + if payload.get("merge_authority"): + return SESSION_MERGE_AUTHORITY_LABEL if session else MERGE_AUTHORITY_LABEL + return SESSION_INFORMATIONAL_LABEL if session else INFORMATIONAL_LABEL + + +def review_authority( + product: str, + *, + config: Mapping[str, Any] | None = None, + lane: str | None = None, + transport: str | None = None, + config_source: str = "packaged_default", +) -> dict[str, Any]: + """Return the effective review posture for `product` under `config`. + + The lane declaration decides the configured posture, and the maintained role + decision can only narrow it: a lane that declares merge authority but whose + reviewer role is denied, incapable, or unqualified renders as informational + rather than claiming an authority this run does not have. The reverse is not + possible here, because an informational lane is passed to ``decide_role`` as + informational and no eligible decision can widen it. + """ + lane_id = lane or (PARTICIPANTS[product].review_lane if product in PARTICIPANTS else None) + if not lane_id: + raise ConfigError(f"{product} has no review lane to report authority for") + lanes = config.get("lanes") if isinstance(config, Mapping) else None + declared = lanes.get(lane_id) if isinstance(lanes, Mapping) else None + if isinstance(declared, Mapping): + policy_source = "repository" + declaration = normalize_lane(lane_id, declared, config=config) + elif lane_id in REFERENCE_PROVIDERS: + policy_source = "starter" + declaration = normalize_lane(lane_id, reference_review_config(lane_id), config=config) + else: + raise ConfigError(f"unknown review lane {lane_id!r}") + configured = bool(declaration.get("merge_authority")) and not bool( + declaration.get("informational") + ) + qualification = declaration.get("role_qualification") + decision = decide_role( + product, + "reviewer", + transport=transport, + config=config, + merge_authority=configured, + qualification=qualification if isinstance(qualification, str) else None, + ) + narrowed = decision["scope"] != "unrestricted" or decision["status"] == "ineligible" + merge_authority = configured and not narrowed + if not configured: + reason = "lane_informational" + elif narrowed: + reason = decision["reason"] if decision["status"] == "ineligible" else "role_scope_informational" + else: + reason = "lane_merge_authority" + payload = { + "schema": SCHEMA, + "product": product, + "lane": lane_id, + "policy_source": policy_source, + "config_source": config_source, + "configured_merge_authority": configured, + "merge_authority": merge_authority, + "scope": decision["scope"], + "reason": reason, + "eligibility": decision, + } + payload["label"] = authority_label(payload) + return payload + + +def _trusted_base_config(repo_root: Path, base_ref: str) -> tuple[Mapping[str, Any] | None, str]: + """Read `code-mower.yml` as the trusted base ref has it, never the checkout. + + An audit runs against a PR-head checkout, so the configuration sitting in the + working tree is the change under review. Reporting a posture from it would + let an unmerged promotion or demotion take effect in the comment header + before it is approved, so implicit discovery reads the base ref the same way + :func:`context_audit.required_for_repo` does. + + Only a successful trusted-tree lookup that proves the file is absent selects + the maintained lane defaults: that base configures no lane, so the default is + the repository's active policy. Everything else -- a missing or invalid base + ref, a failed or timed-out Git command, tracked configuration that does not + parse -- leaves the trusted answer unknown. Unknown is not evidence of a + posture, so it is reported as unavailable rather than resolved either to the + maintained default (which would grant starter authority nothing verified) or + to the proposed head configuration (which is exactly what must not be read). + """ + import subprocess + + from .config import _YamlSubsetParser + + try: + listing = subprocess.run( + ["git", "ls-tree", "--name-only", base_ref, "--", REPOSITORY_CONFIG_FILENAME], + cwd=repo_root, capture_output=True, text=True, check=True, timeout=10, + ) + except (OSError, ValueError, subprocess.SubprocessError): + return None, TRUSTED_BASE_UNAVAILABLE + if not listing.stdout.strip(): + return None, "packaged_default" + try: + shown = subprocess.run( + ["git", "show", f"{base_ref}:{REPOSITORY_CONFIG_FILENAME}"], + cwd=repo_root, capture_output=True, text=True, check=True, timeout=10, + ) + parsed = _YamlSubsetParser(shown.stdout).parse() + if not isinstance(parsed, Mapping): + raise ConfigError("top-level config must be a mapping") + except (OSError, ValueError, TypeError, subprocess.SubprocessError, ConfigError): + return None, TRUSTED_BASE_UNAVAILABLE + return parsed, "trusted_base_config" + + +def resolve_repository_config( + *, + config_path: str | Path | None = None, + repo_root: str | Path | None = None, + base_ref: str = DEFAULT_BASE_REF, +) -> tuple[Mapping[str, Any] | None, str]: + """Load the repository configuration that decides this run's posture. + + An explicitly selected configuration is authoritative: it is never replaced + by the packaged starter, and an unreadable one is an error rather than a + silent downgrade to a different posture. Without an explicit selection, a Git + checkout is read at its trusted base ref rather than at the head under + review, and only a base that verifiably configures nothing falls back to the + maintained lane defaults; a base that could not be read at all reports + :data:`TRUSTED_BASE_UNAVAILABLE` instead of a default posture. A directory + that is not a Git checkout has no base ref to trust, so its own file is the + configuration it runs under. + """ + from . import config as code_mower_config + + if config_path is not None: + path = Path(config_path).expanduser() + if not path.is_file(): + raise ConfigError(f"selected repository configuration not found: {path}") + return code_mower_config.load_config(path), "explicit_repository_config" + if repo_root is not None: + root = Path(repo_root).expanduser() + if base_ref and (root / ".git").exists(): + return _trusted_base_config(root, base_ref) + candidate = root / REPOSITORY_CONFIG_FILENAME + if candidate.is_file(): + return code_mower_config.load_config(candidate), "repository_config" + return None, "packaged_default" + + +def effective_merge_authority( + product: str, + *, + config_path: str | Path | None = None, + repo_root: str | Path | None = None, + base_ref: str = DEFAULT_BASE_REF, + lane: str | None = None, + override: bool | None = None, +) -> dict[str, Any]: + """Resolve the posture an audit wrapper should render for this run. + + The configured decision is computed first, and an operator override is read + against it rather than instead of it. An override can only narrow: a flag or + environment value asking for merge authority cannot grant it to an + informational lane, a denied role policy, an unavailable capability or an + unqualified role, and it never skips an explicitly selected configuration + that could not be read. An override asking for informational is always + honoured, and whichever source decided the rendered posture is named. + + A trusted base that could not be read leaves the repository's policy unknown, + so the posture renders informational with a bounded action rather than + granting the packaged starter's defaults; a positive override cannot widen + that either. + """ + config, config_source = resolve_repository_config( + config_path=config_path, repo_root=repo_root, base_ref=base_ref + ) + payload = review_authority( + product, config=config, lane=lane, config_source=config_source + ) + if config_source == TRUSTED_BASE_UNAVAILABLE: + # No trusted policy was read, so nothing here is evidence of merge + # authority. The lane defaults computed above describe the packaged + # starter, not this repository, so the rendered posture is the bounded + # non-authoritative one and says what would make it resolvable. + payload["configured_merge_authority"] = False + payload["merge_authority"] = False + payload["scope"] = "informational" + payload["policy_source"] = "unavailable" + payload["reason"] = TRUSTED_BASE_UNAVAILABLE + payload["action"] = TRUSTED_BASE_UNAVAILABLE_ACTION + payload["label"] = authority_label(payload) + if override is None: + return payload + payload["operator_override"] = override + if not override or payload["merge_authority"]: + # Narrowing to informational, or agreeing with the computed posture: the + # operator decided the rendered result either way. + payload["merge_authority"] = override + payload["policy_source"] = "operator" + payload["config_source"] = "operator_override" + payload["reason"] = "operator_override" + if not override: + payload["scope"] = "informational" + else: + # A positive override cannot widen what the configuration narrowed; the + # computed reason stays the rendered one so the header is not a claim the + # repository's policy does not support. + payload["override_ignored"] = True + payload["label"] = authority_label(payload) + return payload diff --git a/src/code_mower/session.py b/src/code_mower/session.py index ec6fcabf..b24ce9b8 100644 --- a/src/code_mower/session.py +++ b/src/code_mower/session.py @@ -232,7 +232,11 @@ def render_session(payload: Mapping[str, Any]) -> str: roles.append("builder via " + mode.replace("_", " ")) if member["reviewer"]: review = member["reviewer"] - policy = "merge-authority lane" if review["merge_authority"] else "informational lane" + # Single-sourced wording: the session payload already carries the + # effective lane posture and its role decision, so rendering reuses + # the shared label instead of restating the policy here. + from .review_authority import authority_label + policy = authority_label(review, session=True) roles.append(f"reviewer: {review['lane']} ({policy})") lines.append(f"- {member['name']}: {', '.join(roles)}") execution = member.get("execution") diff --git a/tests/test_adoption_polish_955.py b/tests/test_adoption_polish_955.py new file mode 100644 index 00000000..b36a2d09 --- /dev/null +++ b/tests/test_adoption_polish_955.py @@ -0,0 +1,1360 @@ +"""Adoption polish: effective review authority, superseded bridge, concise doctor.""" + +from __future__ import annotations + +import contextlib +import io +import json +import os +import subprocess +import sys +import unittest +from pathlib import Path +from unittest import mock + +sys.path.insert(0, str(Path(__file__).resolve().parents[1] / "src")) + +from code_mower import devin_readiness, init as code_mower_init, migration, review_authority, session +from code_mower.doctor_checks.models import DoctorCheck, DoctorReport +from code_mower.doctor_checks.output import render_doctor_summary, render_doctor_text +from code_mower.yaml_subset import ConfigError + + +def _review_lane(**overrides): + lane = { + "type": "review", + "driver": "claude_cli", + "provider": "claude", + "labels": {"needs": "needs-claude-audit", "done": "claude-audit-done", "blocked": "claude-audit-blocked"}, + "merge_authority": True, + "informational": False, + } + lane.update(overrides) + return lane + + +class EffectiveReviewAuthorityTests(unittest.TestCase): + def test_starter_lane_keeps_maintained_merge_authority(self): + payload = review_authority.review_authority("claude") + self.assertTrue(payload["merge_authority"]) + self.assertEqual(payload["label"], "merge-authority lane") + self.assertEqual(payload["policy_source"], "starter") + self.assertEqual(payload["reason"], "lane_merge_authority") + + def test_informational_repository_lane_renders_informational(self): + config = { + "lanes": { + "claude_audit": _review_lane(merge_authority=False, informational=True) + } + } + payload = review_authority.review_authority("claude", config=config) + self.assertFalse(payload["merge_authority"]) + self.assertEqual(payload["label"], "informational only") + self.assertEqual(payload["policy_source"], "repository") + self.assertEqual(payload["reason"], "lane_informational") + self.assertEqual(payload["scope"], "informational") + + def test_qualified_lane_narrowed_by_denied_role_policy(self): + config = { + "lanes": {"claude_audit": _review_lane()}, + "role_policy": {"claude": {"reviewer": {"enabled": False}}}, + } + payload = review_authority.review_authority("claude", config=config) + self.assertTrue(payload["configured_merge_authority"]) + self.assertFalse(payload["merge_authority"]) + self.assertEqual(payload["reason"], "policy_denied") + self.assertEqual(payload["label"], "informational only") + + def test_codex_lane_posture_is_read_per_product(self): + config = { + "lanes": { + "codex": _review_lane( + driver="codex_cli", + provider="codex", + merge_authority=False, + informational=True, + ) + } + } + self.assertFalse(review_authority.review_authority("codex", config=config)["merge_authority"]) + # An unconfigured lane for another product is unaffected. + self.assertTrue(review_authority.review_authority("claude", config=config)["merge_authority"]) + + def test_operator_override_is_reported_as_the_source(self): + payload = review_authority.effective_merge_authority("claude", override=False) + self.assertFalse(payload["merge_authority"]) + self.assertEqual(payload["policy_source"], "operator") + self.assertEqual(payload["config_source"], "operator_override") + self.assertEqual(payload["label"], "informational only") + + def test_session_rendering_uses_the_shared_label(self): + payload = { + "repo": "o/r", + "host": "claude", + "orchestrator": "claude", + "status": "prepared", + "participants": [ + { + "id": "claude", + "name": "Claude Code", + "builder": None, + "note": "", + "reviewer": { + "lane": "claude_audit", + "merge_authority": False, + "informational": True, + "policy_source": "repository", + "readiness": "unchecked", + }, + } + ], + "instructions": [], + } + text = session.render_session(payload) + self.assertIn("reviewer: claude_audit (informational lane)", text) + self.assertNotIn("merge-authority", text) + + +class HistoricalFixtureTests(unittest.TestCase): + """Recorded wording stays readable without becoming the configured posture.""" + + def test_recorded_header_is_not_a_configured_posture_claim(self): + recorded = "## Claude audit (merge-authority lane)\n\nHead SHA: `abc`\n" + self.assertIn(review_authority.MERGE_AUTHORITY_LABEL, recorded) + config = { + "lanes": { + "claude_audit": _review_lane(merge_authority=False, informational=True) + } + } + current = review_authority.review_authority("claude", config=config) + self.assertEqual(current["label"], review_authority.INFORMATIONAL_LABEL) + + def test_labels_are_stable_strings(self): + self.assertEqual(review_authority.MERGE_AUTHORITY_LABEL, "merge-authority lane") + self.assertEqual(review_authority.INFORMATIONAL_LABEL, "informational only") + self.assertEqual(review_authority.SESSION_INFORMATIONAL_LABEL, "informational lane") + + +class NonWideningOverrideTests(unittest.TestCase): + """A positive flag or environment value can never widen computed authority.""" + + def _config(self, path: Path, body: str) -> Path: + path.write_text(body, encoding="utf-8") + return path + + def setUp(self): + import tempfile + + self.root = Path(tempfile.mkdtemp()) + self.addCleanup(lambda: __import__("shutil").rmtree(self.root, ignore_errors=True)) + + def test_positive_override_cannot_widen_an_informational_lane(self): + config = self._config( + self.root / "informational.yml", + "version: 1\n" + "lanes:\n" + " claude_audit:\n" + " type: review\n" + " driver: claude_cli\n" + " provider: claude\n" + " merge_authority: false\n" + " informational: true\n" + " labels:\n" + " needs: needs-claude-audit\n" + " done: claude-audit-done\n" + " blocked: claude-audit-blocked\n", + ) + payload = review_authority.effective_merge_authority( + "claude", config_path=config, override=True + ) + self.assertFalse(payload["merge_authority"]) + self.assertEqual(payload["label"], "informational only") + self.assertEqual(payload["reason"], "lane_informational") + self.assertTrue(payload["override_ignored"]) + self.assertEqual(payload["policy_source"], "repository") + + def test_positive_override_cannot_widen_a_denied_role_policy(self): + config = self._config( + self.root / "denied.yml", + "version: 1\n" + "role_policy:\n" + " codex:\n" + " reviewer:\n" + " enabled: false\n", + ) + payload = review_authority.effective_merge_authority( + "codex", config_path=config, override=True + ) + self.assertFalse(payload["merge_authority"]) + self.assertEqual(payload["reason"], "policy_denied") + self.assertTrue(payload["override_ignored"]) + + def test_positive_override_is_honoured_when_the_configuration_agrees(self): + payload = review_authority.effective_merge_authority("claude", override=True) + self.assertTrue(payload["merge_authority"]) + self.assertEqual(payload["policy_source"], "operator") + self.assertEqual(payload["reason"], "operator_override") + self.assertNotIn("override_ignored", payload) + + def test_negative_override_still_narrows_a_merge_authority_lane(self): + payload = review_authority.effective_merge_authority("codex", override=False) + self.assertFalse(payload["merge_authority"]) + self.assertEqual(payload["scope"], "informational") + self.assertEqual(payload["policy_source"], "operator") + + def test_override_does_not_skip_an_explicitly_missing_configuration(self): + for override in (True, False): + with self.subTest(override=override): + with self.assertRaises(ConfigError): + review_authority.effective_merge_authority( + "claude", + config_path=self.root / "absent.yml", + override=override, + ) + + +class RepositoryConfigSelectionTests(unittest.TestCase): + def test_explicit_missing_config_is_an_error_not_a_starter_fallback(self): + with self.assertRaises(ConfigError): + review_authority.resolve_repository_config(config_path="no-such-config.yml") + + def test_checkout_without_config_falls_back_to_maintained_defaults(self): + config, source = review_authority.resolve_repository_config( + repo_root=Path(__file__).resolve().parent / "does-not-exist" + ) + self.assertIsNone(config) + self.assertEqual(source, "packaged_default") + + +class TrustedBaseAuthorityTests(unittest.TestCase): + """Implicit discovery reads active policy, not the change under review.""" + + LANE = ( + "version: 1\n" + "lanes:\n" + " claude_audit:\n" + " type: review\n" + " driver: claude_cli\n" + " provider: claude\n" + " merge_authority: {authority}\n" + " informational: {informational}\n" + " labels:\n" + " needs: needs-claude-audit\n" + " done: claude-audit-done\n" + " blocked: claude-audit-blocked\n" + ) + + def _git(self, *args: str) -> None: + subprocess.run( + ["git", *args], cwd=self.root, check=True, capture_output=True, text=True + ) + + def setUp(self): + import tempfile + + self.root = Path(tempfile.mkdtemp()) + self.addCleanup(lambda: __import__("shutil").rmtree(self.root, ignore_errors=True)) + self._git("init", "--initial-branch", "main") + self._git("config", "user.email", "lane@example.invalid") + self._git("config", "user.name", "Lane") + self._git("config", "commit.gpgsign", "false") + + def _commit(self, body: str, message: str) -> None: + (self.root / "code-mower.yml").write_text(body, encoding="utf-8") + self._git("add", "code-mower.yml") + self._git("commit", "-m", message) + + def test_a_pr_promoting_its_own_lane_reports_the_base_policy(self): + self._commit( + self.LANE.format(authority="false", informational="true"), "base policy" + ) + # The checkout is the PR head, which proposes merge authority. + (self.root / "code-mower.yml").write_text( + self.LANE.format(authority="true", informational="false"), encoding="utf-8" + ) + payload = review_authority.effective_merge_authority( + "claude", repo_root=self.root, base_ref="main" + ) + self.assertFalse(payload["merge_authority"]) + self.assertEqual(payload["config_source"], "trusted_base_config") + self.assertEqual(payload["reason"], "lane_informational") + + def test_a_pr_demoting_its_own_lane_also_reports_the_base_policy(self): + self._commit( + self.LANE.format(authority="true", informational="false"), "base policy" + ) + (self.root / "code-mower.yml").write_text( + self.LANE.format(authority="false", informational="true"), encoding="utf-8" + ) + payload = review_authority.effective_merge_authority( + "claude", repo_root=self.root, base_ref="main" + ) + self.assertTrue(payload["merge_authority"]) + self.assertEqual(payload["config_source"], "trusted_base_config") + + def test_a_base_without_a_configuration_keeps_the_maintained_default(self): + (self.root / "README.md").write_text("x\n", encoding="utf-8") + self._git("add", "README.md") + self._git("commit", "-m", "no config") + (self.root / "code-mower.yml").write_text( + self.LANE.format(authority="false", informational="true"), encoding="utf-8" + ) + config, source = review_authority.resolve_repository_config( + repo_root=self.root, base_ref="main" + ) + self.assertIsNone(config) + self.assertEqual(source, "packaged_default") + + def test_unavailable_discovery_never_falls_back_to_the_head_checkout(self): + self._commit( + self.LANE.format(authority="false", informational="true"), "base policy" + ) + (self.root / "code-mower.yml").write_text( + self.LANE.format(authority="true", informational="false"), encoding="utf-8" + ) + config, source = review_authority.resolve_repository_config( + repo_root=self.root, base_ref="refs/heads/no-such-base" + ) + self.assertIsNone(config) + self.assertEqual(source, review_authority.TRUSTED_BASE_UNAVAILABLE) + + def test_a_missing_base_ref_never_grants_the_maintained_default(self): + # A base that could not be read proves nothing. Reporting the starter's + # defaults would claim merge authority no trusted policy supports. + self._commit( + self.LANE.format(authority="true", informational="false"), "base policy" + ) + payload = review_authority.effective_merge_authority( + "codex", repo_root=self.root, base_ref="refs/heads/no-such-base" + ) + self.assertFalse(payload["merge_authority"]) + self.assertFalse(payload["configured_merge_authority"]) + self.assertEqual( + payload["config_source"], review_authority.TRUSTED_BASE_UNAVAILABLE + ) + self.assertEqual(payload["reason"], review_authority.TRUSTED_BASE_UNAVAILABLE) + self.assertEqual(payload["label"], review_authority.INFORMATIONAL_LABEL) + self.assertTrue(payload["action"]) + + def test_malformed_tracked_configuration_is_unavailable_not_default(self): + # Tracked, but not a configuration mapping at all. + self._commit("- not\n- a mapping\n", "malformed base policy") + config, source = review_authority.resolve_repository_config( + repo_root=self.root, base_ref="main" + ) + self.assertIsNone(config) + self.assertEqual(source, review_authority.TRUSTED_BASE_UNAVAILABLE) + + def test_a_failed_git_lookup_is_unavailable_not_default(self): + self._commit( + self.LANE.format(authority="true", informational="false"), "base policy" + ) + broken = self.root / "broken" + broken.mkdir() + (broken / ".git").write_text("not a git dir\n", encoding="utf-8") + config, source = review_authority.resolve_repository_config( + repo_root=broken, base_ref="main" + ) + self.assertIsNone(config) + self.assertEqual(source, review_authority.TRUSTED_BASE_UNAVAILABLE) + + def test_a_positive_override_cannot_widen_an_unavailable_base(self): + self._commit( + self.LANE.format(authority="true", informational="false"), "base policy" + ) + payload = review_authority.effective_merge_authority( + "claude", + repo_root=self.root, + base_ref="refs/heads/no-such-base", + override=True, + ) + self.assertFalse(payload["merge_authority"]) + self.assertTrue(payload["override_ignored"]) + + def test_verified_absence_is_distinguished_from_unavailability(self): + # Proving the base tracks no configuration is evidence; it selects the + # maintained defaults, and nothing else in this class does. + (self.root / "README.md").write_text("x\n", encoding="utf-8") + self._git("add", "README.md") + self._git("commit", "-m", "no config") + _, absent = review_authority.resolve_repository_config( + repo_root=self.root, base_ref="main" + ) + _, unavailable = review_authority.resolve_repository_config( + repo_root=self.root, base_ref="refs/heads/no-such-base" + ) + self.assertEqual(absent, "packaged_default") + self.assertEqual(unavailable, review_authority.TRUSTED_BASE_UNAVAILABLE) + + def test_an_explicit_selection_still_wins_over_the_trusted_base(self): + self._commit( + self.LANE.format(authority="true", informational="false"), "base policy" + ) + selected = self.root / "selected.yml" + selected.write_text( + self.LANE.format(authority="false", informational="true"), encoding="utf-8" + ) + payload = review_authority.effective_merge_authority( + "claude", config_path=selected, repo_root=self.root, base_ref="main" + ) + self.assertFalse(payload["merge_authority"]) + self.assertEqual(payload["config_source"], "explicit_repository_config") + + +class FetchedBaseAuthorityTests(unittest.TestCase): + """Both wrappers render the posture of the base revision they fetched. + + The local base ref can be stale or absent when a wrapper starts. Resolving + then reports the policy the audit's own fetch is about to replace, while the + review compares against the refreshed revision -- so a repository demotion + would keep merge-authority wording, and a base that is merely not fetched yet + would report as unavailable. These drive the real entry points with offline + fakes so the ordering, not just the resolver, is covered. + """ + + LANES = ( + "version: 1\n" + "lanes:\n" + " claude_audit:\n" + " type: review\n" + " driver: claude_cli\n" + " provider: claude\n" + " merge_authority: {authority}\n" + " informational: {informational}\n" + " labels:\n" + " needs: needs-claude-audit\n" + " done: claude-audit-done\n" + " blocked: claude-audit-blocked\n" + " codex:\n" + " type: audit\n" + " driver: local_cli\n" + " provider: codex\n" + " merge_authority: {authority}\n" + " informational: {informational}\n" + " labels:\n" + " needs: needs-codex-audit\n" + " done: codex-audit-done\n" + " blocked: codex-audit-blocked\n" + ) + AUTHORITATIVE = LANES.format(authority="true", informational="false") + DEMOTED = LANES.format(authority="false", informational="true") + + def _git(self, *args: str) -> str: + return subprocess.run( + ["git", *args], cwd=self.repo, check=True, capture_output=True, text=True + ).stdout.strip() + + def setUp(self): + import shutil + import tempfile + + self.tmp = Path(tempfile.mkdtemp()) + self.addCleanup(lambda: shutil.rmtree(self.tmp, ignore_errors=True)) + self.repo = self.tmp / "repo" + self.repo.mkdir() + self._git("init", "--initial-branch", "main") + self._git("config", "user.email", "lane@example.invalid") + self._git("config", "user.name", "Lane") + self._git("config", "commit.gpgsign", "false") + + def _commit(self, body: str, message: str) -> str: + (self.repo / "code-mower.yml").write_text(body, encoding="utf-8") + self._git("add", "code-mower.yml") + self._git("commit", "-m", message) + return self._git("rev-parse", "HEAD") + + def _stale_demotion(self) -> None: + """Leave `origin/main` on the authoritative policy the remote demoted.""" + self._commit(self.AUTHORITATIVE, "authoritative policy") + self._git("update-ref", "refs/remotes/origin/main", "main") + self._commit(self.DEMOTED, "demote the review lane") + # The PR head under review keeps declaring merge authority, so reading + # the checkout instead of the fetched base would also be wrong. + (self.repo / "code-mower.yml").write_text(self.AUTHORITATIVE, encoding="utf-8") + + def _fetch_effect(self): + """Advance `origin/main` the way the wrapper's real fetch would.""" + + def effect(*args, **kwargs): + self._git("update-ref", "refs/remotes/origin/main", "main") + return self._git("rev-parse", "main") + + return effect + + def _request(self, product: str, override=None): + return review_authority.AuthorityRequest(product=product, override=override) + + def _run_codex(self, config): + from code_mower import codex_audit_pr as cap + + worktree = self.tmp / "worktree" + worktree.mkdir(exist_ok=True) + head = "d" * 40 + pr_payload = {"head": {"sha": head, "ref": "human/fix"}, "title": "Fix"} + parsed = cap.CodexVerdict(verdict="PASS", prose="Summary:\n\nNone.") + diagnostics = cap.ReviewContextDiagnostics( + base_ref=config.base_ref, + head_sha=head, + changed_file_count=1, + diff_bytes=128, + requested_max_bytes=config.max_diff_bytes, + hard_limit_bytes=( + config.max_diff_hard_limit_bytes + or cap.DEFAULT_MAX_DIFF_HARD_LIMIT_BYTES + ), + included_diff_bytes=128, + effective_budget_usd=config.max_budget_usd or cap.DEFAULT_MAX_BUDGET_USD, + ) + with ( + mock.patch.dict( + "os.environ", + { + "PYTEST_CURRENT_TEST": "", + "CODE_MOWER_VERDICT_ARTIFACT_DIR": str(self.tmp / "verdicts"), + "GITHUB_RUN_ID": "", + }, + ), + mock.patch.object(cap, "fetch_pull_request", side_effect=[pr_payload] * 2), + mock.patch.object(cap, "preflight_codex_cli", return_value="codex-test"), + mock.patch.object(cap, "_discover_venv", return_value=None), + mock.patch.object(cap, "_fetch_pr_head"), + mock.patch.object(cap, "_fetch_base_ref", side_effect=self._fetch_effect()), + mock.patch.object( + cap, "_build_review_context_diagnostics", return_value=diagnostics + ), + mock.patch.object(cap, "_create_temp_worktree", return_value=worktree), + mock.patch.object(cap, "_remove_worktree"), + mock.patch.object(cap, "run_codex_review", return_value=("review", "")), + mock.patch.object( + cap, + "run_codex_verdict_structuring", + return_value=(parsed, '{"structured_output":"pass"}', ""), + ), + mock.patch.object(cap, "post_pr_comment", return_value={"html_url": "u"}), + ): + return cap.audit_pr(config, "owner/repo", 42) + + def _codex_config(self, **kwargs): + from code_mower import codex_audit_pr as cap + + return cap.AuditConfig( + "token", + {"owner/repo": self.repo}, + include_plan_context=False, + include_decision_context=False, + **{"merge_authority": False, **kwargs}, + ) + + def test_codex_drops_authority_when_the_fetch_brings_a_demotion(self): + # `merge_authority=True` stands in for a posture resolved against the + # stale local ref: the fetched base demoted the lane, so the comment the + # audit renders must not keep that wording. + self._stale_demotion() + result = self._run_codex( + self._codex_config( + merge_authority=True, authority_request=self._request("codex") + ) + ) + self.assertIn(review_authority.INFORMATIONAL_LABEL, result.comment_body) + self.assertNotIn(review_authority.MERGE_AUTHORITY_LABEL, result.comment_body) + + def test_codex_reports_a_base_that_only_the_fetch_made_available(self): + # Nothing named `origin/main` locally yet: resolving before the fetch + # would report the base unavailable for a repository that has a policy. + self._commit(self.AUTHORITATIVE, "authoritative policy") + result = self._run_codex(self._codex_config(authority_request=self._request("codex"))) + self.assertIn(review_authority.MERGE_AUTHORITY_LABEL, result.comment_body) + + def test_codex_without_a_request_keeps_the_authority_it_was_given(self): + # Direct callers and recorded fixtures decided authority themselves. + self._stale_demotion() + result = self._run_codex(self._codex_config(merge_authority=True)) + self.assertIn(review_authority.MERGE_AUTHORITY_LABEL, result.comment_body) + + def test_codex_positive_override_cannot_widen_the_fetched_demotion(self): + self._stale_demotion() + result = self._run_codex( + self._codex_config( + merge_authority=True, + authority_request=self._request("codex", override=True), + ) + ) + self.assertIn(review_authority.INFORMATIONAL_LABEL, result.comment_body) + + def _run_claude(self, config): + from code_mower import claude_audit_pr as cap + + head = "d" * 40 + pr_payload = {"head": {"sha": head, "ref": "human/fix"}, "title": "Fix"} + parsed = cap.ClaudeVerdict(verdict="PASS", prose="Summary:\n\nNone.") + advance = self._fetch_effect() + + def build_diff_context(*args, **kwargs): + # Stands in for the real builder, which fetches the base and pins the + # revision it diffed against onto the context it returns. + return cap.DiffContext( + "stat", "diff", ("src/app.py",), False, 1000, 1000, 40, 40, + fetched_base_ref=advance(), + ) + + with ( + mock.patch.dict( + "os.environ", + { + "PYTEST_CURRENT_TEST": "", + "CODE_MOWER_VERDICT_ARTIFACT_DIR": str(self.tmp / "verdicts"), + "GITHUB_RUN_ID": "", + }, + ), + mock.patch.object(cap, "fetch_pull_request", side_effect=[pr_payload] * 2), + mock.patch.object(cap, "_build_diff_context", side_effect=build_diff_context), + mock.patch.object(cap.code_mower_prompts, "load_review_prompt", return_value=""), + mock.patch.object( + cap, + "run_claude_audit", + return_value=(parsed, '{"structured_output":"pass"}', ""), + ), + mock.patch.object(cap, "post_pr_comment", return_value={"html_url": "u"}), + ): + return cap.audit_pr(config, "owner/repo", 42) + + def _claude_config(self, **kwargs): + from code_mower import claude_audit_pr as cap + + return cap.ClaudeAuditConfig( + "token", + {"owner/repo": self.repo}, + include_plan_context=False, + include_decision_context=False, + **{"merge_authority": False, **kwargs}, + ) + + def test_claude_drops_authority_when_the_diff_base_carries_a_demotion(self): + self._stale_demotion() + result = self._run_claude( + self._claude_config( + merge_authority=True, authority_request=self._request("claude") + ) + ) + self.assertIn(review_authority.INFORMATIONAL_LABEL, result.comment_body) + self.assertNotIn(review_authority.MERGE_AUTHORITY_LABEL, result.comment_body) + + def test_claude_reports_a_base_that_only_the_fetch_made_available(self): + self._commit(self.AUTHORITATIVE, "authoritative policy") + result = self._run_claude( + self._claude_config(authority_request=self._request("claude")) + ) + self.assertIn(review_authority.MERGE_AUTHORITY_LABEL, result.comment_body) + + def test_claude_without_a_request_keeps_the_authority_it_was_given(self): + self._stale_demotion() + result = self._run_claude(self._claude_config(merge_authority=True)) + self.assertIn(review_authority.MERGE_AUTHORITY_LABEL, result.comment_body) + + def test_the_posture_and_the_review_consume_the_same_fetched_revision(self): + # The rendered posture must match the policy at the exact revision the + # diff was taken from, not the checkout and not the pre-fetch ref. + self._stale_demotion() + fetched = self._git("rev-parse", "main") + at_fetched = review_authority.effective_merge_authority( + "claude", repo_root=self.repo, base_ref=fetched + ) + self.assertFalse(at_fetched["merge_authority"]) + at_head = review_authority.effective_merge_authority( + "claude", config_path=self.repo / "code-mower.yml" + ) + self.assertTrue(at_head["merge_authority"]) + result = self._run_claude( + self._claude_config( + merge_authority=True, authority_request=self._request("claude") + ) + ) + self.assertIn(at_fetched["label"], result.comment_body) + self.assertNotIn(at_head["label"], result.comment_body) + + def test_a_historical_diff_context_records_no_fetched_revision(self): + from code_mower import claude_audit_pr as cap + + context = cap.DiffContext("stat", "diff", (), False, 1, 1, 1, 1) + self.assertEqual(context.fetched_base_ref, "") + self.assertEqual(tuple(context), ("stat", "diff", False)) + + + # Every consumer below the fetch reads the revision the audit fetched. + # + # Resolving the posture against the fetched revision is not enough on its + # own: `base_ref` stays a mutable name, so the trusted-ref lookups, the + # review context and the review itself could still resolve it again later + # and read a different commit. The tests below advance `origin/main` *after* + # the fetch -- the way an upstream merge landing mid-review would -- and + # prove the rendered posture and the downstream review and context all stay + # on the one fetched snapshot. + + def _advance_the_tracking_ref(self) -> str: + """Land a re-promotion on `origin/main` after the audit fetched it.""" + + self._commit(self.AUTHORITATIVE, "re-promote the review lane upstream") + self._git("update-ref", "refs/remotes/origin/main", "main") + return self._git("rev-parse", "main") + + def test_codex_review_and_context_stay_on_the_fetched_revision(self): + from code_mower import codex_audit_pr as cap + + self._stale_demotion() + observed: dict[str, str] = {} + worktree = self.tmp / "worktree" + worktree.mkdir(exist_ok=True) + head = "d" * 40 + pr_payload = {"head": {"sha": head, "ref": "human/fix"}, "title": "Fix"} + parsed = cap.CodexVerdict(verdict="PASS", prose="Summary:\n\nNone.") + + def prepare(**kwargs): + observed["context"] = kwargs["base_ref"] + # Upstream moves on while this review runs. Anything that resolves + # the name again from here reads the re-promotion, not the fetch. + observed["moved_to"] = self._advance_the_tracking_ref() + return None + + def diagnostics(local_repo, **kwargs): + observed["diagnostics"] = kwargs["base_ref"] + return cap.ReviewContextDiagnostics( + base_ref=kwargs["base_ref"], + head_sha=head, + changed_file_count=1, + diff_bytes=128, + included_diff_bytes=128, + ) + + def review(config, *args, **kwargs): + observed["review"] = config.base_ref + return ("review", "") + + config = self._codex_config( + merge_authority=True, authority_request=self._request("codex") + ) + with ( + mock.patch.dict( + "os.environ", + { + "PYTEST_CURRENT_TEST": "", + "CODE_MOWER_VERDICT_ARTIFACT_DIR": str(self.tmp / "verdicts"), + "GITHUB_RUN_ID": "", + }, + ), + mock.patch.object(cap, "fetch_pull_request", side_effect=[pr_payload] * 2), + mock.patch.object(cap, "preflight_codex_cli", return_value="codex-test"), + mock.patch.object(cap, "_discover_venv", return_value=None), + mock.patch.object(cap, "_fetch_pr_head"), + mock.patch.object(cap, "_fetch_base_ref", side_effect=self._fetch_effect()), + mock.patch.object(cap.context_audit, "prepare", side_effect=prepare), + mock.patch.object( + cap, "_build_review_context_diagnostics", side_effect=diagnostics + ), + mock.patch.object(cap, "_create_temp_worktree", return_value=worktree), + mock.patch.object(cap, "_remove_worktree"), + mock.patch.object(cap, "run_codex_review", side_effect=review), + mock.patch.object( + cap, + "run_codex_verdict_structuring", + return_value=(parsed, '{"structured_output":"pass"}', ""), + ), + mock.patch.object(cap, "post_pr_comment", return_value={"html_url": "u"}), + ): + result = cap.audit_pr(config, "owner/repo", 42) + + fetched = self._git("rev-parse", "refs/remotes/origin/main~1") + self.assertNotEqual(observed["moved_to"], fetched) + for consumer in ("context", "diagnostics", "review"): + self.assertEqual(observed[consumer], fetched, consumer) + # The named ref would have read the re-promotion instead. + self.assertNotIn("origin/main", observed.values()) + self.assertIn(review_authority.INFORMATIONAL_LABEL, result.comment_body) + + def test_claude_review_and_context_stay_on_the_fetched_revision(self): + from code_mower import claude_audit_pr as cap + + self._stale_demotion() + observed: dict[str, str] = {} + head = "d" * 40 + pr_payload = {"head": {"sha": head, "ref": "human/fix"}, "title": "Fix"} + parsed = cap.ClaudeVerdict(verdict="PASS", prose="Summary:\n\nNone.") + advance = self._fetch_effect() + + def build_diff_context(*args, **kwargs): + return cap.DiffContext( + "stat", "diff", ("src/app.py",), False, 1000, 1000, 40, 40, + fetched_base_ref=advance(), + ) + + def prepare(**kwargs): + observed["context"] = kwargs["base_ref"] + observed["moved_to"] = self._advance_the_tracking_ref() + return None + + def load_review_prompt(*args, **kwargs): + observed["doctrine"] = kwargs["trusted_git_ref"] + return "" + + def audit(config, prompt): + observed["review"] = config.base_ref + observed["prompt_names"] = config.base_ref in prompt + return (parsed, '{"structured_output":"pass"}', "") + + config = self._claude_config( + merge_authority=True, authority_request=self._request("claude") + ) + with ( + mock.patch.dict( + "os.environ", + { + "PYTEST_CURRENT_TEST": "", + "CODE_MOWER_VERDICT_ARTIFACT_DIR": str(self.tmp / "verdicts"), + "GITHUB_RUN_ID": "", + }, + ), + mock.patch.object(cap, "fetch_pull_request", side_effect=[pr_payload] * 2), + mock.patch.object(cap, "_build_diff_context", side_effect=build_diff_context), + mock.patch.object(cap.context_audit, "prepare", side_effect=prepare), + mock.patch.object( + cap.code_mower_prompts, + "load_review_prompt", + side_effect=load_review_prompt, + ), + mock.patch.object(cap, "run_claude_audit", side_effect=audit), + mock.patch.object(cap, "post_pr_comment", return_value={"html_url": "u"}), + ): + result = cap.audit_pr(config, "owner/repo", 42) + + fetched = self._git("rev-parse", "refs/remotes/origin/main~1") + self.assertNotEqual(observed["moved_to"], fetched) + for consumer in ("context", "doctrine", "review"): + self.assertEqual(observed[consumer], fetched, consumer) + self.assertTrue(observed["prompt_names"]) + self.assertIn(review_authority.INFORMATIONAL_LABEL, result.comment_body) + + def test_a_force_push_race_still_renders_the_fetched_base(self): + from code_mower import claude_audit_pr as cap + + self._stale_demotion() + fetched = self._git("rev-parse", "main") + head = "d" * 40 + pr_payload = {"head": {"sha": head, "ref": "human/fix"}, "title": "Fix"} + advance = self._fetch_effect() + + def build_diff_context(*args, **kwargs): + # The base is fetched before the head mismatch is detected, so the + # stale notice knows which revision the audit had already taken. + raise cap._FetchedHeadMismatchWithBase(head, "e" * 40, advance()) + + config = self._claude_config( + merge_authority=True, authority_request=self._request("claude") + ) + with ( + mock.patch.dict( + "os.environ", + { + "PYTEST_CURRENT_TEST": "", + "CODE_MOWER_VERDICT_ARTIFACT_DIR": str(self.tmp / "verdicts"), + "GITHUB_RUN_ID": "", + }, + ), + mock.patch.object(cap, "fetch_pull_request", side_effect=[pr_payload] * 2), + mock.patch.object(cap, "_build_diff_context", side_effect=build_diff_context), + mock.patch.object(cap, "post_pr_comment", return_value={"html_url": "u"}), + ): + result = cap.audit_pr(config, "owner/repo", 42) + + self.assertEqual(result.verdict, "STALE") + self.assertIn(review_authority.INFORMATIONAL_LABEL, result.comment_body) + self.assertNotIn(review_authority.MERGE_AUTHORITY_LABEL, result.comment_body) + self.assertTrue(fetched) + + def test_the_mismatch_stays_catchable_as_the_shared_exception(self): + from code_mower import claude_audit_pr as cap + from code_mower.provider_runners import FetchedHeadMismatch + + error = cap._FetchedHeadMismatchWithBase("a" * 40, "b" * 40, "c" * 40) + self.assertIsInstance(error, FetchedHeadMismatch) + self.assertEqual(error.expected_sha, "a" * 40) + self.assertEqual(error.actual_sha, "b" * 40) + self.assertEqual(error.fetched_base_ref, "c" * 40) + + +class PortableStarterCommandTests(unittest.TestCase): + """The packaged starter has no repository path a rendered command can pin.""" + + STARTER = devin_readiness.PACKAGED_STARTER_SOURCE + # Stands in for the installation-specific path the starter resolves to at + # runtime; the literal prefix is assembled the way privacy_scan.py writes its + # own patterns. + INSTALLED = "/" + "opt/venv/lib/code_mower/templates/code-mower.example.yml" + + def test_starter_doctor_command_uses_the_supported_selector(self): + command = devin_readiness.doctor_command( + config_path=self.INSTALLED, + profile="recommended", + config_source=self.STARTER, + devin=True, + ) + self.assertEqual( + command, + "`code-mower doctor --packaged-starter --profile recommended --devin`", + ) + self.assertNotIn(self.INSTALLED, command) + + def test_starter_transport_selection_is_portable_and_still_staged(self): + steps = devin_readiness.select_transport_command( + "devin_api_v3", + config_path=self.INSTALLED, + profile="recommended", + config_source=self.STARTER, + ) + self.assertNotIn(self.INSTALLED, steps) + self.assertIn( + "code-mower init --packaged-starter --profile recommended " + "--set-transport devin=devin_api_v3 --dry-run", + steps, + ) + self.assertIn("--apply --output-dir", steps) + + def test_starter_verification_inspects_the_installed_configuration(self): + # Preview and staging read the package resource, but an install writes the + # repository's own configuration and leaves the starter unchanged, so the + # final check must select the installed file at the same profile. + steps = devin_readiness.select_transport_command( + "devin_api_v3", + config_path=self.INSTALLED, + profile="advanced", + config_source=self.STARTER, + ) + preview, _, verification = steps.partition("install them through") + self.assertIn( + "code-mower init --packaged-starter --profile advanced " + "--set-transport devin=devin_api_v3 --dry-run", + preview, + ) + self.assertIn("--packaged-starter --profile advanced --set-transport", preview) + self.assertIn( + "`code-mower doctor code-mower.yml --profile advanced --devin`", + verification, + ) + self.assertNotIn("--packaged-starter", verification) + self.assertNotIn(self.INSTALLED, steps) + + def test_repository_verification_keeps_the_configuration_it_installs_over(self): + # A repository finding installs over its own file, so nothing redirects. + steps = devin_readiness.select_transport_command( + "devin_api_v3", config_path="ops/mower.yml", profile="recommended" + ) + _, _, verification = steps.partition("install them through") + self.assertIn( + "`code-mower doctor ops/mower.yml --profile recommended --devin`", + verification, + ) + self.assertNotIn("code-mower.yml", verification) + + def test_the_installed_configuration_path_matches_what_init_writes(self): + self.assertEqual( + devin_readiness.INSTALLED_CONFIG_PATH, code_mower_init.ADOPTION_CONFIG_PATH + ) + + def test_repository_configuration_is_never_replaced_by_the_starter(self): + repository = "code-mower.yml" + steps = devin_readiness.select_transport_command( + "devin_api_v3", config_path=repository, profile="recommended" + ) + self.assertIn(repository, steps) + self.assertNotIn("--packaged-starter", steps) + self.assertNotIn("--easy", steps) + + def test_a_non_recommended_starter_profile_keeps_its_selected_profile(self): + # The selector names the package resource and chooses no profile, so a + # starter finding under any profile stays pinned to the one it describes. + command = devin_readiness.doctor_command( + config_path=self.INSTALLED, profile="advanced", config_source=self.STARTER + ) + self.assertIn("--packaged-starter", command) + self.assertIn("--profile advanced", command) + self.assertNotIn(self.INSTALLED, command) + self.assertNotIn("--easy", command) + + def test_a_starter_profile_containing_spaces_stays_quoted(self): + command = devin_readiness.doctor_command( + config_path=self.INSTALLED, profile="my profile", config_source=self.STARTER + ) + self.assertIn("--packaged-starter --profile 'my profile'", command) + + def test_paths_and_profiles_containing_spaces_stay_quoted(self): + spaced = "/" + "srv/Code Mower/code-mower.yml" + command = devin_readiness.doctor_command( + config_path=spaced, profile="my profile", devin=True + ) + self.assertIn("'/" + "srv/Code Mower/code-mower.yml'", command) + self.assertIn("--profile 'my profile'", command) + + def test_custom_lane_guidance_names_the_starter_without_a_path(self): + guidance = devin_readiness.custom_lane_guidance( + "devin_api_v3", + config_path=self.INSTALLED, + profile="recommended", + config_source=self.STARTER, + lanes=("house_devin",), + ) + self.assertNotIn(self.INSTALLED, guidance) + self.assertIn("packaged starter configuration (--packaged-starter)", guidance) + self.assertIn("`house_devin`", guidance) + + def test_readiness_findings_carry_the_starter_source_into_remediation(self): + findings = devin_readiness.devin_readiness( + None, + transport="devin_api_v3", + config_profile="recommended", + config_path=self.INSTALLED, + config_source=self.STARTER, + env={}, + ) + rendered = "\n".join( + f"{finding.remediation}\n{json.dumps(finding.detail, default=str)}" + for finding in findings + ) + self.assertNotIn(self.INSTALLED, rendered) + self.assertIn("--packaged-starter", rendered) + + +class PackagedStarterSelectorTests(unittest.TestCase): + """`--packaged-starter` selects the maintained resource, not a cwd-local file. + + The rendered command is only portable if the selector it names resolves the + same configuration from any directory, so these assert the CLI-level + selection against decoy files rather than comparing command strings. + """ + + def _decoy_dir(self) -> Path: + import tempfile + + root = Path(tempfile.mkdtemp()) + self.addCleanup(lambda: __import__("shutil").rmtree(root, ignore_errors=True)) + # Both cwd-local files that `--easy` would prefer: doctor picks up + # code-mower.yml, init picks up code-mower.example.yml. + for name in ("code-mower.yml", "code-mower.example.yml"): + (root / name).write_text("lanes: {}\n", encoding="utf-8") + return root + + def test_the_packaged_starter_resolver_ignores_cwd_local_decoys(self): + from code_mower import package as code_mower_package + + root = self._decoy_dir() + cwd = Path.cwd() + os.chdir(root) + self.addCleanup(os.chdir, cwd) + resolved = code_mower_package.packaged_starter_config_path() + self.assertTrue(resolved.is_file()) + self.assertEqual(resolved.name, "code-mower.example.yml") + for decoy in ("code-mower.yml", "code-mower.example.yml"): + self.assertNotEqual(resolved.resolve(), (root / decoy).resolve()) + + def test_doctor_and_init_select_the_same_config_under_decoys(self): + from code_mower import doctor as code_mower_doctor + from code_mower import init as code_mower_init + from code_mower import package as code_mower_package + + root = self._decoy_dir() + cwd = Path.cwd() + os.chdir(root) + self.addCleanup(os.chdir, cwd) + expected = code_mower_package.packaged_starter_config_path().resolve() + + selected: list[Path] = [] + with mock.patch.object( + code_mower_doctor, "run_doctor", side_effect=RuntimeError("stop") + ) as run_doctor: + with self.assertRaises(RuntimeError): + code_mower_doctor.main(["--packaged-starter", "--profile", "advanced"]) + selected.append(Path(run_doctor.call_args.kwargs["config_path"]).resolve()) + self.assertEqual(run_doctor.call_args.kwargs["config_source"], "packaged_starter") + # The selector chooses the resource, never the profile. + self.assertEqual(run_doctor.call_args.kwargs["profile"], "advanced") + + with mock.patch.object( + code_mower_init, "load_config", side_effect=RuntimeError("stop") + ) as load_config: + with self.assertRaises(RuntimeError): + code_mower_init.main( + ["--packaged-starter", "--profile", "advanced", "--dry-run"] + ) + selected.append(Path(load_config.call_args.args[0]).resolve()) + + self.assertEqual(selected, [expected, expected]) + + def test_a_contradictory_explicit_config_is_rejected_not_ignored(self): + from code_mower import doctor as code_mower_doctor + from code_mower import init as code_mower_init + + root = self._decoy_dir() + cwd = Path.cwd() + os.chdir(root) + self.addCleanup(os.chdir, cwd) + selections = ( + (code_mower_doctor.main, ["code-mower.yml", "--packaged-starter"]), + ( + code_mower_init.main, + ["code-mower.yml", "--packaged-starter", "--dry-run"], + ), + ) + for main, argv in selections: + with contextlib.redirect_stderr(io.StringIO()) as err: + status = main(argv) + self.assertEqual(status, 1) + self.assertIn("--packaged-starter", err.getvalue()) + + def test_the_cli_routes_the_selector_without_injecting_a_default_config(self): + from code_mower import cli as code_mower_cli + + with mock.patch.object( + code_mower_cli.code_mower_init, "main", return_value=0 + ) as init_main: + code_mower_cli._init_main(["--packaged-starter", "--dry-run"]) + self.assertEqual(init_main.call_args.args[0], ["--packaged-starter", "--dry-run"]) + + +class SupersededDevinBridgeTests(unittest.TestCase): + def _repo(self, *paths: str) -> Path: + import tempfile + + root = Path(tempfile.mkdtemp()) + self.addCleanup(lambda: __import__("shutil").rmtree(root, ignore_errors=True)) + for path in paths: + target = root / path + target.parent.mkdir(parents=True, exist_ok=True) + target.write_text("# legacy\n", encoding="utf-8") + return root + + def test_no_devin_repository_reports_nothing(self): + summary = migration._superseded_devin_bridge_summary(self._repo(), files=[]) + self.assertEqual(summary["status"], "skip") + self.assertEqual(summary["reason"], "no_superseded_bridge_files") + self.assertEqual(summary["paths"], []) + + def test_bridge_and_labeler_pair_reports_bounded_migration(self): + root = self._repo( + ".github/workflows/devin-audit-bridge.yml", + ".github/workflows/devin-audit-labeler.yml", + ) + files = [ + {"path": ".github/workflows/devin-audit-bridge.yml", "tracked": True}, + {"path": ".github/workflows/devin-audit-labeler.yml", "tracked": True}, + ] + summary = migration._superseded_devin_bridge_summary(root, files=files) + self.assertEqual(summary["status"], "warn") + self.assertEqual(summary["reason"], "superseded_bridge_pair") + self.assertEqual(summary["transport"], "devin_api_v3") + self.assertEqual( + summary["paths"], + [ + ".github/workflows/devin-audit-bridge.yml", + ".github/workflows/devin-audit-labeler.yml", + ], + ) + action = summary["next_action"] + self.assertIn("superseded", action) + self.assertIn("devin_api_v3", action) + self.assertIn("--set-transport", action) + self.assertIn("--dry-run", action) + self.assertIn("never deletes or rewrites", action) + # Bounded: only the observed files are named. + self.assertNotIn("tools/devin_audit_bridge.py", action) + + def test_single_legacy_file_is_still_reported(self): + root = self._repo("tools/devin_audit_bridge.py") + summary = migration._superseded_devin_bridge_summary(root, files=[]) + self.assertEqual(summary["status"], "warn") + self.assertEqual(summary["reason"], "superseded_bridge_files") + self.assertEqual(summary["paths"], ["tools/devin_audit_bridge.py"]) + + def test_detection_never_removes_the_files(self): + root = self._repo(".github/workflows/devin-audit-bridge.yml") + migration._superseded_devin_bridge_summary(root, files=[]) + self.assertTrue((root / ".github/workflows/devin-audit-bridge.yml").is_file()) + + def test_legacy_paths_are_setup_drift_candidates(self): + for path in migration.SUPERSEDED_DEVIN_BRIDGE_PATHS: + self.assertTrue(migration._is_setup_candidate_path(path), path) + + def test_reported_option_matches_the_supported_selection_flag(self): + self.assertEqual( + migration.DEVIN_TRANSPORT_OPTION, devin_readiness.TRANSPORT_OPTION + ) + + def test_next_action_includes_the_superseded_migration(self): + superseded = {"status": "warn", "next_action": "migrate the superseded bridge"} + action = migration._setup_drift_next_action( + changed_count=0, + standalone_pin={"status": "skip"}, + builder_hint={"status": "skip"}, + repo_path_hint={"status": "pass"}, + superseded_bridge=superseded, + ) + self.assertEqual(action, "migrate the superseded bridge") + + def test_text_rendering_surfaces_the_superseded_transport(self): + payload = { + "status": "warn", + "repo_path": "/tmp/repo", + "profile": "recommended", + "counts": {}, + "next_action": "migrate", + "superseded_bridge": { + "status": "warn", + "reason": "superseded_bridge_pair", + "transport": "devin_api_v3", + "paths": [".github/workflows/devin-audit-bridge.yml"], + "next_action": "preview the transport selection", + }, + } + text = migration.render_setup_drift_text(payload) + self.assertIn("Superseded transport: WARN superseded_bridge_pair", text) + self.assertIn("superseded_by=devin_api_v3", text) + self.assertIn("Superseded transport next: preview the transport selection", text) + + def test_text_rendering_omits_the_section_when_absent(self): + payload = { + "status": "pass", + "repo_path": "/tmp/repo", + "profile": "recommended", + "counts": {}, + "next_action": "ok", + "superseded_bridge": {"status": "skip", "reason": "no_superseded_bridge_files"}, + } + self.assertNotIn("Superseded transport", migration.render_setup_drift_text(payload)) + + +def _report(checks): + return DoctorReport( + config_path="code-mower.yml", + provider_templates_path="providers.yml", + profile="recommended", + checks=tuple(checks), + ) + + +class ConciseDoctorViewTests(unittest.TestCase): + def setUp(self): + self.report = _report( + [ + DoctorCheck( + name="doctor.adoption.posture_hint", + status="warn", + message="hosted-builders posture", + remediation="ignore local CLI warnings", + ), + DoctorCheck(name="github.token", status="fail", message="token missing"), + DoctorCheck( + name="provider.devin.optional", status="warn", message="devin not selected" + ), + DoctorCheck( + name="provider.graphify.optional", status="warn", message="graphify absent" + ), + DoctorCheck(name="config.lanes", status="pass", message="ok"), + ] + ) + + def test_summary_leads_with_failures_and_keeps_counts(self): + text = render_doctor_summary(self.report) + self.assertIn("Code Mower doctor (concise)", text) + self.assertIn("Adoption posture: WARN doctor.adoption.posture_hint", text) + self.assertIn("Active failures and owner actions", text) + self.assertIn("FAIL github.token", text) + self.assertIn("Remaining detail by group", text) + self.assertIn("--json", text) + # Optional-provider warning detail is counted, not listed line by line. + self.assertNotIn("devin not selected", text) + + def test_full_text_view_keeps_every_check(self): + text = render_doctor_text(self.report) + self.assertIn("devin not selected", text) + self.assertIn("graphify absent", text) + + def test_summary_reports_a_clean_run(self): + text = render_doctor_summary(_report([DoctorCheck(name="config.lanes", status="pass", message="ok")])) + self.assertIn("No active failures or owner actions.", text) + + def test_summary_handles_an_empty_report(self): + self.assertIn("No checks ran.", render_doctor_summary(_report([]))) + + def test_json_detail_is_unchanged_by_the_concise_flag(self): + payload = self.report.as_dict() + self.assertEqual(len(payload["checks"]), 5) + + def test_summary_keeps_local_detail_out_of_nothing_it_did_not_receive(self): + # Privacy: the summary renders only fields the report already carried. + # The home prefix is assembled the way scripts/privacy_scan.py writes its + # own patterns, so the assertion does not become a tracked literal. + home_prefix = "/" + "Users/" + text = render_doctor_summary(self.report) + for line in text.splitlines(): + self.assertNotIn(home_prefix, line) + + +class ConciseDoctorCliTests(unittest.TestCase): + def _run(self, *args): + return subprocess.run( + [sys.executable, "-m", "code_mower.doctor", *args], + cwd=Path(__file__).resolve().parents[1], + env={ + "PATH": "/usr/bin:/bin", + "PYTHONPATH": str(Path(__file__).resolve().parents[1] / "src"), + "HOME": str(Path.home()), + }, + capture_output=True, + text=True, + ) + + def test_concise_and_advanced_are_mutually_exclusive(self): + result = self._run("--concise", "--advanced") + self.assertEqual(result.returncode, 2) + self.assertIn("not allowed with argument", result.stderr) + + def test_json_output_is_json_even_with_concise(self): + result = self._run("src/code_mower/templates/code-mower.example.yml", "--concise", "--json") + self.assertIn(result.returncode, (0, 1)) + json.loads(result.stdout) + + +class PromptPackDevinGuidanceTests(unittest.TestCase): + def setUp(self): + self.text = ( + Path(__file__).resolve().parents[1] / "docs" / "orchestrator-prompt-pack.md" + ).read_text(encoding="utf-8") + + def test_optional_devin_section_is_opt_in_and_uses_supported_commands(self): + self.assertIn("## Optional Devin Setup Prompt", self.text) + self.assertIn("The default adoption is", self.text) + self.assertIn("--set-transport devin=devin_api_v3 --dry-run", self.text) + self.assertIn("code-mower doctor CONFIG --profile PROFILE", self.text) + self.assertIn(".code-mower.generated", self.text) + + def test_guidance_keeps_staging_and_authority_boundaries(self): + self.assertIn("selecting a transport grants no review or", self.text) + self.assertIn("Do not delete or rewrite repository-owned workflow files", self.text) + self.assertIn("do not start paid sessions", self.text) + + def test_role_and_lease_guidance_is_referenced_not_restated(self): + self.assertIn("docs/participant-qualification.md", self.text) + + def test_packaged_starter_posture_names_the_portable_selector(self): + self.assertIn( + "code-mower doctor --packaged-starter --profile PROFILE --devin", self.text + ) + self.assertIn( + "Never substitute the starter for a repository configuration", self.text + ) + + def test_the_prompt_pack_does_not_call_easy_a_packaged_starter_selector(self): + # `--easy` resolves against cwd-local files, so the pack must not offer it + # as the way to name the maintained package resource. + self.assertNotIn("code-mower doctor --easy --devin", self.text) + self.assertIn("--easy does not", self.text) + + +if __name__ == "__main__": + unittest.main() diff --git a/tests/test_audit_comment_posture.py b/tests/test_audit_comment_posture.py index 560f27ef..5bbc0bf2 100644 --- a/tests/test_audit_comment_posture.py +++ b/tests/test_audit_comment_posture.py @@ -94,8 +94,13 @@ def test_cli_posture_defaults_can_be_overridden(self) -> None: self.assertTrue(parse_codex_args(["--merge-authority"]).merge_authority) with patch.dict("os.environ", {}, clear=True): - self.assertTrue(parse_claude_args([]).merge_authority) + # Unset is not a posture claim: the wrapper resolves the effective + # posture from the repository configuration this run uses, and only + # an explicit flag or env override states one here. + self.assertIsNone(parse_claude_args([]).merge_authority) + self.assertIsNone(parse_codex_args([]).merge_authority) self.assertFalse(parse_claude_args(["--informational"]).merge_authority) + self.assertTrue(parse_claude_args(["--merge-authority"]).merge_authority) if __name__ == "__main__":