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": "", + }