From fb22be35d74583be4d4332e5f5c87d5836175c47 Mon Sep 17 00:00:00 2001 From: Jeff Huber Date: Mon, 14 Sep 2026 22:55:38 -0700 Subject: [PATCH] Establish the pure builder lineage contract (stage 1 of #963) A pull request can be built by more than one Code Mower builder lane. The opener, the branch prefix and the single active `builder:*` label each describe at most one of those lanes, so consumers that each composed those signals slightly differently stopped agreeing about who wrote the current diff -- and a lane was admitted to review its own work. This stage ships only the shared decision, as a pure contract. Nothing imports it yet; producers, stores, publication, reconciliation and consumer activation are later stages, so accepted main behaviour is unchanged. * `code_mower.builder_lineage` -- immutable contribution episodes bound to repository, pull request, branch, source lane, destination lane, expected head and resulting head; ordered handoff and same-writer continuation; one current writer and the full contributor set; exact-head resolution that waits rather than guessing; strict announced-marker grammar; raw comment, slurped-page and selected-history validation before any normalization. * `code_mower.lineage_identity` -- the canonical own-reviewer label/account floor, the configured-authority marker trust predicate, and the bounded published-episode and admission contracts later stages consume. * `tools/builder_lineage.py` -- the vendored copy a generated product gate imports, plus its materialization and package-manifest entries. The API is organised around five invariants rather than per-call checks: 1. Exact-target resolution and reviewer admission require the complete repository, positive non-boolean bounded pull request number, nonempty valid branch and full head, through one `require_exact_target` contract. Incomplete target data is a `target_invalid` conflict, never a quiet downgrade to identity-only -- that downgrade skipped branch binding, so evidence from another branch resolved as evidence about this one. Identity-only is a route callers select on purpose (`resolve_configured_identity`, which takes no evidence argument), and a context that is partially populated or carries episodes is never absent. 2. `canonical_identity` normalizes account aliases and branch prefixes once; lane naming, branch lookup, the complete resolver, the carried context and the reviewer floor all consume that one representation. Compatible spellings collapse and contradictory ones refuse regardless of insertion order, and longest-prefix matching now looks values up on the canonical key, so trimming a key no longer strands its lookup. 3. `bounded_arrivals` enforces the documented 560 raw-arrival budget (528 cumulative public entries plus 32 private) while the input is walked, so an oversized source is never materialised and a lazy one is not consumed past the first refused arrival. Collectors and the composer preserve validated raw arrivals uncollapsed; only the owning resolver deduplicates, where it can also refuse a contradiction. Malformed members are the documented contract error, not an incidental attribute failure. 4. `lineage_comment_marker` validates the whole chain through the resolver's own walk and refuses zero, malformed, conflicting or unchained input. Slicing to the bound turned an over-long chain into a successful publication that had quietly lost its newest episodes. 5. The owning packaging test materializes a real product tree through `init` and runs a clean subprocess that can only see the generated `tools/` directory, asserting module origins and exercising marker parsing, exact resolution and an unchanged compatibility helper there. tools/decisions.py is untouched; its environment-reading parity is stage 3. Refs #963 Closes #990 Builder-Provider: claude Builder-Executor: claude_cli --- CHANGELOG.md | 22 + code-mower-package-manifest.json | 10 + src/code_mower/builder_lineage.py | 1502 ++++++++++++++++++++++ src/code_mower/init.py | 10 + src/code_mower/lineage_identity.py | 408 ++++++ src/code_mower/package_manifest.py | 5 + tests/lineage_contract_fixtures.py | 226 ++++ tests/test_builder_lineage_contract.py | 949 ++++++++++++++ tests/test_builder_lineage_transport.py | 600 +++++++++ tests/test_lineage_contract_packaging.py | 513 ++++++++ tests/test_lineage_identity_contract.py | 444 +++++++ tools/builder_lineage.py | 1502 ++++++++++++++++++++++ 12 files changed, 6191 insertions(+) create mode 100644 src/code_mower/builder_lineage.py create mode 100644 src/code_mower/lineage_identity.py create mode 100644 tests/lineage_contract_fixtures.py create mode 100644 tests/test_builder_lineage_contract.py create mode 100644 tests/test_builder_lineage_transport.py create mode 100644 tests/test_lineage_contract_packaging.py create mode 100644 tests/test_lineage_identity_contract.py create mode 100644 tools/builder_lineage.py diff --git a/CHANGELOG.md b/CHANGELOG.md index 5d62c1e2..2f37aaa7 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -7,6 +7,28 @@ later entries are regular releases. ## Unreleased +### Added + +- A pure builder contribution lineage contract (`code_mower.builder_lineage`, + `code_mower.lineage_identity`, plus the vendored `tools/builder_lineage.py` + the generated gate helper imports). It resolves who wrote a pull request at an + exact head from ordered, immutably bound contribution episodes: one current + writer, the full contributor set, and a concise owner action whenever the + evidence is stale, unbound, duplicated or contradictory. The same decision + also covers identity, configured branch provenance and published evidence + together, including when there are no episodes at all, so a configured + branch/label disagreement is unresolved rather than an ordinary + single-builder pull request. Exact-target resolution and reviewer admission + require the complete repository, pull request number, branch and head: + incomplete target data is a refusal, never a quiet downgrade to the + identity-only answer, which is a route callers select on purpose. Account + aliases and configured branch prefixes are reconciled once, so compatible + spellings collapse and contradictory ones refuse whatever order they were + written in. Announced-but-broken lineage markers, successful-but-unreadable + comment reads and chains that could not resolve all fail closed instead of + being reported as an absent history or published with an episode missing. No + consumer behaviour changes in this release: nothing imports the contract yet. + ### Fixed - Coworker fast-search responses with `source_id`, missing titles, or `Text` diff --git a/code-mower-package-manifest.json b/code-mower-package-manifest.json index 9ab45d8d..3d1ef286 100644 --- a/code-mower-package-manifest.json +++ b/code-mower-package-manifest.json @@ -372,6 +372,11 @@ "source": "src/code_mower/builder_experiment.py", "target": "src/code_mower/builder_experiment.py" }, + { + "kind": "core", + "source": "tools/builder_lineage.py", + "target": "src/code_mower/builder_lineage.py" + }, { "kind": "core", "source": "src/code_mower/builder_runs.py", @@ -1207,6 +1212,11 @@ "source": "src/code_mower/lane_status.py", "target": "src/code_mower/lane_status.py" }, + { + "kind": "core", + "source": "src/code_mower/lineage_identity.py", + "target": "src/code_mower/lineage_identity.py" + }, { "kind": "core", "source": "src/code_mower/local_cli_commands.py", diff --git a/src/code_mower/builder_lineage.py b/src/code_mower/builder_lineage.py new file mode 100644 index 00000000..9f0bf594 --- /dev/null +++ b/src/code_mower/builder_lineage.py @@ -0,0 +1,1502 @@ +"""Exact-head builder contribution lineage: the pure contract. + +A pull request can be built by more than one Code Mower builder lane. The +opener, the branch prefix and the single active ``builder:*`` label each +describe at most one of those lanes, so any of them alone will misdescribe a +pull request that changed hands. This module keeps the ordered contribution +history instead, and derives the one current writer from it. + +Trust rules this module exists to enforce: + +* A contribution episode is evidence produced by a verified handoff and + delivery boundary that observed the source writer going quiescent and + observed both heads. A caller-supplied boolean, a pull request body marker, a + commit trailer, the opener or the most recent label are none of them able to + attest that a takeover happened. +* Episodes are bound to repository, pull request, branch, source lane, + destination lane, expected head and resulting head. An episode that does not + bind to the pull request under decision is not evidence about it. +* Resolution is exact-head. Lineage that stops short of the current head is + *waiting*, never a guess about who wrote the current diff. +* Conflicting, duplicated, unchained or unbound evidence fails closed with one + concise owner action rather than picking a winner. + +Purity is part of the contract, not an implementation detail. Nothing here +reads the environment, touches a store, opens a socket or imports an adapter: +every decision is a function of its explicit arguments. That is what lets the +same answer be computed by the package, by the vendored ``tools/`` copy inside +a generated product repository, and by a reviewer host that has no private +state at all. Recording, publication, label application and consumer +activation are deliberately somewhere else. + +Everything here is metadata-only: lane names, a repository slug, a pull request +number, a branch name and commit shas. +""" + +from __future__ import annotations + +import hashlib +import json +import re +from dataclasses import dataclass, field +from typing import Any, Callable, Iterable, Mapping, Sequence + + +SCHEMA = "code_mower.builderLineage.v1" +EPISODE_SCHEMA = "code_mower.contributionEpisode.v1" +RECORD_SCHEMA = "code_mower.builderLineageRecord.v1" + +#: Hidden marker used to publish bounded lineage metadata on a pull request. +#: Only comments from an already trusted author are parsed; the marker is a +#: transport, never an authorization. +LINEAGE_MARKER = "CODE_MOWER_BUILDER_LINEAGE" +LINEAGE_MARKER_RE = re.compile( + r"", + re.DOTALL, +) + +#: Marker *presence*, decided without looking at the payload at all. +#: +#: :data:`LINEAGE_MARKER_RE` only matches a complete, object-shaped, terminated +#: marker, so looking for evidence with it alone means a broken marker is not +#: seen rather than read as broken. Absence and unreadability are opposite +#: answers: one admits an independent reviewer on the ordinary single-builder +#: story, the other must stop. Presence is found first, and the payload is then +#: required to parse. +LINEAGE_MARKER_PRESENT_RE = re.compile(r"" + + +def _reject_duplicate_keys(pairs: Sequence[tuple[str, Any]]) -> dict[str, Any]: + """``object_pairs_hook`` that refuses an object naming a key twice. + + ``json.loads`` keeps the last value for a repeated key, so one marker can + carry two answers to the same question -- two ``episodes`` lists, two + ``schema`` values, two ``resulting_head`` shas inside one episode -- and + every reader silently agrees on whichever came last. Which one describes + the diff is exactly what must not be decided by parser order. This applies + at every depth, so a conflicting nested binding is refused too. + """ + + seen: dict[str, Any] = {} + for key, value in pairs: + if key in seen: + raise ValueError(f"duplicate key in published builder lineage: {key}") + seen[key] = value + return seen + + +def _loads_without_duplicate_keys(payload: str) -> Any: + return json.loads(payload, object_pairs_hook=_reject_duplicate_keys) + + +def episodes_from_comment_body(body: str) -> tuple[ContributionEpisode, ...]: + """Parse lineage markers out of one already trusted comment body. + + The caller decides trust. This function never treats the presence of a + marker as evidence that its author was allowed to publish one. + """ + + if body is None: + body = "" + if not isinstance(body, str): + raise LineageError("published builder lineage is unreadable") + text = _text(body) + present = LINEAGE_MARKER_PRESENT_RE.findall(text) + if not present: + # An ordinary comment. Not evidence, and not a failure either. + return () + if len(present) > 1: + # Two markers on one comment cannot both be "the" published lineage, + # and which one describes the head is exactly what may not be guessed. + raise LineageError("published builder lineage is ambiguous") + matches = LINEAGE_MARKER_RE.findall(text[:MAX_MARKER_BODY_CHARS]) + if len(matches) != 1: + # The marker is there, but no single complete object payload parses out + # of it: unterminated, not an object, or cut off past the bound. + raise LineageError("published builder lineage is unreadable") + try: + payload = _loads_without_duplicate_keys(matches[0]) + except (ValueError, RecursionError): + raise LineageError("published builder lineage is unreadable") from None + if not isinstance(payload, Mapping) or payload.get("schema") != SCHEMA: + raise LineageError("published builder lineage schema is unsupported") + items = payload.get("episodes") + if not isinstance(items, list) or len(items) > MAX_EPISODES: + raise LineageError("published builder lineage is unreadable") + if not items: + # A marker announces lineage. The publisher refuses to publish zero + # episodes, so a trusted marker carrying an empty chain is not a + # history that happens to be empty -- it is a claim that contradicts + # itself, and reading it as ordinary absence is how announced evidence + # disappears into the single-builder answer. + raise LineageError("published builder lineage declares no episodes") + return tuple(episode_from_mapping(item) for item in items) + + +def published_episodes( + comments: Sequence[Mapping[str, Any]], + *, + trusted_author: Callable[[str], bool], +) -> tuple[ContributionEpisode, ...]: + """Collect lineage episodes published by already trusted comment authors. + + The hidden marker is a transport for bounded metadata. Trust comes from the + caller's author check, never from the marker being present, so an untrusted + commenter cannot assert a takeover into existence -- and equally cannot + force a stop by posting a deliberately broken one. + + Records are validated, not skipped: an entry that is not a readable comment + raises rather than quietly shrinking the history it was part of. + """ + + def arrivals(): + for comment in comments or (): + if not isinstance(comment, Mapping): + raise LineageError("the comment history holds an entry that is not a comment") + require_comment_record(comment, what="the comment history") + login = comment_author_login(comment) + if not login or not trusted_author(login): + continue + yield from episodes_from_comment_body(comment.get("body") or "") + + return tuple(bounded_arrivals(arrivals())) + + +def merge_episodes( + recorded: Sequence[Any] = (), + incoming: Sequence[Any] = (), +) -> tuple[ContributionEpisode, ...]: + """Concatenate two sources of raw arrivals, validated and bounded. + + Deliberately no deduplication. Collapsing here would let two independently + collapsed inputs each arrive under the cap and together exceed it, and the + owning resolver already collapses exactly once -- in the one place that can + also see a contradiction at a position and refuse it. Every arrival is + coerced to an episode here, so a malformed member is the documented + contract error rather than an incidental attribute failure later. + """ + + def arrivals(): + for source in (recorded, incoming): + for item in source or (): + yield require_episode(item) + + return tuple(bounded_arrivals(arrivals())) + + +@dataclass(frozen=True) +class LineageContext: + """The trusted exact-head evidence a consumer carries into resolution. + + Every field has to come from something the caller verified for itself: the + repository it is running in, the head it fetched from the pull request, and + episodes published by an author it already trusts. An empty context is not + a failure -- it is the honest statement that this call has no exact-head + evidence, and resolution falls back to the ordinary identity-only answer, + which still refuses a configured branch/label disagreement. + """ + + repo: str = "" + pr_number: Any = 0 + branch: str = "" + head_sha: str = "" + episodes: tuple[ContributionEpisode, ...] = field(default_factory=tuple) + + +#: A consumer that has no exact-head evidence at all. +NO_LINEAGE = LineageContext() + + +def lineage_context( + *, + repo: str, + pr_number: Any, + branch: str = "", + head_sha: str | None = "", + comments: Sequence[Mapping[str, Any]] | None = (), + trusted_author: Callable[[str], bool] | None = None, +) -> LineageContext: + """Assemble exact-head lineage evidence for one consumer entry path. + + A *deliberately* absent target -- nothing supplied at all -- is the + ordinary no-evidence case and yields :data:`NO_LINEAGE`, which callers + resolve through the explicit identity-only route. A partially populated or + malformed target is not absent: it raises, because quietly returning + :data:`NO_LINEAGE` for it skipped branch binding and decided a diff nobody + bound. Evidence that arrives with no target to bind it to raises for the + same reason. Unreadable published evidence propagates as + :class:`LineageError` for the caller's fail-closed handling. + """ + + episodes: tuple[ContributionEpisode, ...] = () + if trusted_author is not None: + episodes = published_episodes(comments or (), trusted_author=trusted_author) + supplied = any(_text(value) for value in (repo, branch, head_sha)) or ( + pr_number not in (None, 0, False, "") + ) + if not supplied: + if episodes: + raise LineageError( + "published builder lineage arrived without a pull request to bind it to" + ) + return NO_LINEAGE + target = require_exact_target( + repo=repo, pr_number=pr_number, branch=branch, head_sha=head_sha + ) + return LineageContext( + repo=target.repo, + pr_number=target.pr_number, + branch=target.branch, + head_sha=target.head_sha, + episodes=episodes, + ) + + +def resolve_lineage_context( + context: LineageContext | None, + *, + identity: Mapping[str, Any] | None, + labels: Sequence[str] = (), + author: str = "", +) -> Lineage: + """Resolve a carried :class:`LineageContext` through the one decision. + + Only a context that is *exactly* absent takes the explicit identity-only + route. A partially populated context, or one carrying episodes, is a claim + about a specific pull request and goes through exact-target resolution, + which refuses it rather than answering about a diff it never bound. + """ + + if context is None or context == NO_LINEAGE: + return resolve_configured_identity( + identity=identity, labels=labels, author=author + ) + return resolve_builder_lineage( + identity=identity, + labels=labels, + author=author, + repo=context.repo, + pr_number=context.pr_number, + branch=context.branch, + head_sha=context.head_sha, + episodes=context.episodes, + ) + + +# --- pure keys and label planning -------------------------------------------- + + +def pr_key(repo: str, pr_number: Any) -> str: + """A stable, opaque key for one pull request's private lineage record. + + Pure by design: the recording side lives in a later stage, but both sides + have to derive the same key from the same pair, and deriving it twice in + two places is how they stop matching. + """ + + seed = json.dumps([_text(repo).lower(), _pr_number(pr_number)], sort_keys=True) + return "l" + hashlib.sha256(seed.encode()).hexdigest()[:62] + + +def builder_label_for(lane: str, identity: Mapping[str, Any] | None = None) -> str: + """The one active label that names ``lane`` as the current writer.""" + + writer = _lane(lane) + if not writer: + return "" + label_map = ( + identity.get("labels") if isinstance(identity, Mapping) else None + ) + if isinstance(label_map, Mapping): + for label in sorted(_text(item) for item in label_map): + if _lane(label_map.get(label)) == writer and label.startswith("builder:"): + return label + return f"builder:{writer}" + + +def builder_label_plan( + lineage: Lineage, + *, + current_labels: Sequence[str] = (), + identity: Mapping[str, Any] | None = None, +) -> dict[str, Any]: + """Plan reconciliation to exactly one active builder label. + + A plan, never an application: this stage computes what would have to change + and nothing mutates. The label set says who may write next, and after a + verified takeover that is exactly one lane. Historical contributions are + not deleted by this -- they live in the recorded lineage, which is what + reviewer exclusion and the status projection read. Unresolved lineage plans + no mutation at all; a label moved on a guess is the failure this contract + exists to stop. + """ + + label_map = identity.get("labels") if isinstance(identity, Mapping) else None + known = ( + {_text(label): _lane(lane) for label, lane in label_map.items()} + if isinstance(label_map, Mapping) + else {} + ) + present = tuple( + dict.fromkeys( + label + for label in (_text(item) for item in current_labels) + if label and (label.startswith("builder:") or known.get(label)) + ) + ) + if not lineage.resolved or not lineage.current_writer: + return { + "schema": SCHEMA, + "status": "blocked", + "reason": lineage.reason if not lineage.resolved else "no_builder_identity", + "head_sha": lineage.head_sha, + "current_writer": lineage.current_writer, + "add": [], + "remove": [], + "owner_action": lineage.owner_action or _OWNER_ACTIONS["label_outside_lineage"], + } + + writer = lineage.current_writer + target = next( + ( + label + for label in present + if known.get(label) == writer or label == f"builder:{writer}" + ), + "", + ) or builder_label_for(writer, identity) + remove = [label for label in present if label != target] + add = [] if target in present else [target] + return { + "schema": SCHEMA, + "status": "reconcile" if (add or remove) else "current", + "reason": lineage.reason, + "head_sha": lineage.head_sha, + "current_writer": writer, + "add": add, + "remove": remove, + "owner_action": "", + } diff --git a/src/code_mower/init.py b/src/code_mower/init.py index ba970690..f9ab824c 100644 --- a/src/code_mower/init.py +++ b/src/code_mower/init.py @@ -226,6 +226,16 @@ "product-support-helper", "0644", ), + ( + # The pure lineage contract the gate helper resolves through. A gate + # runner in a generated product repository has no Code Mower package + # installed, so the dependency has to travel with the helper or the + # import fails and the gate cannot evaluate at all. + "tools/builder_lineage.py", + "builder_lineage.py", + "product-support-helper", + "0644", + ), ( "tools/decisions.py", "decisions.py", diff --git a/src/code_mower/lineage_identity.py b/src/code_mower/lineage_identity.py new file mode 100644 index 00000000..9e20a173 --- /dev/null +++ b/src/code_mower/lineage_identity.py @@ -0,0 +1,408 @@ +"""Canonical reviewer identity and admission: the pure contract. + +Every direct reviewer wrapper used to answer "may I review this?" on its own +terms, leaning on label/author exclusion or a product-specific deny list. None +of those survive a takeover, where the opener, the branch and the active label +can each name a different lane than the one that wrote the current diff. + +This module is the pure half of the one admission seam: given an identity +contract, trusted pull request metadata and the episodes a caller already +established as trusted, it decides whether a reviewer lane is independent of +the diff. Fetching the metadata, reading the environment, loading a private +record and running a provider are all somewhere else, and deliberately so -- +an admission decision that cannot be computed from explicit inputs cannot be +audited from them either. + +Role eligibility is a separate decision and is not consulted here: a qualified +lane can still be a contributor, and an independent lane can still be +unqualified. +""" + +from __future__ import annotations + +import json +from typing import Any, Callable, Mapping, Sequence + +from .builder_lineage import ( + SCHEMA, + ContributionEpisode, + IdentityConflictError, + Lineage, + LineageError, + canonical_identity, + merge_episodes, + published_episodes, + require_comment_list, + resolve_builder_lineage, +) + + +#: The environment variable the deployment's identity contract is rendered +#: into. Named here so consumers and tests agree on it; nothing in this module +#: reads it, because a pure decision may not depend on ambient state. +AUTHOR_EXCLUSION_ENV = "CODE_MOWER_AUTHOR_EXCLUSION_JSON" + +#: Accounts a reviewer lane writes under. Used only as a floor, so an +#: unconfigured or malformed identity contract cannot make a contributing +#: reviewer admissible; it never widens who may review. +LANE_ACCOUNT_FLOOR: Mapping[str, tuple[str, ...]] = { + "devin": ( + "devin-ai-integration", + "devin-ai-integration[bot]", + "devin-cli-audit-bot", + "devin-cli-audit-bot[bot]", + ), + "codex": ("chatgpt-codex-connector[bot]", "codex[bot]"), + "claude": ("claude[bot]", "claude-bot"), +} + + +class ReviewerNotIndependent(RuntimeError): + """Raised when a reviewer lane may not gate the pull request under review. + + ``reason`` and ``owner_action`` are bounded metadata safe to surface in a + public comment; no diagnostic output, path or provider reference is carried. + """ + + def __init__(self, decision: Mapping[str, Any]) -> None: + self.decision = dict(decision) + self.reason = str(decision.get("reason") or "") + self.owner_action = str(decision.get("owner_action") or "") + super().__init__( + f"{decision.get('lane') or 'reviewer'} lane is not admitted: " + f"{self.reason}; {self.owner_action}" + ) + + +#: The deployment's identity contract misnames the reviewer's own lane, or +#: gives one normalized key two answers. It is the core's conflict class rather +#: than a second one, so a contradiction found while canonicalizing and one +#: found while flooring are the same failure to every caller. +ReviewerIdentityInvalid = IdentityConflictError + + +def identity_from_json(raw: str | None) -> Mapping[str, Any]: + """Parse the author-exclusion identity contract from explicit text. + + A missing or unparsable value disables lane naming rather than inventing + one, which keeps an unconfigured checkout behaving as it does today. The + text is always supplied by the caller: reading it from the environment is a + consumer's job, so that this decision stays reproducible from its inputs. + """ + + text = raw or "" + if not text: + return {"enabled": False} + try: + parsed = json.loads(text) + except (ValueError, RecursionError): + return {"enabled": False} + return parsed if isinstance(parsed, Mapping) else {"enabled": False} + + +#: Compatibility spelling for callers that already say ``load_identity``. It +#: requires the raw text explicitly; there is deliberately no environment +#: fallback in the pure contract. +load_identity = identity_from_json + + +def account_key(login: Any) -> str: + """The form account lookups actually use: trimmed and case-folded.""" + + return str(login or "").strip().lower() + + +def normalized_account_map(authors: Any) -> dict: + """The account map keyed the way resolution reads it. + + A thin view onto :func:`~code_mower.builder_lineage.canonical_identity`, + not a second normalization: the reviewer floor and the resolver have to + compose over the very same keys, and the way they stopped doing so was by + each normalizing for itself. + """ + + return canonical_identity({"authors": authors})["authors"] + + +def _claim_own_identity(mapping: dict, key: str, reviewer: str, kind: str) -> None: + """Make ``key`` name ``reviewer``, or refuse if it already names another.""" + + present = mapping.get(key) + named = str(present).strip().lower() if isinstance(present, str) else "" + if named and named != reviewer: + raise ReviewerIdentityInvalid( + f"reviewer_identity_invalid: the configured {kind} `{key}` names " + f"lane `{named}`, but it is the {reviewer} lane's own {kind}. " + f"Correct {AUTHOR_EXCLUSION_ENV} before running a {reviewer} " + f"review; a reviewer that cannot name its own lane cannot be " + f"excluded from its own contribution." + ) + mapping[key] = reviewer + + +def identity_with_lane_floor( + identity: Mapping[str, Any] | None, lane: str +) -> Mapping[str, Any]: + """Guarantee the reviewer lane can be named, whatever the configuration says. + + Reviewer independence is decided by naming lanes. A missing, disabled or + malformed identity contract would name none of them, and an unnameable lane + cannot be recognised as a contributor -- which would silently admit exactly + the reviewer this seam exists to exclude. So the lane's own label and + accounts are always present. Only the reviewer's own lane is synthesized: + this adds exclusion and never admission. + + It is a floor, not a default. ``setdefault`` leaves a present-but-useless + mapping alone -- ``{"builder:codex": ""}`` keeps naming no lane -- so a + blank or malformed own entry is overwritten, and one that names a + *different* lane is a configuration error the reviewer refuses on rather + than silently correcting, because the deployment believes something about + its own identity that is not true. Disabling the contract does not make a + misnamed own lane safe, so the check runs either way. + + The floor raises the fields it is responsible for and leaves the rest of + the deployment's contract -- ``branch_prefixes``, + ``require_verified_lineage`` and anything else -- intact. Rebuilding the + mapping from scratch dropped the configured branch identity every wrapper + was rendered to use, so a ``codex/`` branch labelled ``builder:claude`` + came back a sole Claude writer, admitting Codex to its own diff. + + It composes over the canonical representation, so it floors the very keys + resolution will look up and an alias cannot outrank the canonical account. + """ + + reviewer = str(lane or "").strip().lower() + floored = canonical_identity(identity) + if reviewer: + _claim_own_identity(floored["labels"], f"builder:{reviewer}", reviewer, "label") + for login in LANE_ACCOUNT_FLOOR.get(reviewer, ()): + _claim_own_identity( + floored["authors"], account_key(login), reviewer, "account" + ) + floored["enabled"] = True + return floored + + +def marker_author_trust( + authorities: Sequence[str] = (), +) -> Callable[[str], bool]: + """Who a consumer may read published lineage markers from. + + This is deliberately the *same* rule the gate, the labelers and the + reviewer wrappers apply: the repository's configured decision authorities, + and nobody else. A lineage marker is a transport for bounded metadata, so + being able to post an audit comment on a pull request is not being able to + assert a takeover of it. An unconfigured checkout trusts nobody and reads + no published evidence, which leaves it behaving exactly as it does without + this seam. + + The authority list is supplied by the caller. Resolving it from the + environment here would make the same marker trusted or untrusted depending + on where the predicate happened to be built. + """ + + allowed = { + str(item).strip().lower().lstrip("@") + for item in authorities + if str(item).strip() + } + + def trusted(login: str) -> bool: + return bool(allowed) and str(login).strip().lower().lstrip("@") in allowed + + return trusted + + +def trusted_published_episodes( + comments: Any, + *, + authorities: Sequence[str] = (), + what: str = "the published builder lineage", +) -> tuple[ContributionEpisode, ...]: + """Validate a raw comment history, then read the markers it is trusted for. + + Validation comes first and unconditionally. A successful read that is not a + list of comment objects has not answered the question, and filtering it + down to what happens to be a mapping reports an unreadable history as an + absent one -- which is the answer that admits a reviewer. + """ + + records = require_comment_list(comments, what=what) + trusted = [str(item).strip() for item in authorities if str(item).strip()] + if not trusted: + return () + return published_episodes(records, trusted_author=marker_author_trust(trusted)) + + +def combine_evidence( + recorded: Sequence[ContributionEpisode] = (), + published: Sequence[ContributionEpisode] = (), +) -> tuple[ContributionEpisode, ...]: + """All contribution evidence a reviewer is allowed to read, in order. + + The durable record is the producing host's own; published markers are the + transport for a reviewer running somewhere that record does not exist -- + which is the ordinary case for an independent reviewer host, whose private + store is empty. Reading the private store alone therefore answers "no + takeover happened" on exactly the hosts where the question matters, so both + are carried through as validated, bounded raw arrivals. Nothing is + collapsed here: the owning resolver deduplicates exactly once, where it can + also see a contradiction at a position and refuse it. + """ + + return merge_episodes(recorded, published) + + +def pr_lineage( + *, + repo: str, + pr_number: Any, + pr_meta: Mapping[str, Any], + head_sha: str, + identity: Mapping[str, Any] | None, + episodes: Sequence[ContributionEpisode] = (), +) -> Lineage: + """Resolve lineage from trusted pull request metadata at an exact head. + + ``pr_meta`` must be the metadata the caller fetched from GitHub itself and + ``head_sha`` the head it pinned; passing a head the caller did not verify + would make every downstream decision unverified too. ``identity`` is + explicit for the same reason -- a wrapper and the gate that read the same + pull request must not disagree because one of them found a different + contract in its environment. + + Admission is an exact-target claim, so the complete repository, pull + request number, branch and head are required. Metadata that names no branch + is a ``target_invalid`` conflict, not an identity-only answer about an + unbound diff. + """ + + labels = [ + str(label.get("name") or "") + for label in (pr_meta.get("labels") or []) + if isinstance(label, Mapping) + ] + user = pr_meta.get("user") + author = str((user.get("login") if isinstance(user, Mapping) else "") or "") + head = pr_meta.get("head") + branch = str((head.get("ref") if isinstance(head, Mapping) else "") or "") + # The wrapper decides admission from the same composition the gate does, so + # the configured branch identity reaches it here too -- including when there + # are no episodes at all. + return resolve_builder_lineage( + identity=identity, + labels=labels, + author=author, + repo=repo, + pr_number=pr_number, + branch=branch, + head_sha=head_sha, + episodes=episodes, + ) + + +def reviewer_admission( + lane: str, + *, + repo: str, + pr_number: Any, + pr_meta: Mapping[str, Any], + head_sha: str, + identity: Mapping[str, Any] | None, + episodes: Sequence[ContributionEpisode] = (), +) -> dict[str, Any]: + """Decide whether ``lane`` may review this exact head. Fails closed. + + The reviewer's own lane is floored before resolution, so a contract that + cannot name it still excludes it. An identity contract that *misnames* it + raises :class:`ReviewerIdentityInvalid` rather than resolving, because a + deployment that is wrong about its own reviewer has not asked a question + this seam can answer. + """ + + floored = identity_with_lane_floor(identity, lane) + try: + lineage = pr_lineage( + repo=repo, + pr_number=pr_number, + pr_meta=pr_meta, + head_sha=head_sha, + identity=floored, + episodes=episodes, + ) + except LineageError: + return { + "schema": SCHEMA, + "lane": str(lane or "").strip().lower(), + "admitted": False, + "reason": "lineage_unreadable", + "head_sha": str(head_sha or ""), + "contributors": [], + "current_writer": "", + "owner_action": ( + "builder contribution evidence for this pull request could not " + "be read; re-record it from the verified handoff" + ), + } + return lineage.admission(lane) + + +def require_independent_reviewer( + lane: str, + *, + repo: str, + pr_number: Any, + pr_meta: Mapping[str, Any], + head_sha: str, + identity: Mapping[str, Any] | None, + episodes: Sequence[ContributionEpisode] = (), +) -> dict[str, Any]: + """Admit ``lane`` or raise :class:`ReviewerNotIndependent`.""" + + decision = reviewer_admission( + lane, + repo=repo, + pr_number=pr_number, + pr_meta=pr_meta, + head_sha=head_sha, + identity=identity, + episodes=episodes, + ) + if not decision["admitted"]: + raise ReviewerNotIndependent(decision) + return decision + + +def require_reviewer_lane( + lane: str, + repo: str, + pr_number: Any, + pr_meta: Mapping[str, Any], + head_sha: str, + *, + identity: Mapping[str, Any] | None, + episodes: Sequence[ContributionEpisode] = (), +) -> dict[str, Any]: + """Admission that reports refusal as a plain ``RuntimeError``. + + Direct reviewer wrappers already surface ``RuntimeError`` as an operator + message with their normal exit handling, so this keeps the shared decision + from needing per-wrapper exception plumbing. The message is bounded + metadata plus one owner action; no diagnostic output or path is included. + """ + + try: + return require_independent_reviewer( + lane, + repo=repo, + pr_number=pr_number, + pr_meta=pr_meta, + head_sha=head_sha, + identity=identity, + episodes=episodes, + ) + except ReviewerNotIndependent as exc: + raise RuntimeError( + f"{lane} reviewer lane is not admitted for {repo}#{pr_number} at " + f"{str(head_sha)[:12]}: {exc.reason}; {exc.owner_action}" + ) from None diff --git a/src/code_mower/package_manifest.py b/src/code_mower/package_manifest.py index d9387afb..94c781c7 100644 --- a/src/code_mower/package_manifest.py +++ b/src/code_mower/package_manifest.py @@ -362,6 +362,11 @@ ("tools/blind_review_artifacts.py", "src/code_mower/blind_review_artifacts.py", "core"), ("tools/audit_handoff_log.py", "src/code_mower/audit_handoff_log.py", "core"), ("tools/audit_labeler_lib.py", "src/code_mower/audit_labeler_lib.py", "core"), + # The vendored copy is the canonical one, exactly as for the labeler helper: + # a generated product gate imports `tools/`, so shipping the package module + # from anywhere else would let the two drift where no `src/` assertion looks. + ("tools/builder_lineage.py", "src/code_mower/builder_lineage.py", "core"), + ("src/code_mower/lineage_identity.py", "src/code_mower/lineage_identity.py", "core"), ("tools/audit_limits.py", "src/code_mower/audit_limits.py", "core"), ("tools/audit_progress.py", "src/code_mower/audit_progress.py", "core"), ( diff --git a/tests/lineage_contract_fixtures.py b/tests/lineage_contract_fixtures.py new file mode 100644 index 00000000..9038e871 --- /dev/null +++ b/tests/lineage_contract_fixtures.py @@ -0,0 +1,226 @@ +"""Pure fixtures for the builder lineage contract. + +Deliberately self-contained. The lineage contract is pure, so its fixtures must +be too: importing a consumer test module here would pull an adapter -- and its +environment, store and network expectations -- into tests whose whole point is +that the decision needs none of them. Nothing in this module reads the +environment, touches the filesystem or imports anything outside +``code_mower.builder_lineage``. +""" + +from __future__ import annotations + +from typing import Any + +from code_mower.builder_lineage import ( + CONTINUATION_WRITER_STATE, + ContributionEpisode, + lineage_comment_marker, +) + + +REPO = "codemower-ai/code-mower" +BRANCH = "devin/959-release-dogfood" +PR = 959 + +OPENED = "a" * 40 +TAKEN = "b" * 40 +MOVED = "c" * 40 + +AUTHORITY = "codemower-ai" +OUTSIDER = "a-passer-by" + +#: The deployment identity contract the product renders, including the branch +#: provenance fields that are read only together. +IDENTITY: dict[str, Any] = { + "enabled": True, + "labels": { + "builder:devin": "devin", + "builder:codex": "codex", + "builder:claude": "claude", + }, + "authors": { + "devin-ai-integration[bot]": "devin", + "chatgpt-codex-connector[bot]": "codex", + "claude[bot]": "claude", + }, + "branch_prefixes": { + "devin/": "devin", + "codex/": "codex", + "claude/": "claude", + "feature/cx-": "codex", + }, + "require_verified_lineage": True, +} + +#: The same contract without the branch fields: a deployment that never asked +#: for verified lineage keeps the answer it has always had. +UNCONFIGURED: dict[str, Any] = { + "enabled": True, + "labels": dict(IDENTITY["labels"]), + "authors": dict(IDENTITY["authors"]), +} + + +def head(index: int) -> str: + """A distinct, well-formed 40-hex head for chain position ``index``.""" + + return f"{index:040x}" + + +def takeover(**overrides: Any) -> ContributionEpisode: + """The primary fixture: Devin opened it, Codex verifiably took it over.""" + + payload: dict[str, Any] = dict( + sequence=1, + repo=REPO, + pr_number=PR, + branch=BRANCH, + source_lane="devin", + destination_lane="codex", + expected_head=OPENED, + resulting_head=TAKEN, + writer_state="terminated", + ) + payload.update(overrides) + return ContributionEpisode(**payload) + + +def continuation(*, sequence: int, expected: str, resulting: str, lane: str = "codex"): + """An ordinary fix round by the lane that already holds the pen.""" + + return ContributionEpisode( + sequence=sequence, + kind="continuation", + repo=REPO, + pr_number=PR, + branch=BRANCH, + source_lane=lane, + destination_lane=lane, + expected_head=expected, + resulting_head=resulting, + writer_state=CONTINUATION_WRITER_STATE, + ) + + +def variant(episode: ContributionEpisode, **overrides: Any) -> ContributionEpisode: + """The same episode with fields replaced, still strictly constructed.""" + + payload = { + "sequence": episode.sequence, + "kind": episode.kind, + "repo": episode.repo, + "pr_number": episode.pr_number, + "branch": episode.branch, + "source_lane": episode.source_lane, + "destination_lane": episode.destination_lane, + "expected_head": episode.expected_head, + "resulting_head": episode.resulting_head, + "writer_state": episode.writer_state, + } + payload.update(overrides) + return ContributionEpisode(**payload) + + +def chain(length: int) -> tuple[ContributionEpisode, ...]: + """A takeover followed by ``length - 1`` continuations by the new writer.""" + + episodes = [takeover(resulting_head=head(1))] + for index in range(2, length + 1): + episodes.append( + continuation( + sequence=index, + expected=episodes[-1].resulting_head, + resulting=head(index), + ) + ) + return tuple(episodes) + + +def cumulative_comments(episodes, *, author: str = AUTHORITY) -> list[dict[str, Any]]: + """One published comment per round, each carrying the whole chain so far. + + This is what the supported publication contract actually produces, so it is + what the arrival bound has to accommodate. + """ + + return [ + comment(body=lineage_comment_marker(episodes[:length]), author=author) + for length in range(1, len(episodes) + 1) + ] + + +def comment(*, body: str, author: str = AUTHORITY, field: str = "user") -> dict[str, Any]: + """One comment record under either supported transport's author field.""" + + return {field: {"login": author}, "body": body} + + +def published(episodes, *, author: str = AUTHORITY, field: str = "user"): + """A single trusted comment carrying one published lineage marker.""" + + return comment(body=lineage_comment_marker(tuple(episodes)), author=author, field=field) + + +def pr_meta( + *, + author: str = "devin-ai-integration[bot]", + labels: tuple[str, ...] = ("builder:codex",), + branch: str = BRANCH, + sha: str = TAKEN, +) -> dict[str, Any]: + """Trusted pull request metadata, in GitHub's own shape.""" + + return { + "user": {"login": author}, + "head": {"ref": branch, "sha": sha}, + "labels": [{"name": name} for name in labels], + } + + +#: Records whose meaning cannot be recovered. Each is a *present* field holding +#: the wrong type, or a null in a position GitHub never nulls. +MALFORMED_COMMENT_RECORDS: tuple[dict[str, Any], ...] = ( + {"user": {"login": AUTHORITY}, "body": 12345}, + {"user": {"login": AUTHORITY}, "body": {"text": "hi"}}, + {"user": {"login": AUTHORITY}, "body": ["hi"]}, + {"user": AUTHORITY, "body": "hi"}, + {"user": 7, "body": "hi"}, + {"user": [AUTHORITY], "body": "hi"}, + {"user": {"login": {"name": AUTHORITY}}, "body": "hi"}, + {"user": {"login": 7}, "body": "hi"}, + {"user": {"login": [AUTHORITY]}, "body": "hi"}, + {"user": {"login": None}, "body": "hi"}, + {"user": {"login": AUTHORITY}, "body": None}, + {"author": {"login": None}, "body": "hi"}, + {"author": AUTHORITY, "body": "hi"}, + {"author": {"login": 7}, "body": "hi"}, +) + +#: GitHub's own schema, which must keep working: a comment from a deleted +#: account carries ``user: null``, and ``body`` is optional on some +#: representations. Neither names an author or a marker, and neither is an +#: error. +VALID_COMMENT_RECORDS: tuple[dict[str, Any], ...] = ( + {"user": None, "body": "a deleted account said this"}, + {"user": {"login": AUTHORITY}}, + {"user": {}, "body": "an author object naming nobody"}, + {"body": "a record with no author field at all"}, + {"author": None, "body": "the gh/GraphQL transport, deleted account"}, + {"author": {"login": AUTHORITY}, "body": "the gh/GraphQL transport"}, + {"user": {"login": AUTHORITY}, "body": "ordinary comment"}, +) + +#: Successful reads that carry no readable history. None of them is "no +#: comments", and normalising them into one is what admits a reviewer onto a +#: diff whose takeover marker was in the part that got dropped. +INVALID_COMMENT_RESPONSES: tuple[Any, ...] = ( + None, + False, + {}, + {"comments": [{"user": {"login": AUTHORITY}, "body": "hi"}]}, + "a string", + [{"user": {"login": AUTHORITY}, "body": "hi"}, "not a comment"], + [None], + [[{"user": {"login": AUTHORITY}, "body": "hi"}]], +) diff --git a/tests/test_builder_lineage_contract.py b/tests/test_builder_lineage_contract.py new file mode 100644 index 00000000..b9e33288 --- /dev/null +++ b/tests/test_builder_lineage_contract.py @@ -0,0 +1,949 @@ +"""Exact-head builder contribution lineage: the owning contract regressions. + +The primary fixture is the shape that produced the original defect: a pull +request opened by Devin on a Devin branch, an explicit verified Codex takeover, +and a Codex final head. Every signal that used to be consulted on its own -- +the opener, the active label, the branch prefix -- names a different lane here. + +These cases run on explicit plain inputs with no environment, store, network or +adapter involvement. That is not an accident of how they are written; it is the +property the contract is being accepted for. +""" + +from __future__ import annotations + +import dataclasses +import inspect +import unittest + +from code_mower.builder_lineage import ( + MAX_EPISODE_ARRIVALS, + MAX_EPISODES, + ContributionEpisode, + IdentityConflictError, + LineageError, + branch_lane_from_identity, + builder_label_for, + builder_label_plan, + canonical_identity, + continuation_episode, + episode_from_handoff, + episode_from_mapping, + lanes_from_identity, + pr_key, + require_exact_target, + resolve_builder_lineage, + resolve_configured_identity, + resolve_identity_only, + resolve_lineage, +) + +from lineage_contract_fixtures import ( + BRANCH, + IDENTITY, + MOVED, + OPENED, + PR, + REPO, + TAKEN, + UNCONFIGURED, + chain, + continuation, + head, + takeover, + variant, +) + + +def resolve(**overrides): + kwargs = dict( + repo=REPO, + pr_number=PR, + branch=BRANCH, + head_sha=TAKEN, + episodes=(takeover(),), + opener_lane="devin", + label_lanes=("codex",), + ) + kwargs.update(overrides) + return resolve_lineage(**kwargs) + + +class ExactHeadBindingTests(unittest.TestCase): + """Repository, pull request, branch and head bind an episode immutably.""" + + def test_a_verified_takeover_names_both_contributors_and_one_writer(self): + lineage = resolve() + self.assertEqual(lineage.status, "resolved") + self.assertEqual(lineage.contributors, ("devin", "codex")) + self.assertEqual(lineage.current_writer, "codex") + self.assertEqual(lineage.builder_label, "builder:codex") + self.assertEqual(lineage.stale_builder_labels, ()) + self.assertFalse(lineage.independent("devin")) + self.assertFalse(lineage.independent("codex")) + self.assertTrue(lineage.independent("claude")) + self.assertEqual( + lineage.independent_lanes(("devin", "codex", "claude")), ("claude",) + ) + + def test_a_stale_label_is_reported_rather_than_treated_as_a_conflict(self): + lineage = resolve(label_lanes=("devin",)) + self.assertEqual(lineage.status, "resolved") + self.assertEqual(lineage.current_writer, "codex") + self.assertEqual(lineage.stale_builder_labels, ("devin",)) + + def test_a_label_for_an_uninvolved_lane_fails_closed(self): + lineage = resolve(label_lanes=("claude",)) + self.assertEqual(lineage.status, "conflict") + self.assertEqual(lineage.reason, "label_outside_lineage") + self.assertTrue(lineage.owner_action) + self.assertFalse(lineage.independent("claude")) + + def test_an_opener_outside_the_lineage_fails_closed(self): + self.assertEqual( + resolve(opener_lane="claude", label_lanes=()).reason, + "opener_outside_lineage", + ) + + def test_evidence_bound_elsewhere_is_not_evidence_about_this_pull_request(self): + for overrides in ( + {"repo": "other/repo"}, + {"pr_number": 960}, + {"branch": "codex/959-other"}, + ): + with self.subTest(**overrides): + lineage = resolve(episodes=(takeover(**overrides),)) + self.assertEqual(lineage.status, "conflict") + self.assertEqual(lineage.reason, "episode_unbound") + + def test_lineage_short_of_the_current_head_waits_rather_than_guessing(self): + lineage = resolve(head_sha=MOVED) + self.assertEqual(lineage.status, "waiting") + self.assertEqual(lineage.reason, "lineage_behind_head") + self.assertEqual(lineage.contributors, ()) + self.assertEqual(lineage.current_writer, "") + self.assertFalse(lineage.independent("claude")) + + def test_an_abbreviated_or_missing_head_is_never_resolved(self): + for head_sha in ("", "abc1234", TAKEN[:39], TAKEN.upper() + "0"): + with self.subTest(head_sha=head_sha): + lineage = resolve(head_sha=head_sha) + self.assertEqual(lineage.status, "conflict") + self.assertEqual(lineage.reason, "target_invalid") + + def test_a_destination_that_never_moved_the_head_is_writer_not_contributor(self): + lineage = resolve( + episodes=(takeover(resulting_head=OPENED),), + head_sha=OPENED, + label_lanes=("codex",), + ) + self.assertEqual(lineage.contributors, ("devin",)) + self.assertEqual(lineage.current_writer, "codex") + self.assertTrue(lineage.independent("claude")) + self.assertFalse(lineage.independent("devin")) + + def test_a_second_takeover_preserves_the_whole_ordered_history(self): + lineage = resolve( + episodes=( + takeover(), + takeover( + sequence=2, + source_lane="codex", + destination_lane="claude", + expected_head=TAKEN, + resulting_head=MOVED, + ), + ), + head_sha=MOVED, + label_lanes=("claude",), + ) + self.assertEqual(lineage.contributors, ("devin", "codex", "claude")) + self.assertEqual(lineage.current_writer, "claude") + self.assertEqual(lineage.independent_lanes(("devin", "codex", "claude")), ()) + + +class EpisodeShapeTests(unittest.TestCase): + """A handoff and a continuation cannot be forged into one another.""" + + def test_construction_rejects_self_handoffs_and_malformed_fields(self): + for overrides in ( + {"destination_lane": "devin"}, + {"expected_head": "zz"}, + {"sequence": 0}, + {"sequence": MAX_EPISODES + 1}, + {"sequence": True}, + {"writer_state": "running"}, + {"kind": "something-else"}, + {"repo": "no-slash"}, + {"pr_number": 0}, + ): + with self.subTest(**overrides): + with self.assertRaises(LineageError): + takeover(**overrides) + + def test_the_constructor_rejects_every_unusable_pull_request_number(self): + """`0` and `False` equal the normalizer's "unusable" sentinel. + + Checking the field by round-tripping it through that normalizer and + comparing accepted exactly the two values it uses to say "no", so an + episode could bind itself to a pull request that cannot exist. The + constructor asks whether the value *is* a pull request number instead. + """ + + for pr_number in (0, False, True, -1, -959, 2**31, 2**63): + with self.subTest(pr_number=pr_number): + with self.assertRaises(LineageError): + takeover(pr_number=pr_number) + + def test_the_constructor_takes_an_integer_and_does_not_parse_one(self): + """Parsing is `episode_from_mapping`'s job, done before construction.""" + + for pr_number in ("959", " 959 ", 959.0, None, [959], {"n": 959}): + with self.subTest(pr_number=pr_number): + with self.assertRaises(LineageError): + takeover(pr_number=pr_number) + self.assertEqual(episode_from_mapping(takeover().as_dict()).pr_number, PR) + + def test_valid_pull_request_numbers_at_both_bounds_construct(self): + for pr_number in (1, 2, PR, 2**31 - 1): + with self.subTest(pr_number=pr_number): + self.assertEqual(takeover(pr_number=pr_number).pr_number, pr_number) + + def test_a_zero_pull_request_number_is_malformed_wherever_it_arrives(self): + payload = takeover().as_dict() + payload["pr_number"] = 0 + with self.assertRaises(LineageError): + episode_from_mapping(payload) + self.assertEqual( + resolve(episodes=(payload,), label_lanes=()).reason, "episode_malformed" + ) + + def test_a_continuation_must_be_same_lane_and_self_quiescent(self): + good = continuation(sequence=2, expected=TAKEN, resulting=MOVED) + self.assertEqual(good.source_lane, good.destination_lane) + for overrides in ( + {"destination_lane": "claude"}, + {"writer_state": "terminated"}, + ): + with self.subTest(**overrides): + with self.assertRaises(LineageError): + variant(good, **overrides) + + def test_a_handoff_may_not_carry_the_continuation_writer_state(self): + with self.assertRaises(LineageError): + takeover(writer_state="self_quiescent") + + def test_only_the_recorded_current_writer_may_continue_the_lineage(self): + tip = takeover() + self.assertEqual( + continuation_episode(tip, lane="codex", resulting_head=MOVED).sequence, 2 + ) + for lane in ("devin", "claude", "", "not a lane"): + with self.subTest(lane=lane): + with self.assertRaises(LineageError): + continuation_episode(tip, lane=lane, resulting_head=MOVED) + + def test_a_continuation_must_move_the_head_and_stay_on_the_branch(self): + tip = takeover() + with self.assertRaises(LineageError): + continuation_episode(tip, lane="codex", resulting_head=tip.resulting_head) + with self.assertRaises(LineageError): + continuation_episode( + tip, lane="codex", resulting_head=MOVED, branch="codex/elsewhere" + ) + + def test_episode_parsing_is_strict_about_its_field_set(self): + payload = takeover().as_dict() + self.assertEqual(episode_from_mapping(payload), takeover()) + for mutate in ( + lambda item: item.pop("kind"), + lambda item: item.update({"extra": 1}), + lambda item: item.update({"schema": "something.else"}), + ): + with self.subTest(mutate=mutate): + broken = takeover().as_dict() + mutate(broken) + with self.assertRaises(LineageError): + episode_from_mapping(broken) + + def test_an_episode_is_built_from_a_plain_handoff_record(self): + record = { + "target_pr": f"{REPO}#{PR}", + "target_branch": BRANCH, + "source_lane": "devin", + "destination_lane": "codex", + "expected_head": OPENED, + } + episode = episode_from_handoff( + record, resulting_head=TAKEN, writer_state="terminated", sequence=1 + ) + self.assertEqual(episode, takeover()) + with self.assertRaises(LineageError): + episode_from_handoff( + record, + resulting_head=TAKEN, + writer_state="terminated", + sequence=1, + repo="other/repo", + ) + with self.assertRaises(LineageError): + episode_from_handoff( + {"target_pr": "no-hash"}, + resulting_head=TAKEN, + writer_state="terminated", + sequence=1, + ) + + +class BoundedReplayTests(unittest.TestCase): + """Republishing one chain must not look like a malformed lineage. + + The producer publishes the whole chain on every round and a reader merges + that with whatever private record it holds, so a lineage that runs to its + documented full length arrives as ``1 + 2 + ... + 32`` entries plus the + completed chain once more. Counting arrivals against the *lineage* bound + calls an authorised replay malformed -- and refuses it before deduplication, + the only step that could have shown those arrivals to be one chain. + """ + + def test_an_identical_replay_collapses_instead_of_duplicating(self): + lineage = resolve(episodes=(takeover(), takeover(), takeover())) + self.assertEqual(lineage.status, "resolved") + self.assertEqual(lineage.episodes, 1) + self.assertEqual(lineage.contributors, ("devin", "codex")) + + def test_eight_snapshots_of_an_eight_episode_chain_still_resolve(self): + links = chain(8) + arrivals = tuple(episode for _ in range(8) for episode in links) + self.assertGreater(len(arrivals), MAX_EPISODES) + lineage = resolve_lineage( + repo=REPO, + pr_number=PR, + branch=BRANCH, + head_sha=links[-1].resulting_head, + episodes=arrivals, + ) + self.assertEqual(lineage.status, "resolved") + self.assertEqual(lineage.episodes, 8) + self.assertEqual(lineage.current_writer, "codex") + + def test_the_documented_arrival_maximum_is_the_cumulative_total(self): + self.assertEqual(MAX_EPISODES, 32) + self.assertEqual(MAX_EPISODES * (MAX_EPISODES + 1) // 2, 528) + self.assertEqual(MAX_EPISODE_ARRIVALS, 560) + + def test_every_one_of_the_thirty_two_cumulative_snapshots_resolves(self): + links = chain(MAX_EPISODES) + cumulative = tuple( + episode for length in range(1, len(links) + 1) for episode in links[:length] + ) + self.assertEqual(len(cumulative), 528) + lineage = resolve_lineage( + repo=REPO, + pr_number=PR, + branch=BRANCH, + head_sha=links[-1].resulting_head, + episodes=cumulative, + ) + self.assertEqual(lineage.status, "resolved") + self.assertEqual(lineage.episodes, MAX_EPISODES) + self.assertEqual(lineage.current_writer, "codex") + self.assertEqual(lineage.contributors, ("devin", "codex")) + + def test_the_public_and_private_union_resolves_exactly_at_the_bound(self): + links = chain(MAX_EPISODES) + cumulative = tuple( + episode for length in range(1, len(links) + 1) for episode in links[:length] + ) + arrivals = cumulative + links + self.assertEqual(len(arrivals), MAX_EPISODE_ARRIVALS) + lineage = resolve_lineage( + repo=REPO, + pr_number=PR, + branch=BRANCH, + head_sha=links[-1].resulting_head, + episodes=arrivals, + ) + self.assertEqual(lineage.status, "resolved") + self.assertEqual(lineage.episodes, MAX_EPISODES) + + def test_one_arrival_past_the_contract_is_refused_without_being_walked(self): + links = chain(MAX_EPISODES) + cumulative = tuple( + episode for length in range(1, len(links) + 1) for episode in links[:length] + ) + arrivals = cumulative + links + (links[-1],) + self.assertEqual(len(arrivals), MAX_EPISODE_ARRIVALS + 1) + lineage = resolve_lineage( + repo=REPO, + pr_number=PR, + branch=BRANCH, + head_sha=links[-1].resulting_head, + episodes=arrivals, + ) + self.assertEqual(lineage.status, "conflict") + self.assertEqual(lineage.reason, "episode_malformed") + + def test_a_sequence_past_the_lineage_bound_never_constructs(self): + links = chain(MAX_EPISODES) + with self.assertRaises(LineageError): + variant(links[-1], sequence=MAX_EPISODES + 1) + + def test_a_disagreeing_duplicate_inside_a_full_history_fails_closed(self): + links = chain(MAX_EPISODES) + cumulative = tuple( + episode for length in range(1, len(links) + 1) for episode in links[:length] + ) + forged = variant(links[4], destination_lane="claude", source_lane="claude") + lineage = resolve_lineage( + repo=REPO, + pr_number=PR, + branch=BRANCH, + head_sha=links[-1].resulting_head, + episodes=cumulative + (forged,), + ) + self.assertEqual(lineage.status, "conflict") + self.assertEqual(lineage.reason, "episode_duplicated") + + def test_a_stale_full_history_waits_and_a_rebound_one_conflicts(self): + links = chain(MAX_EPISODES) + waiting = resolve_lineage( + repo=REPO, pr_number=PR, branch=BRANCH, head_sha=MOVED, episodes=links + ) + self.assertEqual(waiting.reason, "lineage_behind_head") + unbound = resolve_lineage( + repo=REPO, + pr_number=PR, + branch="codex/959-other", + head_sha=links[-1].resulting_head, + episodes=links, + ) + self.assertEqual(unbound.reason, "episode_unbound") + + +class ConsistentFailureTests(unittest.TestCase): + """Duplicated, unchained, unbound or unverified evidence fails closed.""" + + def test_reordered_gapped_and_unchained_episodes_fail_closed(self): + broken = resolve( + episodes=( + takeover(), + takeover( + sequence=2, + source_lane="codex", + destination_lane="claude", + expected_head=MOVED, + resulting_head=head(9), + ), + ), + head_sha=head(9), + label_lanes=(), + ) + self.assertEqual(broken.reason, "episode_unchained") + + gap = resolve(episodes=(takeover(sequence=2),), label_lanes=("codex",)) + self.assertEqual(gap.reason, "episode_unchained") + + def test_a_lineage_that_begins_with_a_continuation_is_not_trustworthy(self): + lone = continuation(sequence=1, expected=OPENED, resulting=TAKEN) + lineage = resolve(episodes=(lone,), label_lanes=()) + self.assertEqual(lineage.reason, "episode_unchained") + + def test_two_episodes_claiming_one_position_fail_closed(self): + duplicated = resolve( + episodes=(takeover(), takeover(destination_lane="claude")), label_lanes=() + ) + self.assertEqual(duplicated.reason, "episode_duplicated") + + def test_malformed_or_unverified_writer_state_fails_closed(self): + self.assertEqual(resolve(episodes=({"schema": "nope"},)).reason, "episode_malformed") + raw = takeover().as_dict() + raw["writer_state"] = "unknown" + self.assertIn( + resolve(episodes=(raw,)).reason, + {"episode_malformed", "writer_state_unverified"}, + ) + + def test_every_unresolved_result_carries_one_owner_action(self): + for lineage in ( + resolve(head_sha=MOVED), + resolve(label_lanes=("claude",)), + resolve(episodes=({"schema": "nope"},)), + resolve(episodes=(), label_lanes=("codex",)), + ): + with self.subTest(reason=lineage.reason): + self.assertNotEqual(lineage.status, "resolved") + self.assertTrue(lineage.owner_action) + self.assertEqual(lineage.current_writer, "") + self.assertEqual(lineage.contributors, ()) + + +class IdentityOnlyTests(unittest.TestCase): + """The ordinary no-evidence case, and the disagreements it must refuse.""" + + def test_the_ordinary_single_builder_case_still_resolves(self): + lineage = resolve_lineage( + repo=REPO, + pr_number=PR, + branch="claude/963-lineage", + head_sha=TAKEN, + opener_lane="claude", + label_lanes=("claude",), + ) + self.assertEqual(lineage.status, "resolved") + self.assertEqual(lineage.reason, "single_builder") + self.assertEqual(lineage.contributors, ("claude",)) + self.assertFalse(lineage.independent("claude")) + self.assertTrue(lineage.independent("codex")) + + def test_a_pull_request_with_no_builder_identity_excludes_nobody(self): + lineage = resolve_lineage( + repo=REPO, pr_number=PR, branch="fix/typo", head_sha=TAKEN + ) + self.assertEqual(lineage.status, "resolved") + self.assertEqual(lineage.reason, "no_builder_identity") + self.assertEqual(lineage.contributors, ()) + self.assertTrue(lineage.independent("codex")) + + def test_a_conflicting_author_and_label_without_a_handoff_fails_closed(self): + lineage = resolve_lineage( + repo=REPO, + pr_number=PR, + branch=BRANCH, + head_sha=TAKEN, + opener_lane="devin", + label_lanes=("codex",), + ) + self.assertEqual(lineage.status, "conflict") + self.assertEqual(lineage.reason, "conflicting_builder_identity") + self.assertIn("record the handoff", lineage.owner_action) + + def test_identity_only_resolution_needs_no_head(self): + self.assertEqual( + resolve_identity_only(opener_lane="claude", label_lanes=("claude",)).current_writer, + "claude", + ) + self.assertEqual( + resolve_identity_only(opener_lane="devin", label_lanes=("codex",)).status, + "conflict", + ) + + def test_identity_maps_labels_and_authors_onto_lanes(self): + opener, labels = lanes_from_identity( + identity=IDENTITY, + labels=["builder:codex", "tier:R"], + author="devin-ai-integration[bot]", + ) + self.assertEqual(opener, "devin") + self.assertEqual(labels, ("codex",)) + self.assertEqual( + lanes_from_identity( + identity={"enabled": False}, + labels=["builder:codex"], + author="devin-ai-integration[bot]", + ), + ("", ()), + ) + + def test_the_branch_prefix_is_read_only_with_verified_lineage_configured(self): + self.assertEqual( + branch_lane_from_identity(identity=IDENTITY, branch="codex/topic"), "codex" + ) + self.assertEqual( + branch_lane_from_identity(identity=IDENTITY, branch="feature/cx-topic"), + "codex", + ) + self.assertEqual( + branch_lane_from_identity(identity=UNCONFIGURED, branch="codex/topic"), "" + ) + self.assertEqual( + branch_lane_from_identity(identity=IDENTITY, branch="fix/typo"), "" + ) + + def test_the_longest_configured_prefix_wins(self): + identity = dict(IDENTITY) + identity["branch_prefixes"] = {"codex-": "claude", "codex-review/": "codex"} + self.assertEqual( + branch_lane_from_identity(identity=identity, branch="codex-review/topic"), + "codex", + ) + + +class CanonicalIdentityTests(unittest.TestCase): + """One normalization, consumed by every reader of the contract.""" + + def aliased(self, prefixes): + identity = dict(IDENTITY) + identity["branch_prefixes"] = dict(prefixes) + return identity + + def test_canonicalizing_is_idempotent(self): + once = canonical_identity(IDENTITY) + self.assertEqual(canonical_identity(once), once) + + def test_it_leaves_every_other_configured_field_alone(self): + contract = canonical_identity(IDENTITY) + self.assertTrue(contract["require_verified_lineage"]) + self.assertEqual(contract["labels"], IDENTITY["labels"]) + self.assertEqual(canonical_identity(None), {"labels": {}, "authors": {}, "branch_prefixes": {}}) + + def test_padded_and_cased_keys_collapse_without_losing_the_value(self): + """Trimming the key used to strand the lookup on the untrimmed map.""" + + for prefixes in ( + {" Codex/ ": "codex"}, + {"CODEX/": "codex"}, + {"codex/": "codex", " CODEX/ ": "codex"}, + ): + with self.subTest(prefixes=prefixes): + self.assertEqual( + branch_lane_from_identity( + identity=self.aliased(prefixes), branch="codex/topic" + ), + "codex", + ) + + def test_a_padded_account_alias_still_names_its_lane(self): + identity = dict(IDENTITY) + identity["authors"] = {" ChatGPT-Codex-Connector[Bot] ": "codex"} + opener, _ = lanes_from_identity( + identity=identity, labels=[], author="chatgpt-codex-connector[bot]" + ) + self.assertEqual(opener, "codex") + + def test_conflicting_branch_prefix_aliases_refuse_in_either_order(self): + for prefixes in ( + {"Codex/": "claude", "codex/": "codex"}, + {"codex/": "codex", "CODEX/": "claude"}, + {" codex/ ": "devin", "codex/": "codex"}, + ): + with self.subTest(prefixes=prefixes): + with self.assertRaises(IdentityConflictError): + branch_lane_from_identity( + identity=self.aliased(prefixes), branch="codex/topic" + ) + + def test_the_longest_prefix_is_chosen_among_canonical_keys(self): + identity = self.aliased({" Codex- ": "claude", "CODEX-REVIEW/": "codex"}) + self.assertEqual( + branch_lane_from_identity(identity=identity, branch="codex-review/topic"), + "codex", + ) + + def test_the_resolver_and_the_reviewer_floor_agree_on_one_representation(self): + from code_mower.lineage_identity import identity_with_lane_floor + + identity = dict(IDENTITY) + identity["authors"] = {"Codex[Bot]": "codex"} + floored = identity_with_lane_floor(identity, "codex") + self.assertEqual(floored["authors"]["codex[bot]"], "codex") + self.assertEqual(canonical_identity(floored)["authors"], floored["authors"]) + self.assertEqual( + branch_lane_from_identity(identity=floored, branch="codex/topic"), "codex" + ) + + +class OneSharedDecisionTests(unittest.TestCase): + """The complete identity-plus-branch-plus-evidence decision, in one place. + + Every consumer has to reach the same answer from the same inputs, including + when there are no episodes at all. Treating the empty-episode case as a + separate, easier question is how a configured branch/label disagreement came + back ``resolved`` and let a lane review its own diff. + """ + + def compose(self, **overrides): + kwargs = dict( + identity=IDENTITY, + labels=["builder:claude"], + author="a-human", + repo=REPO, + pr_number=PR, + branch="codex/topic", + head_sha=TAKEN, + episodes=(), + ) + kwargs.update(overrides) + return resolve_builder_lineage(**kwargs) + + def test_a_configured_branch_label_disagreement_is_unresolved_with_no_episodes(self): + lineage = self.compose() + self.assertEqual(lineage.episodes, 0) + self.assertEqual(lineage.status, "conflict") + self.assertEqual(lineage.reason, "conflicting_builder_identity") + self.assertFalse(lineage.independent("codex")) + self.assertFalse(lineage.independent("claude")) + + def test_a_custom_configured_prefix_conflicts_the_same_way(self): + self.assertEqual(self.compose(branch="feature/cx-topic").status, "conflict") + + def test_a_matched_branch_and_label_stay_ordinary(self): + lineage = self.compose(branch="claude/topic") + self.assertEqual(lineage.status, "resolved") + self.assertEqual(lineage.current_writer, "claude") + self.assertTrue(lineage.independent("codex")) + + def test_a_deployment_that_configured_no_branch_contract_keeps_its_answer(self): + lineage = self.compose(identity=UNCONFIGURED) + self.assertEqual(lineage.status, "resolved") + self.assertEqual(lineage.current_writer, "claude") + + def test_a_pull_request_with_no_contract_at_all_excludes_nobody(self): + lineage = self.compose(identity={"enabled": False}, labels=[], branch="fix/typo") + self.assertEqual(lineage.status, "resolved") + self.assertEqual(lineage.reason, "no_builder_identity") + + def test_the_branch_never_grants_takeover_authority(self): + """Only a verified episode moves the writer; the branch can only refuse.""" + + lineage = self.compose( + labels=["builder:codex"], + author="devin-ai-integration[bot]", + branch=BRANCH, + episodes=(takeover(),), + ) + self.assertEqual(lineage.status, "resolved") + self.assertEqual(lineage.current_writer, "codex") + self.assertEqual(lineage.contributors, ("devin", "codex")) + + def test_an_incomplete_target_refuses_instead_of_falling_back(self): + """A missing field is not an absent pull request. + + Downgrading to the identity-only answer skipped branch binding + altogether, so evidence recorded against another branch resolved as + evidence about this one. + """ + + for missing in ( + {"head_sha": ""}, + {"head_sha": TAKEN[:39]}, + {"repo": ""}, + {"repo": "no-slash"}, + {"pr_number": 0}, + {"pr_number": False}, + {"pr_number": "959"}, + {"branch": ""}, + {"branch": " "}, + ): + with self.subTest(**missing): + lineage = self.compose(**missing) + self.assertEqual(lineage.status, "conflict") + self.assertEqual(lineage.reason, "target_invalid") + self.assertTrue(lineage.owner_action) + + def test_an_incomplete_target_refuses_even_with_valid_evidence(self): + for missing in ({"branch": ""}, {"repo": ""}, {"head_sha": ""}): + overrides = dict( + labels=["builder:codex"], + author="devin-ai-integration[bot]", + branch=BRANCH, + episodes=(takeover(),), + ) + overrides.update(missing) + with self.subTest(**missing): + self.assertEqual(self.compose(**overrides).reason, "target_invalid") + + def test_evidence_must_bind_to_the_branch_under_decision(self): + """Branch binding is unconditional now that the branch is required.""" + + lineage = self.compose( + labels=["builder:codex"], + author="devin-ai-integration[bot]", + branch="codex/959-other", + episodes=(takeover(),), + ) + self.assertEqual(lineage.reason, "episode_unbound") + + def test_the_identity_only_route_is_selected_and_takes_no_evidence(self): + ordinary = resolve_configured_identity( + identity=IDENTITY, labels=["builder:claude"], author="a-human" + ) + self.assertEqual(ordinary.status, "resolved") + self.assertEqual(ordinary.current_writer, "claude") + + conflicting = resolve_configured_identity( + identity=IDENTITY, + labels=["builder:claude"], + author="a-human", + branch="codex/topic", + ) + self.assertEqual(conflicting.reason, "conflicting_builder_identity") + + self.assertNotIn( + "episodes", inspect.signature(resolve_configured_identity).parameters + ) + + def test_the_shared_target_contract_is_the_only_gate(self): + target = require_exact_target( + repo=REPO, pr_number=PR, branch=BRANCH, head_sha=TAKEN + ) + self.assertEqual( + (target.repo, target.pr_number, target.branch, target.head_sha), + (REPO, PR, BRANCH, TAKEN), + ) + for missing in ( + {"repo": ""}, + {"pr_number": 0}, + {"branch": ""}, + {"head_sha": ""}, + ): + kwargs = dict(repo=REPO, pr_number=PR, branch=BRANCH, head_sha=TAKEN) + kwargs.update(missing) + with self.subTest(**missing): + with self.assertRaises(LineageError): + require_exact_target(**kwargs) + + def test_the_target_number_must_already_be_a_number(self): + """A string that happens to parse is data nobody checked. + + Coercing here would make `"959"` a valid exact target. The coercive + parser belongs at the raw-payload boundary, not at the contract a + caller reaches holding data it claims to have verified. + """ + + for pr_number in ("959", " 959 ", "0959", 959.0, True, False, "abc", None): + with self.subTest(pr_number=pr_number): + with self.assertRaises(LineageError): + require_exact_target( + repo=REPO, pr_number=pr_number, branch=BRANCH, head_sha=TAKEN + ) + self.assertEqual( + self.compose( + pr_number=pr_number, + labels=["builder:codex"], + author="devin-ai-integration[bot]", + branch=BRANCH, + episodes=(takeover(),), + ).reason, + "target_invalid", + ) + target = require_exact_target( + repo=REPO, pr_number=PR, branch=BRANCH, head_sha=TAKEN + ) + self.assertIsInstance(target.pr_number, int) + self.assertEqual(target.pr_number, PR) + + def test_the_raw_payload_parser_still_coerces_its_own_input(self): + """Compatibility at the boundary that legitimately reads text.""" + + payload = takeover().as_dict() + payload["pr_number"] = str(PR) + self.assertEqual(episode_from_mapping(payload).pr_number, PR) + + +class ProjectionTests(unittest.TestCase): + """Bounded metadata only, and a label plan that never guesses.""" + + def test_the_projection_is_metadata_only(self): + payload = resolve().as_dict() + self.assertEqual( + set(payload), + { + "schema", + "status", + "reason", + "head_sha", + "contributors", + "current_writer", + "builder_label", + "stale_builder_labels", + "evidence", + "episodes", + "owner_action", + }, + ) + self.assertEqual(payload["contributors"], ["devin", "codex"]) + + def test_admission_is_closed_and_names_an_owner_action_on_refusal(self): + lineage = resolve() + for lane, admitted in (("devin", False), ("codex", False), ("claude", True)): + with self.subTest(lane=lane): + decision = lineage.admission(lane) + self.assertEqual(decision["admitted"], admitted) + self.assertEqual(decision["current_writer"], "codex") + if not admitted: + self.assertEqual(decision["reason"], "contributor_not_independent") + self.assertTrue(decision["owner_action"]) + self.assertEqual(lineage.admission("")["reason"], "reviewer_lane_invalid") + self.assertEqual( + resolve(head_sha=MOVED).admission("claude")["reason"], "lineage_waiting" + ) + + def test_a_resolved_takeover_plans_exactly_one_active_label(self): + plan = builder_label_plan( + resolve(), current_labels=["builder:devin", "tier:R"], identity=IDENTITY + ) + self.assertEqual(plan["status"], "reconcile") + self.assertEqual(plan["add"], ["builder:codex"]) + self.assertEqual(plan["remove"], ["builder:devin"]) + + def test_an_already_correct_label_set_plans_no_mutation(self): + plan = builder_label_plan( + resolve(), current_labels=["builder:codex"], identity=IDENTITY + ) + self.assertEqual(plan["status"], "current") + self.assertEqual((plan["add"], plan["remove"]), ([], [])) + + def test_unresolved_lineage_plans_no_mutation_at_all(self): + for lineage in (resolve(head_sha=MOVED), resolve(label_lanes=("claude",))): + with self.subTest(reason=lineage.reason): + plan = builder_label_plan( + lineage, current_labels=["builder:devin"], identity=IDENTITY + ) + self.assertEqual(plan["status"], "blocked") + self.assertEqual((plan["add"], plan["remove"]), ([], [])) + self.assertTrue(plan["owner_action"]) + + def test_the_active_label_follows_the_configured_mapping(self): + self.assertEqual(builder_label_for("codex", IDENTITY), "builder:codex") + self.assertEqual(builder_label_for("codex", None), "builder:codex") + self.assertEqual(builder_label_for("", IDENTITY), "") + + def test_the_record_key_is_stable_and_opaque(self): + key = pr_key(REPO, PR) + self.assertEqual(key, pr_key(REPO.upper(), str(PR))) + self.assertNotEqual(key, pr_key(REPO, PR + 1)) + self.assertEqual(len(key), 63) + self.assertNotIn("/", key) + + +class PurityTests(unittest.TestCase): + """The contract is a function of its arguments and nothing else.""" + + def test_resolution_is_deterministic_and_mutates_no_input(self): + episodes = [takeover().as_dict()] + labels = ["codex"] + first = resolve_lineage( + repo=REPO, + pr_number=PR, + branch=BRANCH, + head_sha=TAKEN, + episodes=episodes, + opener_lane="devin", + label_lanes=labels, + ) + second = resolve_lineage( + repo=REPO, + pr_number=PR, + branch=BRANCH, + head_sha=TAKEN, + episodes=episodes, + opener_lane="devin", + label_lanes=labels, + ) + self.assertEqual(first, second) + self.assertEqual(episodes, [takeover().as_dict()]) + self.assertEqual(labels, ["codex"]) + + def test_the_episode_and_lineage_records_are_immutable(self): + episode = takeover() + with self.assertRaises(dataclasses.FrozenInstanceError): + episode.resulting_head = MOVED # type: ignore[misc] + lineage = resolve() + with self.assertRaises(dataclasses.FrozenInstanceError): + lineage.current_writer = "claude" # type: ignore[misc] + self.assertIsInstance(episode, ContributionEpisode) + + +if __name__ == "__main__": # pragma: no cover + unittest.main() diff --git a/tests/test_builder_lineage_transport.py b/tests/test_builder_lineage_transport.py new file mode 100644 index 00000000..0a358c37 --- /dev/null +++ b/tests/test_builder_lineage_transport.py @@ -0,0 +1,600 @@ +"""Strict marker grammar and raw comment/page validation. + +Two opposite answers keep getting collapsed into one. "There is no published +lineage here" admits an independent reviewer on the ordinary single-builder +story; "this could not be read" has to stop. Every case below is about keeping +them apart -- in the marker grammar, in the raw comment record, in the page +array, and in the choice of which history to read at all. + +Trust is decided before parsing throughout, so an outsider can neither assert a +takeover into existence by posting a marker nor force a refusal by posting a +broken one. +""" + +from __future__ import annotations + +import json +import unittest + +from code_mower.builder_lineage import ( + LINEAGE_MARKER, + MAX_EPISODE_ARRIVALS, + MAX_EPISODES, + MAX_MARKER_BODY_CHARS, + NO_LINEAGE, + OMITTED, + LineageError, + bounded_arrivals, + comment_author_login, + comment_body, + episodes_from_comment_body, + flatten_comment_pages, + lineage_comment_marker, + lineage_context, + merge_episodes, + published_episodes, + require_comment_list, + require_comment_record, + resolve_lineage, + resolve_lineage_context, + select_comment_history, +) +from code_mower.lineage_identity import marker_author_trust + +from lineage_contract_fixtures import ( + AUTHORITY, + BRANCH, + IDENTITY, + INVALID_COMMENT_RESPONSES, + MALFORMED_COMMENT_RECORDS, + OUTSIDER, + PR, + REPO, + TAKEN, + VALID_COMMENT_RECORDS, + chain, + comment, + published, + takeover, + variant, +) + + +TRUST = marker_author_trust((AUTHORITY,)) + + +def valid_marker() -> str: + return lineage_comment_marker((takeover(),)) + + +class MarkerGrammarTests(unittest.TestCase): + """A broken marker is unreadable evidence, never absent evidence. + + The payload pattern matches only a complete, object-shaped, terminated + marker. Looking for evidence with it alone means an unterminated or + non-object marker is not read as broken -- it is not seen at all, and a + trusted comment announcing lineage reports none. + """ + + def unreadable(self, body: str) -> None: + with self.assertRaises(LineageError): + episodes_from_comment_body(body) + with self.assertRaises(LineageError): + published_episodes([comment(body=body)], trusted_author=TRUST) + + def test_a_well_formed_marker_round_trips_metadata_only(self): + marker = valid_marker() + self.assertNotIn("/Users", marker) + self.assertNotIn("session", marker) + self.assertEqual( + episodes_from_comment_body("context\n" + marker + "\nmore"), (takeover(),) + ) + + def test_an_unterminated_marker_is_unreadable_not_absent(self): + self.unreadable(valid_marker().replace("-->", "")) + + def test_a_non_object_payload_is_unreadable(self): + self.unreadable(f"") + + def test_malformed_json_is_unreadable(self): + self.unreadable(f'') + + def test_an_empty_payload_is_unreadable(self): + self.unreadable(f"") + + def test_two_markers_on_one_comment_are_ambiguous(self): + self.unreadable(f"{valid_marker()}\n\n{valid_marker()}") + + def test_a_valid_marker_beside_a_broken_one_is_still_ambiguous(self): + self.unreadable(f"{valid_marker()}\n\n") + + def test_a_marker_past_the_body_bound_is_unreadable_not_absent(self): + self.unreadable("x" * MAX_MARKER_BODY_CHARS + "\n" + valid_marker()) + + def test_an_announced_but_empty_chain_is_unreadable(self): + body = json.dumps({"schema": "code_mower.builderLineage.v1", "episodes": []}) + self.unreadable(f"") + + def test_a_chain_past_the_lineage_bound_is_unreadable(self): + payload = { + "schema": "code_mower.builderLineage.v1", + "episodes": [takeover().as_dict()] * (MAX_EPISODES + 1), + } + self.unreadable(f"") + + def test_an_unsupported_schema_is_unreadable(self): + body = json.dumps({"schema": "something.else", "episodes": []}) + self.unreadable(f"") + + def _duplicated(self, key: str, extra: str) -> str: + marker = valid_marker() + head, _, tail = marker.partition("{") + return f"{head}{{{json.dumps(key)}:{extra},{tail}" + + def test_a_duplicate_top_level_key_is_unreadable(self): + for key, extra in ( + ("schema", '"code_mower.builderLineage.v1"'), + ("schema", '"something.else"'), + ("episodes", "[]"), + ): + with self.subTest(key=key, extra=extra): + self.unreadable(self._duplicated(key, extra)) + + def test_a_duplicate_key_inside_an_episode_is_unreadable(self): + marker = valid_marker() + forged = marker.replace( + '"resulting_head"', '"resulting_head":"' + "c" * 40 + '","resulting_head"', 1 + ) + self.assertNotEqual(forged, marker) + self.unreadable(forged) + + def test_a_duplicate_nested_identity_key_is_unreadable(self): + marker = valid_marker() + forged = marker.replace( + '"destination_lane"', '"destination_lane":"claude","destination_lane"', 1 + ) + self.assertNotEqual(forged, marker) + self.unreadable(forged) + + def test_a_body_without_a_marker_yields_nothing(self): + self.assertEqual(episodes_from_comment_body("Codex took this over."), ()) + self.assertEqual(episodes_from_comment_body(""), ()) + + def test_a_genuinely_empty_comment_history_stays_ordinary(self): + self.assertEqual(published_episodes([], trusted_author=TRUST), ()) + self.assertEqual(require_comment_list([], what="history"), ()) + + +class MarkerTrustTests(unittest.TestCase): + """Trust is decided before parsing, and only from configured authorities.""" + + def test_an_untrusted_broken_marker_is_not_authoritative(self): + body = valid_marker().replace("-->", "") + untrusted = [comment(body=body, author=OUTSIDER)] + self.assertEqual(published_episodes(untrusted, trusted_author=TRUST), ()) + + def test_an_untrusted_duplicate_key_marker_is_not_authoritative(self): + marker = valid_marker() + head, _, tail = marker.partition("{") + ambiguous = f'{head}{{"episodes":[],{tail}' + self.assertEqual( + published_episodes( + [comment(body=ambiguous, author=OUTSIDER)], trusted_author=TRUST + ), + (), + ) + + def test_an_untrusted_valid_marker_asserts_nothing(self): + self.assertEqual( + published_episodes([published((takeover(),), author=OUTSIDER)], trusted_author=TRUST), + (), + ) + + def test_unrelated_trusted_comments_stay_non_authoritative(self): + self.assertEqual( + published_episodes( + [comment(body="Looks good to me. Shipping after CI.")], + trusted_author=TRUST, + ), + (), + ) + + def test_a_mixed_history_stops_on_the_broken_comment(self): + history = [ + published((takeover(),)), + comment(body=valid_marker().replace("-->", "")), + ] + with self.assertRaises(LineageError): + published_episodes(history, trusted_author=TRUST) + + def test_an_unconfigured_deployment_trusts_nobody(self): + nobody = marker_author_trust(()) + self.assertFalse(nobody(AUTHORITY)) + self.assertEqual( + published_episodes([published((takeover(),))], trusted_author=nobody), () + ) + + def test_authority_matching_ignores_case_and_a_leading_at_sign(self): + trusted = marker_author_trust(("@CodeMower-AI", " ")) + self.assertTrue(trusted("codemower-ai")) + self.assertTrue(trusted("@codemower-ai")) + self.assertFalse(trusted(OUTSIDER)) + + def test_the_gh_transport_author_field_is_trusted_the_same_way(self): + """A marker published through `gh --json comments` names `author`.""" + + history = [published((takeover(),), field="author")] + self.assertEqual(published_episodes(history, trusted_author=TRUST), (takeover(),)) + self.assertEqual(comment_author_login(history[0]), AUTHORITY) + + def test_the_full_cumulative_publication_is_read_whole(self): + links = chain(MAX_EPISODES) + history = [ + published(links[:length]) for length in range(1, len(links) + 1) + ] + arrivals = published_episodes(history, trusted_author=TRUST) + self.assertEqual(len(arrivals), MAX_EPISODES * (MAX_EPISODES + 1) // 2) + self.assertEqual(len(arrivals), 528) + + +class RawRecordValidationTests(unittest.TestCase): + """Present-and-unreadable is rejected before anything normalizes it.""" + + def test_a_malformed_present_field_is_rejected(self): + for record in MALFORMED_COMMENT_RECORDS: + with self.subTest(record=record): + with self.assertRaises(LineageError): + require_comment_record(record, what="history") + with self.assertRaises(LineageError): + require_comment_list([record], what="history") + with self.assertRaises(LineageError): + published_episodes([record], trusted_author=TRUST) + + def test_githubs_own_schema_keeps_working(self): + for record in VALID_COMMENT_RECORDS: + with self.subTest(record=record): + require_comment_record(record, what="history") + self.assertEqual( + len(require_comment_list(list(VALID_COMMENT_RECORDS), what="history")), + len(VALID_COMMENT_RECORDS), + ) + self.assertEqual( + published_episodes(list(VALID_COMMENT_RECORDS), trusted_author=TRUST), () + ) + + def test_a_successful_but_invalid_read_is_not_an_empty_history(self): + for response in INVALID_COMMENT_RESPONSES: + with self.subTest(response=response): + with self.assertRaises(LineageError): + require_comment_list(response, what="history") + + def test_a_deleted_account_names_no_author_and_no_marker(self): + self.assertEqual(comment_author_login({"user": None, "body": "hi"}), "") + self.assertEqual(comment_author_login({"author": None}), "") + self.assertEqual(comment_author_login({"user": {}}), "") + + def test_an_omitted_body_reads_as_empty_and_a_null_body_raises(self): + self.assertEqual(comment_body({"user": {"login": AUTHORITY}}), "") + with self.assertRaises(LineageError): + comment_body({"user": {"login": AUTHORITY}, "body": None}) + + +class SlurpedPageTests(unittest.TestCase): + """`gh api --paginate --slurp` gives arrays of arrays, and only that.""" + + def test_pages_of_comments_flatten_in_order(self): + pages = [[comment(body="one")], [], [comment(body="two")]] + flattened = flatten_comment_pages(pages) + self.assertEqual([item["body"] for item in flattened], ["one", "two"]) + + def test_a_genuinely_empty_page_set_stays_ordinary(self): + self.assertEqual(flatten_comment_pages([]), []) + self.assertEqual(flatten_comment_pages([[], []]), []) + + def test_a_response_that_is_not_a_page_array_is_refused(self): + for payload in (None, False, {}, {"comments": []}, "text", 7): + with self.subTest(payload=payload): + with self.assertRaises(LineageError): + flatten_comment_pages(payload) + + def test_an_object_wrapper_is_never_accepted_as_a_one_comment_page(self): + for page in ({}, {"comments": []}, comment(body="hi"), None, "text"): + with self.subTest(page=page): + with self.assertRaises(LineageError): + flatten_comment_pages([page]) + + def test_a_malformed_record_on_a_later_page_still_refuses(self): + first = [comment(body="ordinary") for _ in range(100)] + for record in MALFORMED_COMMENT_RECORDS: + with self.subTest(record=record): + with self.assertRaises(LineageError): + flatten_comment_pages([first, [record]]) + + def test_flattened_pages_are_copies_that_do_not_alias_the_input(self): + page = [comment(body="one")] + flattened = flatten_comment_pages([page]) + flattened[0]["body"] = "changed" + self.assertEqual(page[0]["body"], "one") + + +class SelectedHistoryTests(unittest.TestCase): + """An embedded comment list and a REST comment count are not the same thing.""" + + def test_a_rest_numeric_count_is_metadata_not_a_history(self): + for count in (0, 1, 42): + with self.subTest(count=count): + self.assertEqual(select_comment_history(embedded=count), ()) + + def test_a_supported_embedded_list_is_still_read(self): + history = [published((takeover(),), field="author")] + self.assertEqual(len(select_comment_history(embedded=history)), 1) + + def test_an_omitted_history_field_is_ordinary(self): + self.assertEqual(select_comment_history(), ()) + self.assertEqual(select_comment_history(embedded=OMITTED), ()) + + def test_an_explicit_selection_wins_over_whatever_sits_beside_it(self): + chosen = [published((takeover(),))] + self.assertEqual(len(select_comment_history(selected=chosen, embedded=7)), 1) + self.assertEqual(len(select_comment_history(selected=chosen, embedded=[])), 1) + + def test_an_explicitly_empty_selection_wins_over_an_embedded_takeover(self): + self.assertEqual( + select_comment_history(selected=[], embedded=[published((takeover(),))]), () + ) + + def test_a_present_but_malformed_history_fails_closed(self): + for embedded in ( + None, + False, + {}, + "comments", + [[comment(body="hi")]], + [comment(body="hi"), "text"], + [{"user": {"login": 7}, "body": "hi"}], + [{"user": {"login": AUTHORITY}, "body": None}], + ): + with self.subTest(embedded=embedded): + with self.assertRaises(LineageError): + select_comment_history(embedded=embedded) + + def test_a_malformed_selection_fails_closed_beside_a_valid_count(self): + with self.assertRaises(LineageError): + select_comment_history( + selected=[{"user": {"login": 7}, "body": "hi"}], embedded=3 + ) + + def test_a_boolean_is_never_a_comment_count(self): + for value in (True, False): + with self.subTest(value=value): + with self.assertRaises(LineageError): + select_comment_history(embedded=value) + + +class BoundedArrivalTests(unittest.TestCase): + """The raw budget is enforced while the input is walked, not afterwards.""" + + def _counting(self, items): + seen = [] + + def walk(): + for item in items: + seen.append(item) + yield item + + return walk(), seen + + def test_exactly_the_documented_budget_is_accepted(self): + links = chain(MAX_EPISODES) + cumulative = [ + episode for length in range(1, len(links) + 1) for episode in links[:length] + ] + arrivals = cumulative + list(links) + self.assertEqual(len(arrivals), MAX_EPISODE_ARRIVALS) + self.assertEqual(len(merge_episodes(cumulative, links)), MAX_EPISODE_ARRIVALS) + self.assertEqual(len(tuple(bounded_arrivals(arrivals))), MAX_EPISODE_ARRIVALS) + + def test_one_arrival_past_the_budget_refuses_even_when_it_repeats(self): + links = chain(MAX_EPISODES) + cumulative = [ + episode for length in range(1, len(links) + 1) for episode in links[:length] + ] + with self.assertRaises(LineageError): + merge_episodes(cumulative + list(links), (links[-1],)) + + def test_a_lazy_source_is_not_consumed_past_the_first_refused_arrival(self): + links = chain(2) + oversized = [links[0]] * (MAX_EPISODE_ARRIVALS + 50) + walked, seen = self._counting(oversized) + with self.assertRaises(LineageError): + tuple(bounded_arrivals(walked)) + self.assertEqual(len(seen), MAX_EPISODE_ARRIVALS + 1) + + def test_independently_collapsed_inputs_cannot_reset_the_cap(self): + """Collectors keep raw arrivals; only the resolver collapses them.""" + + links = chain(4) + merged = merge_episodes(links, links) + self.assertEqual(len(merged), 8, "repeats are preserved as raw arrivals") + resolved = resolve_lineage( + repo=REPO, + pr_number=PR, + branch=BRANCH, + head_sha=links[-1].resulting_head, + episodes=merged, + ) + self.assertEqual(resolved.status, "resolved") + self.assertEqual(resolved.episodes, 4) + + def test_a_malformed_member_is_the_contract_error_not_an_attribute_failure(self): + for item in ("text", None, 7, {"schema": "nope"}, [takeover()]): + with self.subTest(item=item): + with self.assertRaises(LineageError): + merge_episodes((), (item,)) + + def test_a_contradicting_episode_survives_to_be_refused(self): + links = chain(3) + forged = variant(links[1], destination_lane="claude", source_lane="claude") + merged = merge_episodes(links, (forged,)) + self.assertEqual(len(merged), 4) + self.assertEqual( + resolve_lineage( + repo=REPO, + pr_number=PR, + branch=BRANCH, + head_sha=links[-1].resulting_head, + episodes=merged, + ).reason, + "episode_duplicated", + ) + + def test_a_published_collection_is_bounded_too(self): + links = chain(MAX_EPISODES) + history = [published(links) for _ in range(MAX_EPISODES)] + with self.assertRaises(LineageError): + published_episodes(history, trusted_author=TRUST) + + +class RendererTests(unittest.TestCase): + """Rendering is not a truncation operation.""" + + def test_one_and_thirty_two_episodes_round_trip_unchanged(self): + for length in (1, MAX_EPISODES): + with self.subTest(length=length): + links = chain(length) + parsed = episodes_from_comment_body(lineage_comment_marker(links)) + self.assertEqual(parsed, links) + + def test_an_empty_chain_refuses_rather_than_publishing_nothing(self): + with self.assertRaises(LineageError): + lineage_comment_marker(()) + + def test_a_thirty_third_entry_cannot_disappear(self): + links = chain(MAX_EPISODES) + contradictory = variant( + links[-1], destination_lane="claude", source_lane="claude" + ) + with self.assertRaises(LineageError): + lineage_comment_marker(links + (contradictory,)) + + def test_repeats_beyond_the_bound_render_without_losing_an_episode(self): + """Slicing to the bound silently dropped the newest episodes.""" + + links = chain(MAX_EPISODES) + marker = lineage_comment_marker(links + links[:2]) + self.assertEqual(episodes_from_comment_body(marker), links) + + def test_malformed_unchained_and_conflicting_input_refuse(self): + links = chain(3) + for episodes in ( + ("not an episode",), + ({"schema": "nope"},), + (links[1],), + links[:1] + links[2:], + links + (variant(links[1], destination_lane="claude", source_lane="claude"),), + ): + with self.subTest(episodes=episodes): + with self.assertRaises(LineageError): + lineage_comment_marker(episodes) + + def test_a_refused_chain_produces_no_marker_text(self): + with self.assertRaises(LineageError) as raised: + lineage_comment_marker(()) + self.assertNotIn(LINEAGE_MARKER, str(raised.exception)) + + +class EvidenceAssemblyTests(unittest.TestCase): + """Carried evidence binds to a complete target or refuses.""" + + def test_a_partially_populated_target_refuses_rather_than_reading_as_absent(self): + for missing in ( + {"head_sha": ""}, + {"repo": ""}, + {"pr_number": 0}, + {"branch": ""}, + {"head_sha": TAKEN[:39]}, + ): + kwargs = dict(repo=REPO, pr_number=PR, branch=BRANCH, head_sha=TAKEN) + kwargs.update(missing) + with self.subTest(**missing): + with self.assertRaises(LineageError): + lineage_context(**kwargs) + + def test_a_deliberately_absent_target_is_the_ordinary_no_context_case(self): + self.assertIs(lineage_context(repo="", pr_number=0), NO_LINEAGE) + self.assertIs( + lineage_context(repo="", pr_number=0, comments=[], trusted_author=TRUST), + NO_LINEAGE, + ) + + def test_evidence_with_no_target_to_bind_it_to_refuses(self): + with self.assertRaises(LineageError): + lineage_context( + repo="", + pr_number=0, + comments=[published((takeover(),))], + trusted_author=TRUST, + ) + + def test_a_context_carrying_episodes_is_never_treated_as_absent(self): + carried = lineage_context( + repo=REPO, + pr_number=PR, + branch=BRANCH, + head_sha=TAKEN, + comments=[published((takeover(),))], + trusted_author=TRUST, + ) + self.assertNotEqual(carried, NO_LINEAGE) + resolved = resolve_lineage_context( + carried, + identity=IDENTITY, + labels=["builder:codex"], + author="devin-ai-integration[bot]", + ) + self.assertEqual(resolved.current_writer, "codex") + + def test_a_context_carries_only_trusted_published_evidence(self): + context = lineage_context( + repo=REPO, + pr_number=PR, + branch=BRANCH, + head_sha=TAKEN, + comments=[ + comment(body="unrelated"), + published((takeover(),)), + published((takeover(),), author=OUTSIDER), + ], + trusted_author=TRUST, + ) + self.assertEqual(context.episodes, (takeover(),)) + resolved = resolve_lineage_context( + context, + identity=IDENTITY, + labels=["builder:codex"], + author="devin-ai-integration[bot]", + ) + self.assertEqual(resolved.status, "resolved") + self.assertEqual(resolved.current_writer, "codex") + + def test_an_unreadable_published_history_propagates_rather_than_emptying(self): + with self.assertRaises(LineageError): + lineage_context( + repo=REPO, + pr_number=PR, + branch=BRANCH, + head_sha=TAKEN, + comments=[comment(body=valid_marker().replace("-->", ""))], + trusted_author=TRUST, + ) + + def test_no_evidence_still_reaches_the_one_shared_decision(self): + resolved = resolve_lineage_context( + None, identity=IDENTITY, labels=["builder:claude"], author="a-human" + ) + self.assertEqual(resolved.status, "resolved") + self.assertEqual(resolved.current_writer, "claude") + + +if __name__ == "__main__": # pragma: no cover + unittest.main() diff --git a/tests/test_lineage_contract_packaging.py b/tests/test_lineage_contract_packaging.py new file mode 100644 index 00000000..70d8006e --- /dev/null +++ b/tests/test_lineage_contract_packaging.py @@ -0,0 +1,513 @@ +"""The lineage contract stays pure, stays mirrored, and stays packaged. + +Three properties that are invisible to the behavioural tests and have each +already broken something: + +* **Purity.** The same decision has to be computable by the package, by the + vendored ``tools/`` copy inside a generated product repository, and by a + reviewer host with no private state. One ``os.environ`` read or one store + import would make that false without failing a single behaviour case. +* **Mirror parity.** CI lints and a generated product gate import the vendored + file, not the package one, so drift there is invisible to every assertion + that reaches into ``src/code_mower``. +* **Materialization.** A helper whose dependency is not in the generated + support list fails at import time in the product repository, where nothing + here would ever see it. + +These are checked against the parsed module and its actual imported behaviour, +never against its source text. +""" + +from __future__ import annotations + +import ast +import importlib.util +import json +import os +import subprocess +import sys +import tempfile +import unittest +from pathlib import Path + +from code_mower import builder_lineage, init, lineage_identity, package_manifest + +from lineage_contract_fixtures import BRANCH, PR, REPO, TAKEN, chain, takeover + + +ROOT = Path(__file__).resolve().parents[1] + +#: Modules whose decisions must be a function of their arguments alone. +PURE_MODULES = ("builder_lineage.py", "lineage_identity.py") + +#: Standard-library modules a pure decision may use. Everything here is +#: computation; nothing here can reach the environment, a disk or a socket. +ALLOWED_STDLIB = { + "__future__", + "dataclasses", + "hashlib", + "json", + "re", + "typing", +} + +#: Reaching any of these from a pure module is the failure, whether it arrives +#: as an import, an attribute or a builtin call. +FORBIDDEN_MODULES = { + "os", + "os.path", + "pathlib", + "subprocess", + "socket", + "shutil", + "tempfile", + "urllib", + "urllib.request", + "http", + "requests", +} + +#: Package modules a pure module may not depend on: adapters, transports, +#: stores and consumers. A pure import of one of these is how the contract +#: acquires an environment read it does not declare. +FORBIDDEN_PACKAGE_MODULES = { + "audit_labeler_lib", + "board", + "context_store", + "controller", + "decisions", + "devin_api", + "lane_delivery", + "lane_handoff", + "lane_status", + "provider_runners", +} + +FORBIDDEN_CALLS = {"open", "input", "exec", "eval", "__import__", "compile"} + + +def _module_path(name: str) -> Path: + return ROOT / "src" / "code_mower" / name + + +def _imported_names(tree: ast.AST) -> set[str]: + names: set[str] = set() + for node in ast.walk(tree): + if isinstance(node, ast.Import): + names.update(alias.name for alias in node.names) + elif isinstance(node, ast.ImportFrom): + if node.level: + names.add("." * node.level + (node.module or "")) + elif node.module: + names.add(node.module) + return names + + +class PurityTests(unittest.TestCase): + def trees(self): + for name in PURE_MODULES: + yield name, ast.parse(_module_path(name).read_text(encoding="utf-8")) + + def test_pure_modules_import_nothing_that_can_reach_the_outside(self): + for name, tree in self.trees(): + with self.subTest(module=name): + for imported in _imported_names(tree): + self.assertNotIn(imported, FORBIDDEN_MODULES, imported) + root = imported.split(".")[0] + self.assertNotIn(root, FORBIDDEN_MODULES, imported) + if imported.startswith("."): + relative = imported.lstrip(".") + self.assertNotIn(relative, FORBIDDEN_PACKAGE_MODULES, imported) + elif root: + self.assertIn(root, ALLOWED_STDLIB | {"code_mower"}, imported) + + def test_pure_modules_make_no_io_or_dynamic_execution_calls(self): + for name, tree in self.trees(): + with self.subTest(module=name): + called = { + node.func.id + for node in ast.walk(tree) + if isinstance(node, ast.Call) and isinstance(node.func, ast.Name) + } + self.assertEqual(called & FORBIDDEN_CALLS, set()) + + def test_pure_modules_contain_no_import_inside_a_function(self): + """A deferred import is how a store dependency hides from the header.""" + + for name, tree in self.trees(): + with self.subTest(module=name): + for node in ast.walk(tree): + if not isinstance(node, (ast.FunctionDef, ast.AsyncFunctionDef)): + continue + for inner in ast.walk(node): + self.assertNotIsInstance(inner, (ast.Import, ast.ImportFrom)) + + def test_the_pure_contract_module_depends_on_no_package_module_at_all(self): + tree = ast.parse(_module_path("builder_lineage.py").read_text(encoding="utf-8")) + relative = {name for name in _imported_names(tree) if name.startswith(".")} + self.assertEqual(relative, set()) + + def test_the_identity_module_depends_only_on_the_contract_module(self): + tree = ast.parse(_module_path("lineage_identity.py").read_text(encoding="utf-8")) + relative = {name for name in _imported_names(tree) if name.startswith(".")} + self.assertEqual(relative, {".builder_lineage"}) + + def test_resolution_accumulates_nothing_between_calls(self): + """Module-level tables are constants, not state that grows as it runs.""" + + def snapshot(): + return { + (module.__name__, attribute): repr(getattr(module, attribute)) + for module in (builder_lineage, lineage_identity) + for attribute in dir(module) + if not attribute.startswith("__") + and isinstance(getattr(module, attribute), (list, dict, set)) + } + + before = snapshot() + self.assertTrue(before, "expected module-level tables to compare") + links = chain(4) + for _ in range(3): + builder_lineage.resolve_lineage( + repo=REPO, + pr_number=PR, + branch=BRANCH, + head_sha=links[-1].resulting_head, + episodes=links, + opener_lane="devin", + label_lanes=("codex",), + ) + lineage_identity.identity_with_lane_floor({"enabled": True}, "codex") + self.assertEqual(snapshot(), before) + + +class MirrorTests(unittest.TestCase): + def test_the_vendored_copy_is_byte_identical_to_the_canonical_module(self): + self.assertEqual( + (ROOT / "tools" / "builder_lineage.py").read_bytes(), + (ROOT / "src" / "code_mower" / "builder_lineage.py").read_bytes(), + ) + + def _vendored(self): + """Import the vendored copy standalone, the way a product gate does.""" + + name = "vendored_builder_lineage" + path = ROOT / "tools" / "builder_lineage.py" + spec = importlib.util.spec_from_file_location(name, path) + assert spec and spec.loader + module = importlib.util.module_from_spec(spec) + # `@dataclass` resolves the defining module through `sys.modules` while + # the class body is still executing, so a module executed from a spec + # without being registered first fails on its first dataclass -- before + # any of the behaviour below has a chance to run. Registering it is what + # a real import does; restoring the previous entry afterwards keeps the + # rest of the suite from seeing a second copy of the contract. + previous = sys.modules.get(name) + sys.modules[name] = module + if previous is None: + self.addCleanup(sys.modules.pop, name, None) + else: + self.addCleanup(sys.modules.__setitem__, name, previous) + spec.loader.exec_module(module) + return module + + def test_the_vendored_copy_is_registered_and_then_cleaned_up(self): + """The import seam itself, so a silent regression cannot hide the rest.""" + + name = "vendored_builder_lineage" + self.assertNotIn(name, sys.modules) + module = self._vendored() + self.assertIs(sys.modules[name], module) + self.assertTrue(module.ContributionEpisode.__dataclass_fields__) + + def test_the_vendored_copy_imports_with_no_package_on_the_path(self): + vendored = self._vendored() + self.assertEqual(vendored.MAX_EPISODES, builder_lineage.MAX_EPISODES) + self.assertEqual( + vendored.MAX_EPISODE_ARRIVALS, builder_lineage.MAX_EPISODE_ARRIVALS + ) + + def test_the_vendored_copy_reaches_the_same_decision(self): + vendored = self._vendored() + links = chain(8) + cumulative = [ + episode.as_dict() + for length in range(1, len(links) + 1) + for episode in links[:length] + ] + kwargs = dict( + repo=REPO, + pr_number=PR, + branch=BRANCH, + head_sha=links[-1].resulting_head, + episodes=cumulative, + opener_lane="devin", + label_lanes=("codex",), + ) + mine = builder_lineage.resolve_lineage(**kwargs) + theirs = vendored.resolve_lineage(**kwargs) + self.assertEqual(mine.as_dict(), theirs.as_dict()) + self.assertEqual(mine.status, "resolved") + self.assertEqual(mine.current_writer, "codex") + + def test_the_vendored_copy_refuses_the_same_broken_marker(self): + vendored = self._vendored() + broken = builder_lineage.lineage_comment_marker((takeover(),)).replace("-->", "") + for module in (builder_lineage, vendored): + with self.subTest(module=module.__name__): + with self.assertRaises(module.LineageError): + module.episodes_from_comment_body(broken) + + def test_the_vendored_copy_parses_a_valid_marker_the_same_way(self): + vendored = self._vendored() + marker = builder_lineage.lineage_comment_marker((takeover(),)) + self.assertEqual( + [episode.as_dict() for episode in vendored.episodes_from_comment_body(marker)], + [takeover().as_dict()], + ) + + +class MaterializationTests(unittest.TestCase): + def test_the_contract_travels_with_the_generated_gate_helper(self): + targets = {target for target, _, _, _ in init.PRODUCT_SUPPORT_FILES} + self.assertIn("tools/audit_labeler_lib.py", targets) + self.assertIn("tools/builder_lineage.py", targets) + self.assertIn("tools/decisions.py", targets) + + def test_every_generated_helper_is_copied_from_a_real_package_module(self): + for target, source, _kind, _mode in init.PRODUCT_SUPPORT_FILES: + if not (target.startswith("tools/") and target.endswith(".py")): + continue + if "/" in source: # templated wrappers, not package modules + continue + with self.subTest(target=target): + self.assertTrue( + (ROOT / "src" / "code_mower" / source).is_file(), + f"{target} copies a package module that does not exist", + ) + + def test_both_modules_are_in_the_package_inventory(self): + by_target = { + target: source for source, target, _ in package_manifest.PACKAGE_FILES + } + self.assertEqual( + by_target["src/code_mower/builder_lineage.py"], "tools/builder_lineage.py" + ) + self.assertEqual( + by_target["src/code_mower/lineage_identity.py"], + "src/code_mower/lineage_identity.py", + ) + + def test_every_packaged_source_for_the_contract_exists(self): + for source, target, _kind in package_manifest.PACKAGE_FILES: + if "lineage" not in target: + continue + with self.subTest(target=target): + self.assertTrue((ROOT / source).is_file(), source) + + +STANDALONE_PROBE = ''' +import json, os, sys + +# Everything this repository could lend the probe is removed before the first +# import, so the materialized tree has to stand on its own. +repo = os.path.abspath(sys.argv[1]) +sys.path[:] = [ + entry for entry in sys.path + if entry and not os.path.abspath(entry).startswith(repo) +] +sys.path.insert(0, os.getcwd()) +try: + import code_mower + package_origin = getattr(code_mower, "__file__", "") or "" +except Exception: + package_origin = "" + +import tools.builder_lineage as lineage +import tools.audit_labeler_lib as labeler + +episode = lineage.ContributionEpisode( + sequence=1, repo="acme/widget", pr_number=959, branch="devin/topic", + source_lane="devin", destination_lane="codex", + expected_head="a" * 40, resulting_head="b" * 40, writer_state="terminated", +) +marker = lineage.lineage_comment_marker((episode,)) +parsed = lineage.episodes_from_comment_body("context\\n" + marker) +resolved = lineage.resolve_lineage( + repo="acme/widget", pr_number=959, branch="devin/topic", head_sha="b" * 40, + episodes=parsed, opener_lane="devin", label_lanes=("codex",), +) +try: + lineage.lineage_comment_marker(()) + refused = False +except lineage.LineageError: + refused = True +print(json.dumps({ + "lineage_origin": lineage.__file__, + "labeler_origin": labeler.__file__, + "parsed": len(parsed), + "status": resolved.status, + "writer": resolved.current_writer, + "contributors": list(resolved.contributors), + "unbound": lineage.resolve_lineage( + repo="acme/widget", pr_number=959, branch="", head_sha="b" * 40, + episodes=parsed, + ).reason, + "empty_render_refused": refused, + "matches": labeler.builder_identity_matches( + labels=["builder:codex"], author="codex[bot]", text="", + config={"enabled": True, "labels": {"builder:codex": "codex"}, + "authors": {"codex[bot]": "codex"}}, + ), + "package_origin": package_origin, +})) +''' + + +class MaterializedStandaloneTests(unittest.TestCase): + """The real generated product tree, with no Code Mower package anywhere. + + A product repository's gate runs the *materialized* `tools/` copies with no + package installed. Asserting mirror equality proves the bytes match; it does + not prove the materialized tree imports and decides. This runs init for + real, then a clean subprocess that can only see what init wrote. + """ + + def test_the_materialized_tools_tree_decides_on_its_own(self): + from code_mower import config as code_mower_config + + plan = init.render_init_plan( + code_mower_config.load_config( + ROOT / "src/code_mower/templates/code-mower.example.yml" + ), + package_mode=True, + package_command="code-mower", + ) + with tempfile.TemporaryDirectory() as tmp: + product = Path(tmp) / "product" + init.apply_init_plan(plan, product / ".code-mower.generated") + generated_tools = product / ".code-mower.generated" / "tools" + self.assertTrue((generated_tools / "builder_lineage.py").is_file()) + + # Exactly what a generated gate step sees: the materialized tools + # package, and nothing of this repository. + (generated_tools / "__init__.py").write_text("", encoding="utf-8") + root = generated_tools.parent + (root / "probe.py").write_text(STANDALONE_PROBE, encoding="utf-8") + environment = { + key: value + for key, value in os.environ.items() + if key not in {"PYTHONPATH", "PYTHONHOME"} + and not key.startswith("CODE_MOWER_") + } + result = subprocess.run( + [sys.executable, "-E", "probe.py", str(ROOT)], + cwd=root, + capture_output=True, + text=True, + timeout=180, + env=environment, + ) + self.assertEqual(result.returncode, 0, result.stderr) + payload = json.loads(result.stdout.strip().splitlines()[-1]) + + self.assertEqual( + Path(payload["lineage_origin"]).resolve(), + (generated_tools / "builder_lineage.py").resolve(), + ) + self.assertEqual( + Path(payload["labeler_origin"]).resolve(), + (generated_tools / "audit_labeler_lib.py").resolve(), + ) + self.assertFalse( + payload["package_origin"].startswith(str(ROOT)), + "the probe reached this repository's own package", + ) + self.assertEqual(payload["parsed"], 1) + self.assertEqual(payload["status"], "resolved") + self.assertEqual(payload["writer"], "codex") + self.assertEqual(payload["contributors"], ["devin", "codex"]) + self.assertEqual(payload["unbound"], "target_invalid") + self.assertTrue(payload["empty_render_refused"]) + self.assertEqual(payload["matches"], ["codex"]) + + +class ContractSurfaceTests(unittest.TestCase): + """The names later stages are being told to consume actually exist.""" + + def test_the_contract_module_exports_the_stage_one_surface(self): + for name in ( + "ContributionEpisode", + "Lineage", + "LineageContext", + "LineageError", + "ExactTarget", + "IdentityConflictError", + "bounded_arrivals", + "branch_lane_from_identity", + "builder_label_plan", + "canonical_identity", + "comment_author_login", + "continuation_episode", + "episode_from_handoff", + "episode_from_mapping", + "episodes_from_comment_body", + "flatten_comment_pages", + "lanes_from_identity", + "lineage_comment_marker", + "lineage_context", + "merge_episodes", + "pr_key", + "published_episodes", + "require_comment_list", + "require_comment_record", + "require_episode", + "require_episode_chain", + "require_exact_target", + "resolve_builder_lineage", + "resolve_configured_identity", + "resolve_identity_only", + "resolve_lineage", + "resolve_lineage_context", + "select_comment_history", + ): + with self.subTest(name=name): + self.assertTrue(hasattr(builder_lineage, name), name) + + def test_the_identity_module_exports_the_stage_one_surface(self): + for name in ( + "LANE_ACCOUNT_FLOOR", + "ReviewerIdentityInvalid", + "ReviewerNotIndependent", + "combine_evidence", + "identity_from_json", + "identity_with_lane_floor", + "marker_author_trust", + "normalized_account_map", + "pr_lineage", + "require_independent_reviewer", + "require_reviewer_lane", + "reviewer_admission", + "trusted_published_episodes", + ): + with self.subTest(name=name): + self.assertTrue(hasattr(lineage_identity, name), name) + + def test_the_published_projection_carries_no_private_metadata(self): + payload = builder_lineage.resolve_lineage( + repo=REPO, + pr_number=PR, + branch=BRANCH, + head_sha=TAKEN, + episodes=(takeover(),), + opener_lane="devin", + label_lanes=("codex",), + ).as_dict() + rendered = repr(payload) + builder_lineage.lineage_comment_marker((takeover(),)) + for private in ("/Users", "/tmp", "session", "token", "prompt", "transcript"): + with self.subTest(private=private): + self.assertNotIn(private, rendered) + + +if __name__ == "__main__": # pragma: no cover + unittest.main() diff --git a/tests/test_lineage_identity_contract.py b/tests/test_lineage_identity_contract.py new file mode 100644 index 00000000..f4cea1e9 --- /dev/null +++ b/tests/test_lineage_identity_contract.py @@ -0,0 +1,444 @@ +"""Canonical reviewer identity, alias normalization and pure admission. + +Reviewer independence is decided by naming lanes. A deployment whose contract +cannot name the reviewer's own lane names no contributor for it either, and the +seam then admits exactly the reviewer it exists to exclude. These cases pin the +floor that prevents that, the alias rules that keep it order-independent, and +the admission decision built on top -- all from explicit inputs, with no +environment read anywhere. +""" + +from __future__ import annotations + +import json +import unittest + +from code_mower.builder_lineage import LineageError +from code_mower.lineage_identity import ( + AUTHOR_EXCLUSION_ENV, + LANE_ACCOUNT_FLOOR, + ReviewerIdentityInvalid, + ReviewerNotIndependent, + account_key, + combine_evidence, + identity_from_json, + identity_with_lane_floor, + load_identity, + normalized_account_map, + pr_lineage, + require_independent_reviewer, + require_reviewer_lane, + reviewer_admission, + trusted_published_episodes, +) + +from lineage_contract_fixtures import ( + AUTHORITY, + IDENTITY, + MOVED, + OUTSIDER, + PR, + REPO, + TAKEN, + UNCONFIGURED, + chain, + comment, + pr_meta, + published, + takeover, +) + + +class IdentityLoadingTests(unittest.TestCase): + """Loading is a pure parse of text the caller supplies.""" + + def test_a_valid_contract_parses(self): + self.assertEqual(identity_from_json(json.dumps(IDENTITY)), IDENTITY) + + def test_a_missing_or_unusable_contract_disables_lane_naming(self): + for raw in (None, "", " not json", "[1, 2]", '"text"', "7"): + with self.subTest(raw=raw): + self.assertEqual(identity_from_json(raw), {"enabled": False}) + + def test_the_compatibility_spelling_is_the_same_function(self): + self.assertIs(load_identity, identity_from_json) + + def test_the_contract_variable_is_named_but_never_read_here(self): + self.assertEqual(AUTHOR_EXCLUSION_ENV, "CODE_MOWER_AUTHOR_EXCLUSION_JSON") + + +class OwnLaneFloorTests(unittest.TestCase): + """A floor, not a default: the reviewer's own lane is always nameable.""" + + MINIMAL = { + "enabled": True, + "labels": {"builder:codex": ""}, + "authors": {"codex[bot]": ""}, + } + + def test_a_blank_own_label_and_account_are_overwritten(self): + floored = identity_with_lane_floor(self.MINIMAL, "codex") + self.assertEqual(floored["labels"]["builder:codex"], "codex") + self.assertEqual(floored["authors"]["codex[bot]"], "codex") + + def test_an_invalid_own_mapping_is_overwritten_not_preserved(self): + for invalid in (None, 0, [], {}, " "): + with self.subTest(invalid=invalid): + floored = identity_with_lane_floor( + { + "enabled": True, + "labels": {"builder:codex": invalid}, + "authors": {"codex[bot]": invalid}, + }, + "codex", + ) + self.assertEqual(floored["labels"]["builder:codex"], "codex") + self.assertEqual(floored["authors"]["codex[bot]"], "codex") + + def test_a_missing_or_disabled_contract_still_names_the_reviewer(self): + for identity in (None, {}, {"enabled": False, "labels": {}, "authors": {}}): + with self.subTest(identity=identity): + floored = identity_with_lane_floor(identity, "codex") + self.assertTrue(floored["enabled"]) + self.assertEqual(floored["labels"]["builder:codex"], "codex") + for login in LANE_ACCOUNT_FLOOR["codex"]: + self.assertEqual(floored["authors"][login], "codex") + + def test_only_the_reviewers_own_lane_is_synthesized(self): + floored = identity_with_lane_floor(self.MINIMAL, "claude") + self.assertEqual(floored["labels"]["builder:claude"], "claude") + self.assertNotIn("builder:devin", floored["labels"]) + # The other lane's useless entry is left exactly as configured. + self.assertEqual(floored["labels"]["builder:codex"], "") + + def test_a_conflicting_own_label_or_account_refuses(self): + for identity in ( + {"enabled": True, "labels": {"builder:codex": "claude"}, "authors": {}}, + { + "enabled": True, + "labels": {"builder:codex": "codex"}, + "authors": {"codex[bot]": "devin"}, + }, + # Disabling the contract does not make a misnamed own lane safe. + {"enabled": False, "labels": {"builder:codex": "claude"}, "authors": {}}, + ): + with self.subTest(identity=identity): + with self.assertRaises(ReviewerIdentityInvalid) as raised: + identity_with_lane_floor(identity, "codex") + self.assertIn("reviewer_identity_invalid", str(raised.exception)) + + def test_the_rest_of_the_configured_contract_survives_the_floor(self): + floored = identity_with_lane_floor(IDENTITY, "codex") + self.assertEqual(floored["branch_prefixes"], IDENTITY["branch_prefixes"]) + self.assertTrue(floored["require_verified_lineage"]) + self.assertEqual(floored["labels"]["builder:claude"], "claude") + + def test_flooring_leaves_the_supplied_contract_untouched(self): + original = json.loads(json.dumps(self.MINIMAL)) + identity_with_lane_floor(original, "codex") + self.assertEqual(original, self.MINIMAL) + + +class AccountAliasTests(unittest.TestCase): + """Account names match case-insensitively, so aliases are one account.""" + + def _aliased(self, *pairs): + contract = dict(IDENTITY) + contract["authors"] = dict(pairs) + return contract + + def test_the_key_is_normalized_the_way_resolution_reads_it(self): + self.assertEqual(account_key(" Codex[Bot] "), "codex[bot]") + self.assertEqual(account_key(None), "") + self.assertEqual( + normalized_account_map({" Codex[Bot] ": "codex"}), {"codex[bot]": "codex"} + ) + self.assertEqual(normalized_account_map({" ": "codex"}), {}) + self.assertEqual(normalized_account_map("not a mapping"), {}) + + def test_a_conflicting_alias_refuses_in_either_insertion_order(self): + for pairs in ( + (("Codex[Bot]", "claude"), ("codex[bot]", "codex")), + (("codex[bot]", "codex"), ("Codex[Bot]", "claude")), + ((" codex[bot] ", "claude"), ("codex[bot]", "codex")), + (("CODEX[BOT]", "devin"), ("codex[bot]", "codex")), + ): + with self.subTest(order=pairs): + with self.assertRaises(ReviewerIdentityInvalid): + identity_with_lane_floor(self._aliased(*pairs), "codex") + + def test_a_compatible_alias_is_accepted(self): + for pairs in ( + (("Codex[Bot]", "codex"), ("codex[bot]", "codex")), + (("codex[bot]", "codex"), ("CODEX[BOT]", "Codex")), + ((" codex[bot] ", "codex"),), + ): + with self.subTest(order=pairs): + floored = identity_with_lane_floor(self._aliased(*pairs), "codex") + self.assertEqual(floored["authors"]["codex[bot]"], "codex") + self.assertEqual( + floored["branch_prefixes"], IDENTITY["branch_prefixes"] + ) + + def test_an_alias_cannot_outrank_the_canonical_account(self): + floored = identity_with_lane_floor( + self._aliased( + ("Codex[Bot]", "codex"), ("devin-ai-integration[bot]", "devin") + ), + "codex", + ) + self.assertEqual(floored["authors"]["codex[bot]"], "codex") + self.assertEqual(floored["labels"]["builder:codex"], "codex") + self.assertEqual(floored["authors"]["devin-ai-integration[bot]"], "devin") + + +class PublishedEvidenceTests(unittest.TestCase): + """Raw validation first, then trust, then parsing.""" + + def test_an_unreadable_read_is_refused_before_trust_is_considered(self): + for response in (None, False, {}, [comment(body="hi"), "text"]): + with self.subTest(response=response): + with self.assertRaises(LineageError): + trusted_published_episodes(response, authorities=(AUTHORITY,)) + with self.assertRaises(LineageError): + trusted_published_episodes(response, authorities=()) + + def test_with_no_authorities_configured_nothing_is_read(self): + history = [published((takeover(),))] + self.assertEqual(trusted_published_episodes(history, authorities=()), ()) + + def test_a_trusted_marker_is_read_and_an_untrusted_one_is_not(self): + self.assertEqual( + trusted_published_episodes( + [published((takeover(),))], authorities=(AUTHORITY,) + ), + (takeover(),), + ) + self.assertEqual( + trusted_published_episodes( + [published((takeover(),), author=OUTSIDER)], authorities=(AUTHORITY,) + ), + (), + ) + + def test_the_composer_preserves_raw_arrivals_for_the_owning_resolver(self): + """Collapsing here would let two collapsed inputs reset the raw cap.""" + + links = chain(6) + self.assertEqual(len(combine_evidence(links, links)), 12) + self.assertEqual(len(combine_evidence((), links)), 6) + self.assertEqual(len(combine_evidence(links, ())), 6) + with self.assertRaises(LineageError): + combine_evidence((), ("not an episode",)) + + +class ReviewerAdmissionTests(unittest.TestCase): + """Contributors are refused; an uninvolved lane is admitted. Fails closed.""" + + def admit(self, lane, **overrides): + kwargs = dict( + repo=REPO, + pr_number=PR, + pr_meta=pr_meta(), + head_sha=TAKEN, + identity=IDENTITY, + episodes=(takeover(),), + ) + kwargs.update(overrides) + return reviewer_admission(lane, **kwargs) + + def test_contributors_are_refused_and_an_independent_lane_is_admitted(self): + for lane, admitted in (("devin", False), ("codex", False), ("claude", True)): + with self.subTest(lane=lane): + decision = self.admit(lane) + self.assertEqual(decision["admitted"], admitted) + self.assertEqual(decision["current_writer"], "codex") + self.assertEqual(decision["contributors"], ["devin", "codex"]) + if not admitted: + self.assertEqual(decision["reason"], "contributor_not_independent") + self.assertTrue(decision["owner_action"]) + + def test_admission_uses_the_head_the_caller_pinned(self): + decision = self.admit("claude", head_sha=MOVED) + self.assertFalse(decision["admitted"]) + self.assertEqual(decision["reason"], "lineage_waiting") + + def test_contradictory_signals_without_evidence_refuse_every_lane(self): + for lane in ("devin", "codex", "claude"): + with self.subTest(lane=lane): + decision = self.admit(lane, episodes=()) + self.assertFalse(decision["admitted"]) + self.assertEqual(decision["reason"], "lineage_conflict") + + def test_a_configured_branch_label_disagreement_admits_nobody(self): + disagreeing = pr_meta( + author="a-human", labels=("builder:claude",), branch="codex/topic" + ) + for lane in ("codex", "claude"): + with self.subTest(lane=lane): + decision = self.admit(lane, pr_meta=disagreeing, episodes=()) + self.assertFalse(decision["admitted"]) + self.assertEqual(decision["reason"], "lineage_conflict") + + def test_an_unconfigured_deployment_keeps_its_old_answer(self): + disagreeing = pr_meta( + author="a-human", labels=("builder:claude",), branch="codex/topic" + ) + decision = self.admit( + "codex", pr_meta=disagreeing, episodes=(), identity=UNCONFIGURED + ) + self.assertTrue(decision["admitted"]) + + def test_a_matched_branch_and_label_keep_their_intended_behaviour(self): + matched = pr_meta( + author="a-human", labels=("builder:claude",), branch="claude/topic" + ) + self.assertTrue(self.admit("codex", pr_meta=matched, episodes=())["admitted"]) + refused = self.admit("claude", pr_meta=matched, episodes=()) + self.assertFalse(refused["admitted"]) + self.assertEqual(refused["reason"], "contributor_not_independent") + + def test_a_reviewer_with_no_contract_at_all_still_excludes_itself(self): + """The floor, at the admission boundary rather than in isolation.""" + + decision = reviewer_admission( + "codex", + repo=REPO, + pr_number=PR, + pr_meta=pr_meta( + author="chatgpt-codex-connector[bot]", + labels=(), + branch="codex/topic", + ), + head_sha=TAKEN, + identity={"enabled": False}, + episodes=(), + ) + self.assertFalse(decision["admitted"]) + self.assertEqual(decision["reason"], "contributor_not_independent") + + def test_unreadable_evidence_refuses_rather_than_admitting(self): + decision = self.admit("claude", episodes=({"schema": "nope"},)) + self.assertFalse(decision["admitted"]) + self.assertIn(decision["reason"], {"lineage_conflict", "lineage_unreadable"}) + + def test_an_invalid_lane_name_is_refused(self): + decision = self.admit("") + self.assertFalse(decision["admitted"]) + self.assertEqual(decision["reason"], "reviewer_lane_invalid") + + def test_a_misnamed_own_lane_refuses_before_any_resolution(self): + conflicting = dict(IDENTITY) + conflicting["labels"] = dict(IDENTITY["labels"]) + conflicting["labels"]["builder:codex"] = "claude" + with self.assertRaises(ReviewerIdentityInvalid): + self.admit("codex", identity=conflicting) + + def test_requiring_independence_raises_bounded_metadata_only(self): + with self.assertRaises(ReviewerNotIndependent) as raised: + require_independent_reviewer( + "codex", + repo=REPO, + pr_number=PR, + pr_meta=pr_meta(), + head_sha=TAKEN, + identity=IDENTITY, + episodes=(takeover(),), + ) + message = str(raised.exception) + self.assertIn("contributor_not_independent", message) + self.assertNotIn("/Users", message) + self.assertNotIn("session", message) + + def test_the_wrapper_facing_form_reports_a_plain_runtime_error(self): + with self.assertRaises(RuntimeError) as raised: + require_reviewer_lane( + "codex", + REPO, + PR, + pr_meta(), + TAKEN, + identity=IDENTITY, + episodes=(takeover(),), + ) + message = str(raised.exception) + self.assertIn("contributor_not_independent", message) + self.assertIn(TAKEN[:12], message) + self.assertNotIn("/Users", message) + self.assertNotIn("\n", message) + + def test_an_admitted_lane_is_returned_rather_than_raised(self): + decision = require_reviewer_lane( + "claude", + REPO, + PR, + pr_meta(), + TAKEN, + identity=IDENTITY, + episodes=(takeover(),), + ) + self.assertTrue(decision["admitted"]) + + +class PrLineageTests(unittest.TestCase): + """Metadata shapes the caller may actually be handed.""" + + def test_metadata_that_names_no_branch_refuses_the_exact_target(self): + """Admission is an exact-target claim, so an unbound one cannot answer.""" + + for meta in ( + {}, + {"user": "a-string", "head": None, "labels": None}, + {"labels": ["not-an-object"], "head": {"ref": " "}}, + ): + with self.subTest(meta=meta): + lineage = pr_lineage( + repo=REPO, + pr_number=PR, + pr_meta=meta, + head_sha=TAKEN, + identity=IDENTITY, + episodes=(), + ) + self.assertEqual(lineage.status, "conflict") + self.assertEqual(lineage.reason, "target_invalid") + + def test_a_complete_target_with_no_named_lane_excludes_nobody(self): + lineage = pr_lineage( + repo=REPO, + pr_number=PR, + pr_meta={"labels": ["not-an-object"], "head": {"ref": "fix/typo"}}, + head_sha=TAKEN, + identity=IDENTITY, + episodes=(), + ) + self.assertEqual(lineage.status, "resolved") + self.assertEqual(lineage.reason, "no_builder_identity") + + def test_an_unbound_reviewer_admission_admits_nobody(self): + decision = reviewer_admission( + "claude", + repo=REPO, + pr_number=PR, + pr_meta={"user": {"login": "a-human"}, "labels": []}, + head_sha=TAKEN, + identity=IDENTITY, + episodes=(), + ) + self.assertFalse(decision["admitted"]) + self.assertEqual(decision["reason"], "lineage_conflict") + + def test_the_branch_comes_from_the_metadata_the_caller_fetched(self): + lineage = pr_lineage( + repo=REPO, + pr_number=PR, + pr_meta=pr_meta(author="a-human", labels=(), branch="codex/topic"), + head_sha=TAKEN, + identity=IDENTITY, + episodes=(), + ) + self.assertEqual(lineage.current_writer, "codex") + + +if __name__ == "__main__": # pragma: no cover + unittest.main() diff --git a/tools/builder_lineage.py b/tools/builder_lineage.py new file mode 100644 index 00000000..9f0bf594 --- /dev/null +++ b/tools/builder_lineage.py @@ -0,0 +1,1502 @@ +"""Exact-head builder contribution lineage: the pure contract. + +A pull request can be built by more than one Code Mower builder lane. The +opener, the branch prefix and the single active ``builder:*`` label each +describe at most one of those lanes, so any of them alone will misdescribe a +pull request that changed hands. This module keeps the ordered contribution +history instead, and derives the one current writer from it. + +Trust rules this module exists to enforce: + +* A contribution episode is evidence produced by a verified handoff and + delivery boundary that observed the source writer going quiescent and + observed both heads. A caller-supplied boolean, a pull request body marker, a + commit trailer, the opener or the most recent label are none of them able to + attest that a takeover happened. +* Episodes are bound to repository, pull request, branch, source lane, + destination lane, expected head and resulting head. An episode that does not + bind to the pull request under decision is not evidence about it. +* Resolution is exact-head. Lineage that stops short of the current head is + *waiting*, never a guess about who wrote the current diff. +* Conflicting, duplicated, unchained or unbound evidence fails closed with one + concise owner action rather than picking a winner. + +Purity is part of the contract, not an implementation detail. Nothing here +reads the environment, touches a store, opens a socket or imports an adapter: +every decision is a function of its explicit arguments. That is what lets the +same answer be computed by the package, by the vendored ``tools/`` copy inside +a generated product repository, and by a reviewer host that has no private +state at all. Recording, publication, label application and consumer +activation are deliberately somewhere else. + +Everything here is metadata-only: lane names, a repository slug, a pull request +number, a branch name and commit shas. +""" + +from __future__ import annotations + +import hashlib +import json +import re +from dataclasses import dataclass, field +from typing import Any, Callable, Iterable, Mapping, Sequence + + +SCHEMA = "code_mower.builderLineage.v1" +EPISODE_SCHEMA = "code_mower.contributionEpisode.v1" +RECORD_SCHEMA = "code_mower.builderLineageRecord.v1" + +#: Hidden marker used to publish bounded lineage metadata on a pull request. +#: Only comments from an already trusted author are parsed; the marker is a +#: transport, never an authorization. +LINEAGE_MARKER = "CODE_MOWER_BUILDER_LINEAGE" +LINEAGE_MARKER_RE = re.compile( + r"", + re.DOTALL, +) + +#: Marker *presence*, decided without looking at the payload at all. +#: +#: :data:`LINEAGE_MARKER_RE` only matches a complete, object-shaped, terminated +#: marker, so looking for evidence with it alone means a broken marker is not +#: seen rather than read as broken. Absence and unreadability are opposite +#: answers: one admits an independent reviewer on the ordinary single-builder +#: story, the other must stop. Presence is found first, and the payload is then +#: required to parse. +LINEAGE_MARKER_PRESENT_RE = re.compile(r"" + + +def _reject_duplicate_keys(pairs: Sequence[tuple[str, Any]]) -> dict[str, Any]: + """``object_pairs_hook`` that refuses an object naming a key twice. + + ``json.loads`` keeps the last value for a repeated key, so one marker can + carry two answers to the same question -- two ``episodes`` lists, two + ``schema`` values, two ``resulting_head`` shas inside one episode -- and + every reader silently agrees on whichever came last. Which one describes + the diff is exactly what must not be decided by parser order. This applies + at every depth, so a conflicting nested binding is refused too. + """ + + seen: dict[str, Any] = {} + for key, value in pairs: + if key in seen: + raise ValueError(f"duplicate key in published builder lineage: {key}") + seen[key] = value + return seen + + +def _loads_without_duplicate_keys(payload: str) -> Any: + return json.loads(payload, object_pairs_hook=_reject_duplicate_keys) + + +def episodes_from_comment_body(body: str) -> tuple[ContributionEpisode, ...]: + """Parse lineage markers out of one already trusted comment body. + + The caller decides trust. This function never treats the presence of a + marker as evidence that its author was allowed to publish one. + """ + + if body is None: + body = "" + if not isinstance(body, str): + raise LineageError("published builder lineage is unreadable") + text = _text(body) + present = LINEAGE_MARKER_PRESENT_RE.findall(text) + if not present: + # An ordinary comment. Not evidence, and not a failure either. + return () + if len(present) > 1: + # Two markers on one comment cannot both be "the" published lineage, + # and which one describes the head is exactly what may not be guessed. + raise LineageError("published builder lineage is ambiguous") + matches = LINEAGE_MARKER_RE.findall(text[:MAX_MARKER_BODY_CHARS]) + if len(matches) != 1: + # The marker is there, but no single complete object payload parses out + # of it: unterminated, not an object, or cut off past the bound. + raise LineageError("published builder lineage is unreadable") + try: + payload = _loads_without_duplicate_keys(matches[0]) + except (ValueError, RecursionError): + raise LineageError("published builder lineage is unreadable") from None + if not isinstance(payload, Mapping) or payload.get("schema") != SCHEMA: + raise LineageError("published builder lineage schema is unsupported") + items = payload.get("episodes") + if not isinstance(items, list) or len(items) > MAX_EPISODES: + raise LineageError("published builder lineage is unreadable") + if not items: + # A marker announces lineage. The publisher refuses to publish zero + # episodes, so a trusted marker carrying an empty chain is not a + # history that happens to be empty -- it is a claim that contradicts + # itself, and reading it as ordinary absence is how announced evidence + # disappears into the single-builder answer. + raise LineageError("published builder lineage declares no episodes") + return tuple(episode_from_mapping(item) for item in items) + + +def published_episodes( + comments: Sequence[Mapping[str, Any]], + *, + trusted_author: Callable[[str], bool], +) -> tuple[ContributionEpisode, ...]: + """Collect lineage episodes published by already trusted comment authors. + + The hidden marker is a transport for bounded metadata. Trust comes from the + caller's author check, never from the marker being present, so an untrusted + commenter cannot assert a takeover into existence -- and equally cannot + force a stop by posting a deliberately broken one. + + Records are validated, not skipped: an entry that is not a readable comment + raises rather than quietly shrinking the history it was part of. + """ + + def arrivals(): + for comment in comments or (): + if not isinstance(comment, Mapping): + raise LineageError("the comment history holds an entry that is not a comment") + require_comment_record(comment, what="the comment history") + login = comment_author_login(comment) + if not login or not trusted_author(login): + continue + yield from episodes_from_comment_body(comment.get("body") or "") + + return tuple(bounded_arrivals(arrivals())) + + +def merge_episodes( + recorded: Sequence[Any] = (), + incoming: Sequence[Any] = (), +) -> tuple[ContributionEpisode, ...]: + """Concatenate two sources of raw arrivals, validated and bounded. + + Deliberately no deduplication. Collapsing here would let two independently + collapsed inputs each arrive under the cap and together exceed it, and the + owning resolver already collapses exactly once -- in the one place that can + also see a contradiction at a position and refuse it. Every arrival is + coerced to an episode here, so a malformed member is the documented + contract error rather than an incidental attribute failure later. + """ + + def arrivals(): + for source in (recorded, incoming): + for item in source or (): + yield require_episode(item) + + return tuple(bounded_arrivals(arrivals())) + + +@dataclass(frozen=True) +class LineageContext: + """The trusted exact-head evidence a consumer carries into resolution. + + Every field has to come from something the caller verified for itself: the + repository it is running in, the head it fetched from the pull request, and + episodes published by an author it already trusts. An empty context is not + a failure -- it is the honest statement that this call has no exact-head + evidence, and resolution falls back to the ordinary identity-only answer, + which still refuses a configured branch/label disagreement. + """ + + repo: str = "" + pr_number: Any = 0 + branch: str = "" + head_sha: str = "" + episodes: tuple[ContributionEpisode, ...] = field(default_factory=tuple) + + +#: A consumer that has no exact-head evidence at all. +NO_LINEAGE = LineageContext() + + +def lineage_context( + *, + repo: str, + pr_number: Any, + branch: str = "", + head_sha: str | None = "", + comments: Sequence[Mapping[str, Any]] | None = (), + trusted_author: Callable[[str], bool] | None = None, +) -> LineageContext: + """Assemble exact-head lineage evidence for one consumer entry path. + + A *deliberately* absent target -- nothing supplied at all -- is the + ordinary no-evidence case and yields :data:`NO_LINEAGE`, which callers + resolve through the explicit identity-only route. A partially populated or + malformed target is not absent: it raises, because quietly returning + :data:`NO_LINEAGE` for it skipped branch binding and decided a diff nobody + bound. Evidence that arrives with no target to bind it to raises for the + same reason. Unreadable published evidence propagates as + :class:`LineageError` for the caller's fail-closed handling. + """ + + episodes: tuple[ContributionEpisode, ...] = () + if trusted_author is not None: + episodes = published_episodes(comments or (), trusted_author=trusted_author) + supplied = any(_text(value) for value in (repo, branch, head_sha)) or ( + pr_number not in (None, 0, False, "") + ) + if not supplied: + if episodes: + raise LineageError( + "published builder lineage arrived without a pull request to bind it to" + ) + return NO_LINEAGE + target = require_exact_target( + repo=repo, pr_number=pr_number, branch=branch, head_sha=head_sha + ) + return LineageContext( + repo=target.repo, + pr_number=target.pr_number, + branch=target.branch, + head_sha=target.head_sha, + episodes=episodes, + ) + + +def resolve_lineage_context( + context: LineageContext | None, + *, + identity: Mapping[str, Any] | None, + labels: Sequence[str] = (), + author: str = "", +) -> Lineage: + """Resolve a carried :class:`LineageContext` through the one decision. + + Only a context that is *exactly* absent takes the explicit identity-only + route. A partially populated context, or one carrying episodes, is a claim + about a specific pull request and goes through exact-target resolution, + which refuses it rather than answering about a diff it never bound. + """ + + if context is None or context == NO_LINEAGE: + return resolve_configured_identity( + identity=identity, labels=labels, author=author + ) + return resolve_builder_lineage( + identity=identity, + labels=labels, + author=author, + repo=context.repo, + pr_number=context.pr_number, + branch=context.branch, + head_sha=context.head_sha, + episodes=context.episodes, + ) + + +# --- pure keys and label planning -------------------------------------------- + + +def pr_key(repo: str, pr_number: Any) -> str: + """A stable, opaque key for one pull request's private lineage record. + + Pure by design: the recording side lives in a later stage, but both sides + have to derive the same key from the same pair, and deriving it twice in + two places is how they stop matching. + """ + + seed = json.dumps([_text(repo).lower(), _pr_number(pr_number)], sort_keys=True) + return "l" + hashlib.sha256(seed.encode()).hexdigest()[:62] + + +def builder_label_for(lane: str, identity: Mapping[str, Any] | None = None) -> str: + """The one active label that names ``lane`` as the current writer.""" + + writer = _lane(lane) + if not writer: + return "" + label_map = ( + identity.get("labels") if isinstance(identity, Mapping) else None + ) + if isinstance(label_map, Mapping): + for label in sorted(_text(item) for item in label_map): + if _lane(label_map.get(label)) == writer and label.startswith("builder:"): + return label + return f"builder:{writer}" + + +def builder_label_plan( + lineage: Lineage, + *, + current_labels: Sequence[str] = (), + identity: Mapping[str, Any] | None = None, +) -> dict[str, Any]: + """Plan reconciliation to exactly one active builder label. + + A plan, never an application: this stage computes what would have to change + and nothing mutates. The label set says who may write next, and after a + verified takeover that is exactly one lane. Historical contributions are + not deleted by this -- they live in the recorded lineage, which is what + reviewer exclusion and the status projection read. Unresolved lineage plans + no mutation at all; a label moved on a guess is the failure this contract + exists to stop. + """ + + label_map = identity.get("labels") if isinstance(identity, Mapping) else None + known = ( + {_text(label): _lane(lane) for label, lane in label_map.items()} + if isinstance(label_map, Mapping) + else {} + ) + present = tuple( + dict.fromkeys( + label + for label in (_text(item) for item in current_labels) + if label and (label.startswith("builder:") or known.get(label)) + ) + ) + if not lineage.resolved or not lineage.current_writer: + return { + "schema": SCHEMA, + "status": "blocked", + "reason": lineage.reason if not lineage.resolved else "no_builder_identity", + "head_sha": lineage.head_sha, + "current_writer": lineage.current_writer, + "add": [], + "remove": [], + "owner_action": lineage.owner_action or _OWNER_ACTIONS["label_outside_lineage"], + } + + writer = lineage.current_writer + target = next( + ( + label + for label in present + if known.get(label) == writer or label == f"builder:{writer}" + ), + "", + ) or builder_label_for(writer, identity) + remove = [label for label in present if label != target] + add = [] if target in present else [target] + return { + "schema": SCHEMA, + "status": "reconcile" if (add or remove) else "current", + "reason": lineage.reason, + "head_sha": lineage.head_sha, + "current_writer": writer, + "add": add, + "remove": remove, + "owner_action": "", + }