diff --git a/CHANGELOG.md b/CHANGELOG.md index 5d62c1e2..5b8f51c4 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -7,6 +7,13 @@ later entries are regular releases. ## Unreleased +### Added + +- A pure typed builder-lineage contract with immutable exact targets, validated + contribution chains, explicit comment history and authority accounts, and + contributor-aware reviewer admission. Includes a standalone init support + module; live consumers will adopt the contract in a later stage. + ### 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..a0f15c9c 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": "src/code_mower/builder_lineage.py", + "target": "src/code_mower/builder_lineage.py" + }, { "kind": "core", "source": "src/code_mower/builder_runs.py", @@ -2056,6 +2061,11 @@ "kind": "workflow", "source": "generated", "target": "templates/workflows/trailer-comment-labeler.yml.j2" + }, + { + "kind": "core", + "source": "tools/builder_lineage.py", + "target": "tools/builder_lineage.py" } ], "mode": "materialize", diff --git a/src/code_mower/builder_lineage.py b/src/code_mower/builder_lineage.py new file mode 100644 index 00000000..5726985d --- /dev/null +++ b/src/code_mower/builder_lineage.py @@ -0,0 +1,495 @@ +"""Pure, explicit builder lineage contracts. + +All invalid contract inputs raise ContractError. Target and Episode constructors +own canonicalization; their mapping factories use those same constructors. +Repo, SHA, lane, account, label and prefix signals are trimmed/lowercased. +Branch bindings are validated verbatim (including case), never normalized. + +Only Chain.from_arrivals validates/deduplicates episode streams. Feed public +parse_markers arrivals and private arrivals together, for example with +itertools.chain, exactly once. History([]) means a successful empty fetch; +unavailable history is not an input to this module. No operation performs I/O. +""" +from __future__ import annotations + +from collections.abc import Iterable, Mapping +from dataclasses import dataclass +import json +import re + +MAX_EPISODES = 32 +MAX_RAW_ARRIVALS = 560 +LINEAGE_SCHEMA = "code_mower.builderLineage.v1" +LINEAGE_MARKER = "CODE_MOWER_BUILDER_LINEAGE" +CONTINUATION_WRITER_STATE = "same_writer" +HANDOFF_WRITER_STATES = frozenset({"terminated", "completed", "cancelled"}) + + +class ContractError(ValueError): + """An explicit input violates the lineage contract; no fallback is made.""" + + +def _text(value: object, name: str) -> str: + if not isinstance(value, str) or not value.strip(): + raise ContractError(f"{name} must be a nonempty string") + return value.strip().lower() + + +def _lane(value: object) -> str: + value = _text(value, "lane") + if not re.fullmatch(r"[a-z0-9][a-z0-9_-]*", value): + raise ContractError("invalid lane") + return value + + +def _account(value: object) -> str: + value = _text(value, "account") + if not re.fullmatch(r"[a-z0-9][a-z0-9-]*(?:\[bot\])?", value): + raise ContractError("invalid account") + return value + + +def _repo(value: object) -> str: + value = _text(value, "repo") + if not re.fullmatch(r"[a-z0-9][a-z0-9_.-]*/[a-z0-9][a-z0-9_.-]*", value): + raise ContractError("invalid repo") + return value + + +def _sha(value: object) -> str: + value = _text(value, "SHA") + if not re.fullmatch(r"[0-9a-f]{40}", value): + raise ContractError("SHA must contain 40 hex characters") + return value + + +def _positive(value: object, name: str) -> int: + if type(value) is not int or value <= 0: + raise ContractError(f"{name} must be a positive non-boolean integer") + return value + + +def _branch(value: object) -> str: + if (not isinstance(value, str) or not 0 < len(value) <= 200 + or not re.fullmatch(r"[A-Za-z0-9][A-Za-z0-9/_.-]*", value) + or any(part in value for part in ("..", "//", "@{")) + or value.endswith("/") + or any(p.startswith(".") or p.endswith((".", ".lock")) for p in value.split("/"))): + raise ContractError("invalid exact branch") + return value + + +def _mapping(value: object, allowed: set[str]) -> Mapping: + if not isinstance(value, Mapping) or set(value) - allowed: + raise ContractError("expected mapping with supported fields only") + return value + + +def _unique_pairs(pairs: list[tuple[str, object]]) -> dict: + result = {} + for key, value in pairs: + if key in result: + raise ContractError("duplicate JSON key") + result[key] = value + return result + + +def _json(text: object) -> object: + if not isinstance(text, str): + raise ContractError("expected JSON text") + try: + return json.loads(text, object_pairs_hook=_unique_pairs, + parse_constant=lambda _: _bad_json_constant()) + except (ValueError, RecursionError) as exc: + raise ContractError("invalid or non-unique JSON") from exc + + +def _bad_json_constant() -> None: + raise ContractError("non-finite JSON constant") + + +@dataclass(frozen=True, slots=True, init=False) +class Target: + repo: str + pr_number: int + branch: str + head_sha: str + + def __init__(self, repo: object = None, pr_number: object = None, + branch: object = None, head_sha: object = None): + object.__setattr__(self, "repo", _repo(repo)) + object.__setattr__(self, "pr_number", _positive(pr_number, "PR number")) + object.__setattr__(self, "branch", _branch(branch)) + object.__setattr__(self, "head_sha", _sha(head_sha)) + + @classmethod + def from_mapping(cls, value: object) -> Target: + return cls(**_mapping(value, {"repo", "pr_number", "branch", "head_sha"})) + + +_EPISODE_FIELDS = frozenset({"sequence", "repo", "pr_number", "branch", "source_lane", + "destination_lane", "expected_head", "resulting_head", + "writer_state", "kind"}) + + +@dataclass(frozen=True, slots=True, init=False) +class Episode: + sequence: int + repo: str + pr_number: int + branch: str + source_lane: str + destination_lane: str + expected_head: str + resulting_head: str + writer_state: str + kind: str + + def __init__(self, sequence: object = None, repo: object = None, + pr_number: object = None, branch: object = None, + source_lane: object = None, destination_lane: object = None, + expected_head: object = None, resulting_head: object = None, + writer_state: object = None, kind: object = "handoff"): + target = Target(repo, pr_number, branch, resulting_head) + values = dict(sequence=_positive(sequence, "sequence"), repo=target.repo, + pr_number=target.pr_number, branch=target.branch, + source_lane=_lane(source_lane), destination_lane=_lane(destination_lane), + expected_head=_sha(expected_head), resulting_head=target.head_sha, + writer_state=_text(writer_state, "writer state"), kind=_text(kind, "kind")) + if values["kind"] == "handoff": + if (values["source_lane"] == values["destination_lane"] + or values["writer_state"] not in HANDOFF_WRITER_STATES): + raise ContractError("handoff requires distinct lanes and a verified stopped writer") + elif values["kind"] == "continuation": + if (values["source_lane"] != values["destination_lane"] + or values["writer_state"] != CONTINUATION_WRITER_STATE): + raise ContractError("continuation requires the same writer and lane") + else: + raise ContractError("unknown episode kind") + for key, value in values.items(): + object.__setattr__(self, key, value) + + @classmethod + def from_mapping(cls, value: object) -> Episode: + return cls(**_mapping(value, _EPISODE_FIELDS)) + + def to_mapping(self) -> dict: + return {key: getattr(self, key) for key in sorted(_EPISODE_FIELDS)} + + +def _aliases(value: object, section: str) -> tuple[tuple[str, str], ...]: + if not isinstance(value, Mapping): + raise ContractError(f"{section} must be a mapping") + result = {} + for key, lane in value.items(): + key = _account(key) if section == "authors" else _text(key, section) + lane = _lane(lane) + if key in result and result[key] != lane: + raise ContractError(f"conflicting normalized {section} aliases") + result[key] = lane + return tuple(sorted(result.items())) + + +@dataclass(frozen=True, slots=True, init=False) +class Identity: + """Canonical identity policy; unknown fields reject rather than disappear. + + Prefixes supply provenance only when require_verified_lineage is true. + The longest matching prefix wins. An absent branch contract preserves + identity-only behavior, while all explicit policy fields remain stored. + """ + enabled: bool + labels: tuple[tuple[str, str], ...] + authors: tuple[tuple[str, str], ...] + branch_prefixes: tuple[tuple[str, str], ...] + require_verified_lineage: bool + + def __init__(self, value: object): + value = _mapping(value, {"enabled", "labels", "authors", "branch_prefixes", + "require_verified_lineage"}) + for key in ("enabled", "require_verified_lineage"): + flag = value.get(key, False) + if type(flag) is not bool: + raise ContractError(f"{key} must be boolean") + object.__setattr__(self, key, flag) + for key in ("labels", "authors", "branch_prefixes"): + object.__setattr__(self, key, _aliases(value.get(key, {}), key)) + + @classmethod + def from_mapping(cls, value: object) -> Identity: + return cls(value) + + @classmethod + def from_text(cls, value: object) -> Identity: + return cls(_json(value)) + + def to_mapping(self) -> dict: + return dict(enabled=self.enabled, labels=dict(self.labels), authors=dict(self.authors), + branch_prefixes=dict(self.branch_prefixes), + require_verified_lineage=self.require_verified_lineage) + + def with_reviewer_floor(self, reviewer: object, accounts: object) -> Identity: + """Add explicit own-reviewer accounts and builder label; remapping rejects. + + The floor enables exclusion and retains every declared mapping and + branch policy field. It does not grant marker trust to these accounts. + """ + lane = _lane(reviewer) + accounts = Authorities(accounts).accounts + value = self.to_mapping() + value["enabled"] = True + for section, keys in (("authors", accounts), ("labels", (f"builder:{lane}",))): + for key in keys: + previous = value[section].get(key) + if previous is not None and previous != lane: + raise ContractError("own-reviewer floor cannot remap a declared identity") + value[section][key] = lane + return Identity(value) + + +@dataclass(frozen=True, slots=True, init=False) +class Authorities: + """An explicit immutable account set, independent of builder identities.""" + accounts: frozenset[str] + + def __init__(self, accounts: object): + if not isinstance(accounts, (list, tuple, set, frozenset)): + raise ContractError("authorities must be an explicit account collection") + object.__setattr__(self, "accounts", frozenset(_account(a) for a in accounts)) + + +@dataclass(frozen=True, slots=True) +class _Comment: + body: str + account: str | None + + +def _comment(value: object) -> _Comment: + if not isinstance(value, Mapping): + raise ContractError("comment must be a mapping") + body = value.get("body", "") + if not isinstance(body, str): + raise ContractError("present comment body must be text") + accounts = set() + for field in ("user", "author"): + if field not in value or value[field] is None: + continue + user = value[field] + if not isinstance(user, Mapping): + raise ContractError("present comment user/author must be an object or null") + if "login" in user: + accounts.add(_account(user["login"])) + if len(accounts) > 1: + raise ContractError("conflicting comment authors") + return _Comment(body, next(iter(accounts), None)) + + +@dataclass(frozen=True, slots=True, init=False) +class History: + comments: tuple[_Comment, ...] + + def __init__(self, comments: object): + if not isinstance(comments, list): + raise ContractError("history must be an explicit list of raw comments") + object.__setattr__(self, "comments", tuple(_comment(c) for c in comments)) + + @classmethod + def from_pages(cls, pages: object) -> History: + if not isinstance(pages, list) or any(not isinstance(p, list) for p in pages): + raise ContractError("slurped history must be a list of list pages") + return cls([comment for page in pages for comment in page]) + + +@dataclass(frozen=True, slots=True, init=False) +class Chain: + """A validated Target-bound chain. Use from_arrivals, including for []. + + At most 560 arrivals are consumed, plus the first disallowed probe. Budget + precedes canonicalization and deduplication. At most 32 episodes are kept. + Sequence order is canonical, regardless of replay or arrival order. + """ + target: Target + episodes: tuple[Episode, ...] + raw_arrival_count: int + + def __init__(self, *args, **kwargs): + raise ContractError("use Chain.from_arrivals(target, raw_arrivals)") + + @classmethod + def from_arrivals(cls, target: Target, arrivals: Iterable[Episode | Mapping]) -> Chain: + if not isinstance(target, Target): + raise ContractError("Chain requires an exact Target") + if isinstance(arrivals, (str, bytes, Mapping, Chain)): + raise ContractError("arrivals must be an episode iterable") + try: + iterator = iter(arrivals) + except TypeError as exc: + raise ContractError("arrivals must be an episode iterable") from exc + episodes: dict[int, Episode] = {} + count = 0 + for raw in iterator: + count += 1 + if count > MAX_RAW_ARRIVALS: + raise ContractError("raw episode arrival budget exceeded") + episode = raw if isinstance(raw, Episode) else Episode.from_mapping(raw) + if (episode.repo, episode.pr_number, episode.branch) != ( + target.repo, target.pr_number, target.branch): + raise ContractError("episode does not match the exact Target") + previous = episodes.get(episode.sequence) + if previous is not None and previous != episode: + raise ContractError("conflicting duplicate episode") + episodes[episode.sequence] = episode + if len(episodes) > MAX_EPISODES: + raise ContractError("distinct episode budget exceeded") + ordered = tuple(episodes[key] for key in sorted(episodes)) + for index, episode in enumerate(ordered, 1): + if episode.sequence != index: + raise ContractError("episode sequences must be contiguous from 1") + if index == 1: + if episode.kind != "handoff": + raise ContractError("first episode must be a handoff") + else: + previous = ordered[index - 2] + if (episode.expected_head, episode.source_lane) != ( + previous.resulting_head, previous.destination_lane): + raise ContractError("episode head/lane continuity mismatch") + chain = object.__new__(cls) + object.__setattr__(chain, "target", target) + object.__setattr__(chain, "episodes", ordered) + object.__setattr__(chain, "raw_arrival_count", count) + return chain + + +def parse_markers(history: History, authorities: Authorities) -> Iterable[Mapping]: + """Yield raw trusted arrivals for a single subsequent Chain factory. + + Trust is checked only after History validates every transport record. This + lazy stream never deduplicates, judges current heads, or resets a budget. + Consume it with Chain.from_arrivals; do not materialize unbounded histories. + An announced marker must be one complete, unique-key, nonempty JSON chain. + """ + if not isinstance(history, History) or not isinstance(authorities, Authorities): + raise ContractError("parsing requires History and Authorities") + return _marker_arrivals(history, authorities) + + +def _marker_arrivals(history: History, authorities: Authorities) -> Iterable[Mapping]: + for comment in history.comments: + if comment.account not in authorities.accounts or LINEAGE_MARKER not in comment.body: + continue + if comment.body.count(LINEAGE_MARKER) != 1: + raise ContractError("multiple announced lineage markers") + match = re.search(r"", + comment.body, re.DOTALL) + if match is None: + raise ContractError("malformed or unterminated lineage marker") + payload = _mapping(_json(match.group(1)), {"schema", "episodes"}) + if payload.get("schema") != LINEAGE_SCHEMA: + raise ContractError("unsupported lineage schema") + episodes = payload.get("episodes") + if not isinstance(episodes, list) or not episodes: + raise ContractError("marker must contain a nonempty episode list") + # Never slice a snapshot. Even repeated arrivals count at the Chain. + yield from episodes + + +def render(chain: Chain) -> str: + """Render all episodes of a validated nonempty Chain, without truncation.""" + if not isinstance(chain, Chain) or not chain.episodes: + raise ContractError("render requires a validated nonempty Chain") + payload = dict(schema=LINEAGE_SCHEMA, episodes=[e.to_mapping() for e in chain.episodes]) + return f"" + + +@dataclass(frozen=True, slots=True, init=False) +class Lineage: + """Resolved decision; only resolve/resolve_identity_only construct decisions. + + ready admits unrelated reviewers; waiting and conflict never admit anyone. + A target of None identifies the separate, explicit identity-only decision. + """ + target: Target | None + contributors: tuple[str, ...] + current_writer: str | None + status: str + reason: str + owner_action: str + + def __init__(self, *args, **kwargs): + raise ContractError("Lineage decisions must be resolved") + + +def _decision(target: Target | None, contributors: Iterable[str], writer: str | None, + status: str, reason: str, action: str = "") -> Lineage: + decision = object.__new__(Lineage) + for key, value in dict(target=target, contributors=tuple(sorted(set(contributors))), + current_writer=writer, status=status, reason=reason, + owner_action=action).items(): + object.__setattr__(decision, key, value) + return decision + + +def _signals(identity: Identity, author: object, labels: object, branch: object) -> tuple[set, str | None]: + if not isinstance(identity, Identity): + raise ContractError("resolution requires Identity") + # Empty author explicitly means no known author; malformed types still fail. + account = None if author == "" else _account(author) + if not isinstance(labels, (list, tuple, set, frozenset)): + raise ContractError("labels must be an explicit collection") + label_keys = tuple(_text(label, "label") for label in labels) + branch = _branch(branch) + if not identity.enabled: + return set(), None + label_map, author_map = dict(identity.labels), dict(identity.authors) + lanes = {label_map[label] for label in label_keys if label in label_map} + if account in author_map: + lanes.add(author_map[account]) + prefixes = [(prefix, lane) for prefix, lane in identity.branch_prefixes + if branch.lower().startswith(prefix)] if identity.require_verified_lineage else [] + branch_lane = max(prefixes, key=lambda item: len(item[0]))[1] if prefixes else None + return lanes, branch_lane + + +def _identity_decision(target: Target | None, identity: Identity, author: object, + labels: object, branch: object) -> Lineage: + lanes, branch_lane = _signals(identity, author, labels, branch) + if branch_lane: + lanes.add(branch_lane) + if len(lanes) > 1: + return _decision(target, lanes, None, "conflict", "identity_branch_conflict", + "Provide verified recorded lineage or correct identity metadata.") + writer = next(iter(lanes), None) + return _decision(target, lanes, writer, "ready", "identity_matched" if writer else "no_identity") + + +def resolve_identity_only(identity: Identity, author: object, labels: object, branch: object) -> Lineage: + """Explicit identity-only control; accepts no Target, History or evidence.""" + return _identity_decision(None, identity, author, labels, branch) + + +def resolve(chain: Chain, identity: Identity, author: object, labels: object) -> Lineage: + """Resolve against the Chain's own exact Target; stale final heads wait.""" + if not isinstance(chain, Chain): + raise ContractError("exact resolution requires a validated Chain") + if not chain.episodes: + return _identity_decision(chain.target, identity, author, labels, chain.target.branch) + lanes, branch_lane = _signals(identity, author, labels, chain.target.branch) + contributors = {lane for e in chain.episodes for lane in (e.source_lane, e.destination_lane)} + if branch_lane: + lanes.add(branch_lane) + writer = chain.episodes[-1].destination_lane + if lanes - contributors: + return _decision(chain.target, contributors | lanes, writer, "conflict", + "unrecorded_contributor", "Provide verified lineage for every contributor.") + if chain.episodes[-1].resulting_head != chain.target.head_sha: + return _decision(chain.target, contributors, writer, "waiting", "lineage_head_pending", + "Wait for verified lineage at the current PR head.") + return _decision(chain.target, contributors, writer, "ready", "verified_lineage") + + +def admit(lineage: Lineage, reviewer: object) -> bool: + """Only a ready full decision permits a reviewer outside all contributors.""" + lane = _lane(reviewer) + if not isinstance(lineage, Lineage): + raise ContractError("admission requires a resolved Lineage decision") + return lineage.status == "ready" and lane not in lineage.contributors diff --git a/src/code_mower/init.py b/src/code_mower/init.py index ba970690..7d0e9ebe 100644 --- a/src/code_mower/init.py +++ b/src/code_mower/init.py @@ -182,6 +182,12 @@ ) PRODUCT_SUPPORT_FILES = ( + ( + "tools/builder_lineage.py", + "builder_lineage.py", + "product-support-helper", + "0644", + ), ( "tools/code_mower", "templates/product-support/code_mower", diff --git a/src/code_mower/package_manifest.py b/src/code_mower/package_manifest.py index d9387afb..0cdc6c52 100644 --- a/src/code_mower/package_manifest.py +++ b/src/code_mower/package_manifest.py @@ -9,6 +9,8 @@ DEFAULT_PACKAGE_CONFIG = "code-mower.example.yml" PACKAGE_FILES = ( + ("src/code_mower/builder_lineage.py", "src/code_mower/builder_lineage.py", "core"), + ("tools/builder_lineage.py", "tools/builder_lineage.py", "core"), ("tools/CODE_MOWER_APACHE_LICENSE.txt", "LICENSE", "package"), ("tools/CODE_MOWER_NOTICE.txt", "NOTICE", "package"), ("tools/code_mower_cli.py", "src/code_mower/cli.py", "core"), diff --git a/tests/minimal_lineage_fixtures.py b/tests/minimal_lineage_fixtures.py new file mode 100644 index 00000000..0c652475 --- /dev/null +++ b/tests/minimal_lineage_fixtures.py @@ -0,0 +1,73 @@ +"""Isolated metadata examples for the declared lineage contract.""" +from code_mower.builder_lineage import CONTINUATION_WRITER_STATE, Episode, Target + +REPO = "owner/repo" +PR = 42 +BRANCH = "Devin/42-work" +AUTHORITY = "owner" +OPENED = "a" * 40 +TAKEN = "b" * 40 + + +def head(index): + return f"{index:040x}" + + +def target(**changes): + return Target(**(dict(repo=REPO, pr_number=PR, branch=BRANCH, head_sha=TAKEN) | changes)) + + +def episode(**changes): + return Episode(**(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") | changes)) + + +def episodes(length): + return [episode(resulting_head=head(1)), *[ + episode(sequence=i, kind="continuation", source_lane="codex", + expected_head=head(i - 1), resulting_head=head(i), + writer_state=CONTINUATION_WRITER_STATE) for i in range(2, length + 1) + ]] + + +def identity_mapping(**changes): + return dict(enabled=True, + labels={"builder:devin": "devin", "builder:codex": "codex", + "builder:claude": "claude"}, + authors={"devin-bot[bot]": "devin", "codex-bot[bot]": "codex", + "claude-bot[bot]": "claude"}, + branch_prefixes={"devin/": "devin", "codex/": "codex", "claude/": "claude", + "feature/cx-": "codex"}, + require_verified_lineage=True) | changes + + +def comment(body="ordinary comment", account=AUTHORITY, field="user"): + return {field: {"login": account}, "body": body} + + +# Supported REST and GraphQL shapes, including deleted accounts and optional body. +VALID_COMMENTS = [ + {"user": None, "body": "deleted account"}, + {"user": {"login": AUTHORITY}}, + {"user": {}, "body": "no login"}, + {"body": "no author"}, + {"author": None, "body": "deleted account"}, + {"author": {"login": AUTHORITY}, "body": "ordinary comment"}, +] + + +class BoundedArrivals: + """The 562nd read is a bug, even when every earlier arrival is identical.""" + def __init__(self, value): + self.value = value + self.reads = 0 + + def __iter__(self): + return self + + def __next__(self): + self.reads += 1 + if self.reads > 561: + raise AssertionError("raw arrival iterator overconsumed") + return self.value diff --git a/tests/test_minimal_lineage_contract.py b/tests/test_minimal_lineage_contract.py new file mode 100644 index 00000000..2b6658f6 --- /dev/null +++ b/tests/test_minimal_lineage_contract.py @@ -0,0 +1,246 @@ +"""Behavior tests for immutable inputs, exact chains, and reviewer admission.""" +from dataclasses import FrozenInstanceError +import json +import unittest + +from code_mower.builder_lineage import ( + Chain, ContractError, Episode, Identity, Lineage, Target, admit, + resolve, resolve_identity_only, +) +from minimal_lineage_fixtures import ( + BRANCH, OPENED, PR, REPO, TAKEN, BoundedArrivals, episode, episodes, head, + identity_mapping, target, +) + + +class ValueContractTests(unittest.TestCase): + def test_target_constructor_and_mapping_store_canonical_values(self): + values = dict(repo=" OWNER/Repo ", pr_number=PR, branch=BRANCH, head_sha=f" {TAKEN.upper()} ") + direct = Target(**values) + self.assertEqual(direct, Target.from_mapping(values)) + self.assertEqual(direct, target()) + self.assertEqual(direct.branch, BRANCH) + self.assertNotEqual(direct, target(branch=BRANCH.lower())) + with self.assertRaises(FrozenInstanceError): + direct.branch = "other" + + def test_target_missing_and_malformed_fields_reject(self): + baseline = dict(repo=REPO, pr_number=PR, branch=BRANCH, head_sha=TAKEN) + cases = {"repo": [None, "", "repo", {}, False], + "pr_number": [None, False, True, 0, -1, 1.0, "42"], + "branch": [None, "", " feature/x", "a..b", "a//b", "a/", "a/.b", "a.lock"], + "head_sha": [None, "", "abc1234", "z" * 40, False]} + for field, invalids in cases.items(): + for value in invalids: + with self.subTest(field=field, value=value), self.assertRaises(ContractError): + Target(**(baseline | {field: value})) + with self.subTest(missing=field), self.assertRaises(ContractError): + Target.from_mapping({k: v for k, v in baseline.items() if k != field}) + for value in (None, False, [], {}, {"unknown": True}): + with self.subTest(value=value), self.assertRaises(ContractError): + Target.from_mapping(value) + with self.assertRaises(ContractError): + Target() + + def test_episode_constructor_mapping_equality_and_canonical_storage(self): + fields = episode().to_mapping() | dict(repo=" OWNER/REPO ", source_lane=" DEVIN ", + destination_lane=" CODEX ", expected_head=f" {OPENED.upper()} ", + resulting_head=TAKEN.upper(), kind=" HANDOFF ", + writer_state=" TERMINATED ") + parsed = Episode.from_mapping(fields) + self.assertEqual(parsed, Episode(**fields)) + self.assertEqual(parsed, episode()) + self.assertEqual(parsed.to_mapping()["source_lane"], "devin") + fields["source_lane"] = "claude" + self.assertEqual(parsed.source_lane, "devin") + with self.assertRaises(FrozenInstanceError): + parsed.source_lane = "claude" + + def test_invalid_episode_ingestion_has_one_error(self): + invalid = [dict(pr_number=True), dict(pr_number=0), dict(sequence=False), dict(sequence=0), + dict(branch=""), dict(source_lane=""), dict(destination_lane="bad lane"), + dict(resulting_head="x"), dict(expected_head=None), dict(kind="invented"), + dict(writer_state="running"), dict(writer_state="unknown"), + dict(destination_lane="devin"), + dict(kind="continuation", writer_state="same_writer"), + dict(kind="continuation", source_lane="codex", writer_state="terminated")] + for changes in invalid: + fields = episode().to_mapping() | changes + for factory in (lambda fields=fields: Episode(**fields), lambda fields=fields: Episode.from_mapping(fields)): + with self.subTest(changes=changes), self.assertRaises(ContractError): + factory() + for field in set(episode().to_mapping()) - {"kind"}: + with self.subTest(missing=field), self.assertRaises(ContractError): + Episode.from_mapping({k: v for k, v in episode().to_mapping().items() if k != field}) + + +class IdentityTests(unittest.TestCase): + def test_explicit_mapping_text_and_immutable_snapshot(self): + raw = identity_mapping() + identity = Identity(raw) + self.assertEqual(identity, Identity.from_mapping(raw)) + self.assertEqual(identity, Identity.from_text(json.dumps(raw))) + self.assertEqual(identity.to_mapping(), raw) + raw["authors"]["outsider"] = "codex" + self.assertNotIn("outsider", dict(identity.authors)) + with self.assertRaises(FrozenInstanceError): + identity.enabled = False + for value in (None, "", [], {"labels": None}, {"enabled": 1}, {"opaque": {}}, + {"require_verified_lineage": "yes"}): + with self.subTest(value=value), self.assertRaises(ContractError): + Identity(value) + with self.assertRaises(ContractError): + Identity.from_text('{"enabled":true,"enabled":false}') + + def test_aliases_collapse_or_reject_in_both_insertion_orders(self): + for section, first, second in (("authors", " BOT[bot] ", "bot[bot]"), + ("branch_prefixes", " Feature/CX- ", "feature/cx-"), + ("labels", " BUILDER:CODEX ", "builder:codex")): + for reverse in (False, True): + pairs = [(first, " CODEX "), (second, "codex")] + if reverse: + pairs.reverse() + parsed = Identity(identity_mapping(**{section: dict(pairs)})) + self.assertEqual(getattr(parsed, section), ((second, "codex"),)) + pairs = [(first, "codex"), (second, "claude")] + if reverse: + pairs.reverse() + with self.subTest(section=section, reverse=reverse), self.assertRaises(ContractError): + Identity(identity_mapping(**{section: dict(pairs)})) + + def test_longest_prefix_and_exact_branch_case_are_independent(self): + identity = Identity(identity_mapping(branch_prefixes={" feature/ ": "devin", "FEATURE/CX-": "codex"})) + decision = resolve(Chain.from_arrivals(target(branch="Feature/CX-42"), []), identity, + "codex-bot[bot]", ["builder:codex"]) + self.assertEqual(decision.current_writer, "codex") + self.assertEqual(decision.target.branch, "Feature/CX-42") + self.assertTrue(admit(decision, "claude")) + + def test_own_reviewer_floor_uses_canonical_policy_and_preserves_fields(self): + identity = Identity(identity_mapping(enabled=False, authors={" OWN-BOT ": " CODEX "})) + floor = identity.with_reviewer_floor(" CODEX ", ["own-bot", " SECOND-BOT "]) + self.assertTrue(floor.enabled) + self.assertEqual(floor.labels, identity.labels) + self.assertEqual(floor.branch_prefixes, identity.branch_prefixes) + self.assertEqual(floor.require_verified_lineage, identity.require_verified_lineage) + self.assertEqual(dict(floor.authors), {"own-bot": "codex", "second-bot": "codex"}) + decision = resolve_identity_only(floor, " SECOND-BOT ", [], "Feature/CX-42") + self.assertFalse(admit(decision, " CODEX ")) + self.assertTrue(admit(decision, " Claude ")) + for section, alias in (("authors", " OWN-BOT "), ("labels", " BUILDER:CODEX ")): + with self.subTest(section=section), self.assertRaises(ContractError): + Identity(identity_mapping(**{section: {alias: "claude"}})).with_reviewer_floor("codex", ["own-bot"]) + + +class ChainResolutionTests(unittest.TestCase): + def setUp(self): + self.identity = Identity(identity_mapping()) + + def test_no_raw_overload_no_target_fallback_and_explicit_control(self): + for invalid in (None, [], {}, episode(), target()): + with self.subTest(invalid=invalid), self.assertRaises(ContractError): + resolve(invalid, self.identity, "devin-bot[bot]", []) + for invalid in (None, {}, False): + with self.assertRaises(ContractError): + Chain.from_arrivals(invalid, []) + with self.assertRaises(ContractError): + Chain(target(), []) + with self.assertRaises(ContractError): + Lineage() + control = resolve_identity_only(self.identity, "codex-bot[bot]", [], "codex/42-work") + self.assertIsNone(control.target) + self.assertEqual(control.contributors, ("codex",)) + self.assertFalse(admit(control, "codex")) + with self.assertRaises(TypeError): + resolve_identity_only(self.identity, "", [], BRANCH, []) + with self.assertRaises(TypeError): + resolve(Chain.from_arrivals(target(), []), self.identity, "", [], target()) + + def test_zero_episodes_considers_branch_conflicts_and_custom_no_contract(self): + empty = Chain.from_arrivals(target(), []) + conflict = resolve(empty, self.identity, "codex-bot[bot]", ["builder:codex"]) + self.assertEqual(conflict.status, "conflict") + self.assertEqual(conflict.contributors, ("codex", "devin")) + self.assertTrue(conflict.owner_action) + self.assertFalse(admit(conflict, "claude")) + matched = resolve(empty, self.identity, "devin-bot[bot]", []) + self.assertEqual(matched.reason, "identity_matched") + self.assertTrue(admit(matched, "claude")) + custom = resolve(Chain.from_arrivals(target(branch="feature/cx-42"), []), self.identity, + "codex-bot[bot]", []) + self.assertEqual(custom.current_writer, "codex") + no_contract = Identity(identity_mapping(require_verified_lineage=False)) + self.assertEqual(resolve(empty, no_contract, "codex-bot[bot]", []).status, "ready") + self.assertEqual(resolve(empty, Identity({}), "", []).reason, "no_identity") + + def test_recorded_takeover_excludes_all_contributors_and_waits_on_stale_head(self): + chain = Chain.from_arrivals(target(), [episode()]) + decision = resolve(chain, self.identity, "devin-bot[bot]", ["builder:codex"]) + self.assertEqual((decision.status, decision.reason, decision.current_writer), + ("ready", "verified_lineage", "codex")) + self.assertEqual(decision.contributors, ("codex", "devin")) + for lane in (" DEVIN ", " CODEX "): + self.assertFalse(admit(decision, lane)) + self.assertTrue(admit(decision, " CLAUDE ")) + stale = resolve(Chain.from_arrivals(target(head_sha=OPENED), [episode()]), self.identity, + "devin-bot[bot]", ["builder:codex"]) + self.assertEqual((stale.status, stale.reason), ("waiting", "lineage_head_pending")) + self.assertTrue(stale.owner_action) + self.assertFalse(admit(stale, "claude")) + unrecorded = resolve(chain, self.identity, "claude-bot[bot]", ["builder:codex"]) + self.assertEqual(unrecorded.status, "conflict") + self.assertFalse(admit(unrecorded, "unrelated")) + + def test_multiple_handoffs_and_continuations_keep_every_contributor(self): + items = [episode(), episode(sequence=2, source_lane="codex", destination_lane="claude", + expected_head=TAKEN, resulting_head=head(2)), + episode(sequence=3, kind="continuation", source_lane="claude", destination_lane="claude", + writer_state="same_writer", expected_head=head(2), resulting_head=head(3))] + decision = resolve(Chain.from_arrivals(target(head_sha=head(3)), items), self.identity, + "devin-bot[bot]", ["builder:claude"]) + self.assertEqual(decision.current_writer, "claude") + self.assertEqual(decision.contributors, ("claude", "codex", "devin")) + self.assertTrue(all(not admit(decision, lane) for lane in decision.contributors)) + self.assertTrue(admit(decision, "independent")) + + def test_common_target_contiguity_and_predecessor_are_required(self): + valid = episodes(2) + cases = [[episode(repo="owner/other-repo")], [episode(pr_number=PR + 1)], + [episode(branch=BRANCH.lower())], [valid[1]], [valid[0], episode(sequence=3)], + [valid[0], Episode.from_mapping(valid[1].to_mapping() | {"expected_head": OPENED})], + [valid[0], episode(sequence=2, source_lane="claude", expected_head=head(1))]] + for items in cases: + with self.subTest(items=items), self.assertRaises(ContractError): + Chain.from_arrivals(target(), items) + for invalid in (None, False, 0, "", {}, target()): + with self.subTest(invalid=invalid), self.assertRaises(ContractError): + Chain.from_arrivals(target(), invalid) + + def test_arrival_budget_precedes_dedup_and_never_overconsumes(self): + stream = BoundedArrivals(episode()) + with self.assertRaisesRegex(ContractError, "arrival budget"): + Chain.from_arrivals(target(), stream) + self.assertEqual(stream.reads, 561) + accepted = Chain.from_arrivals(target(), [episode()] * 560) + self.assertEqual(accepted.episodes, (episode(),)) + self.assertEqual(accepted.raw_arrival_count, 560) + conflicting = episode(resulting_head=head(3)) + with self.assertRaisesRegex(ContractError, "conflicting duplicate"): + Chain.from_arrivals(target(), [episode()] * 559 + [conflicting]) + with self.assertRaisesRegex(ContractError, "distinct episode budget"): + Chain.from_arrivals(target(), episodes(33)) + + def test_canonical_duplicates_collapse_and_replay_order_is_irrelevant(self): + raw = episode().to_mapping() | {"source_lane": " DEVIN ", "repo": " OWNER/REPO "} + chain = Chain.from_arrivals(target(), [raw, episode()]) + self.assertEqual(chain.episodes, (episode(),)) + self.assertEqual(chain.raw_arrival_count, 2) + values = episodes(32) + ordered = Chain.from_arrivals(target(head_sha=head(32)), reversed(values)) + self.assertEqual(ordered.episodes, tuple(values)) + with self.assertRaises(FrozenInstanceError): + ordered.target = target() + + +if __name__ == "__main__": + unittest.main() diff --git a/tests/test_minimal_lineage_packaging.py b/tests/test_minimal_lineage_packaging.py new file mode 100644 index 00000000..5671aae7 --- /dev/null +++ b/tests/test_minimal_lineage_packaging.py @@ -0,0 +1,98 @@ +"""Pure import boundary, byte parity, and real isolated init materialization.""" +import ast +import json +from pathlib import Path +import subprocess +import sys +import tempfile +import unittest + +from code_mower import init, package +from code_mower.config import load_config +from minimal_lineage_fixtures import episode, identity_mapping, target + +ROOT = Path(__file__).resolve().parents[1] +PURE_STDLIB = {"__future__", "collections", "dataclasses", "json", "re"} + + +class PackagingTests(unittest.TestCase): + def test_canonical_and_tools_bytes_match(self): + self.assertEqual((ROOT / "src/code_mower/builder_lineage.py").read_bytes(), + (ROOT / "tools/builder_lineage.py").read_bytes()) + + def test_entire_import_graph_is_stdlib_only_and_has_no_io_execution(self): + for relative in ("src/code_mower/builder_lineage.py", "tools/builder_lineage.py"): + tree = ast.parse((ROOT / relative).read_text()) + imports = set() + for node in ast.walk(tree): + if isinstance(node, ast.Import): + imports.update(alias.name.split(".")[0] for alias in node.names) + elif isinstance(node, ast.ImportFrom): + self.assertEqual(node.level, 0, "no project/transitive adapter imports") + imports.add(node.module.split(".")[0]) + elif isinstance(node, ast.Call): + if isinstance(node.func, ast.Name): + self.assertNotIn(node.func.id, {"open", "exec", "eval", "__import__", "compile", "input"}) + elif isinstance(node.func, ast.Attribute): + self.assertNotIn(node.func.attr, {"open", "read_text", "read_bytes", "write_text", + "write_bytes", "getenv", "system", "popen"}) + # Every reachable dependency terminates in the stdlib; there is no + # project helper whose own imports could transitively perform I/O. + self.assertEqual(imports, PURE_STDLIB) + + def test_canonical_manifest_regeneration_equality(self): + expected = package.committed_package_manifest_text(package.generate_committed_package_manifest(ROOT)) + self.assertEqual((ROOT / "code-mower-package-manifest.json").read_text(), expected) + manifest = json.loads(expected) + targets = {entry["target"] for entry in manifest["files_written"]} + self.assertIn("src/code_mower/builder_lineage.py", targets) + self.assertIn("tools/builder_lineage.py", targets) + + def test_init_materialized_helper_import_parse_resolve_without_package_or_repo(self): + config = load_config(ROOT / "src/code_mower/templates/code-mower.example.yml") + plan = init.render_init_plan(config, package_mode=True, repo_root=ROOT) + with tempfile.TemporaryDirectory(dir=ROOT) as scratch: + output = Path(scratch) / "product" + init.apply_init_plan(plan, output, source_root=ROOT) + helper = output / "tools/builder_lineage.py" + self.assertEqual(helper.read_bytes(), (ROOT / "src/code_mower/builder_lineage.py").read_bytes()) + fixture = dict(target=dict(repo=target().repo, pr_number=target().pr_number, + branch=target().branch, head_sha=target().head_sha), + episode=episode().to_mapping(), identity=identity_mapping()) + # -I -S removes the checkout/PYTHONPATH and installed package hooks. + # Only the generated tools directory is exposed to this fresh process. + program = ''' +import importlib.util +import json +from pathlib import Path +import sys +assert importlib.util.find_spec("code_mower") is None +assert "code_mower" not in sys.modules +sys.path.insert(0, str(Path.cwd() / "tools")) +import builder_lineage as core +assert Path(core.__file__).resolve() == Path.cwd() / "tools/builder_lineage.py" +value = json.load(sys.stdin) +target = core.Target.from_mapping(value["target"]) +chain = core.Chain.from_arrivals(target, [value["episode"]]) +body = core.render(chain) +history = core.History([{"user": {"login": "owner"}, "body": body}]) +parsed = core.Chain.from_arrivals(target, core.parse_markers(history, core.Authorities(["owner"]))) +assert parsed == chain +identity = core.Identity.from_mapping(value["identity"]) +decision = core.resolve(parsed, identity, "devin-bot[bot]", ["builder:codex"]) +assert decision.reason == "verified_lineage" +assert decision.current_writer == "codex" +assert not core.admit(decision, "codex") +assert not core.admit(decision, "devin") +assert core.admit(decision, "claude") +assert importlib.util.find_spec("code_mower") is None +print("isolated import/parse/resolve/admit passed") +''' + result = subprocess.run([sys.executable, "-I", "-S", "-c", program], cwd=output, + input=json.dumps(fixture), capture_output=True, text=True, timeout=30) + self.assertEqual(result.returncode, 0, result.stderr) + self.assertEqual(result.stdout.strip(), "isolated import/parse/resolve/admit passed") + + +if __name__ == "__main__": + unittest.main() diff --git a/tests/test_minimal_lineage_transport.py b/tests/test_minimal_lineage_transport.py new file mode 100644 index 00000000..30b74f02 --- /dev/null +++ b/tests/test_minimal_lineage_transport.py @@ -0,0 +1,185 @@ +"""Strict raw transports, trusted grammar, cumulative replay, and rendering.""" +from dataclasses import FrozenInstanceError +from itertools import chain as concatenate +import json +import unittest + +from code_mower.builder_lineage import ( + Authorities, Chain, ContractError, History, Identity, LINEAGE_MARKER, LINEAGE_SCHEMA, + admit, parse_markers, render, resolve, +) +from minimal_lineage_fixtures import ( + AUTHORITY, BRANCH, VALID_COMMENTS, comment, episode, episodes, head, identity_mapping, target, +) + + +def marker(payload): + return f"" + + +def payload(items): + return {"schema": LINEAGE_SCHEMA, "episodes": [e.to_mapping() for e in items]} + + +class HistoryTests(unittest.TestCase): + def test_raw_history_is_explicit_and_pages_have_a_distinct_factory(self): + invalid = (None, False, 0, "", {}, {"comments": []}, (), [None], [False], + [comment(), "wrong"], [[comment()]]) + for value in invalid: + with self.subTest(value=value), self.assertRaises(ContractError): + History(value) + self.assertEqual(History([]).comments, ()) + self.assertEqual(History.from_pages([]), History([])) + self.assertEqual(History.from_pages([[], [comment()], []]), History([comment()])) + for value in (None, False, "", {}, [comment()], [[], comment()], [[[comment()]]], [None]): + with self.subTest(pages=value), self.assertRaises(ContractError): + History.from_pages(value) + with self.assertRaises(TypeError): + History() + + def test_every_present_field_is_validated_even_for_untrusted_comments(self): + records = [comment() | {"body": value} for value in (None, False, 7, {}, [])] + for field in ("user", "author"): + records += [{field: value} for value in (False, 7, "owner", [])] + records += [{field: {"login": value}} for value in (None, False, 7, {}, [], "")] + records += [{"user": {"login": AUTHORITY}, "author": "invalid"}, + {"user": {"login": AUTHORITY}, "author": {"login": "outsider"}}] + for record in records: + with self.subTest(record=record), self.assertRaises(ContractError): + History([comment(), record]) + with self.subTest(page_record=record), self.assertRaises(ContractError): + History.from_pages([[comment()], [record]]) + + def test_nullable_optional_fields_and_immutable_storage(self): + history = History(VALID_COMMENTS) + self.assertEqual(len(history.comments), len(VALID_COMMENTS)) + self.assertEqual(list(parse_markers(history, Authorities([AUTHORITY]))), []) + raw = comment("original") + snapshot = History([raw]) + raw["body"] = "changed" + raw["user"]["login"] = "outsider" + self.assertEqual(snapshot.comments[0].body, "original") + self.assertEqual(snapshot.comments[0].account, AUTHORITY) + with self.assertRaises(FrozenInstanceError): + snapshot.comments = () + with self.assertRaises(FrozenInstanceError): + snapshot.comments[0].body = "changed" + + def test_authority_is_explicit_accounts_and_never_a_callback_or_identity(self): + raw = [f" {AUTHORITY.upper()} ", AUTHORITY] + authorities = Authorities(raw) + self.assertEqual(authorities.accounts, frozenset({AUTHORITY})) + raw.append("outsider") + self.assertNotIn("outsider", authorities.accounts) + for value in (None, False, "owner", {"owner": True}, lambda _: True, + Identity(identity_mapping()), [None], [False]): + with self.subTest(value=value), self.assertRaises(ContractError): + Authorities(value) + with self.assertRaises(FrozenInstanceError): + authorities.accounts = frozenset() + + +class MarkerTests(unittest.TestCase): + def setUp(self): + self.authorities = Authorities([AUTHORITY]) + self.identity = Identity(identity_mapping()) + + def parsed_chain(self, body, *, bound=None, account=AUTHORITY): + return Chain.from_arrivals(target() if bound is None else bound, + parse_markers(History([comment(body, account)]), self.authorities)) + + def test_marker_presence_absence_and_trust_before_grammar(self): + body = render(Chain.from_arrivals(target(), [episode()])) + for account in ("outsider", "devin-bot[bot]"): + for text in (body, LINEAGE_MARKER, f"' + duplicate_episode = good.replace('"sequence":1', '"sequence":1,"sequence":1') + invalid = [LINEAGE_MARKER, f""] + for text in invalid: + with self.subTest(text=text[:100]), self.assertRaises(ContractError): + self.parsed_chain(text) + + def test_marker_target_binding_and_mixed_target_chain(self): + good = render(Chain.from_arrivals(target(), [episode()])) + for bound in (target(repo="owner/other-repo"), target(pr_number=43), target(branch=BRANCH.lower())): + with self.subTest(bound=bound), self.assertRaises(ContractError): + self.parsed_chain(good, bound=bound) + items = episodes(2) + items[1] = type(items[1]).from_mapping(items[1].to_mapping() | {"repo": "owner/other-repo"}) + with self.assertRaises(ContractError): + self.parsed_chain(marker(payload(items))) + + def test_528_public_plus_32_private_arrivals_share_one_budget(self): + items = episodes(32) + comments = [comment(render(Chain.from_arrivals(target(head_sha=head(i)), items[:i]))) + for i in range(1, 33)] + history = History(comments) + combined = concatenate(parse_markers(history, self.authorities), items) + chain = Chain.from_arrivals(target(head_sha=head(32)), combined) + self.assertEqual(chain.raw_arrival_count, 560) + self.assertEqual(chain.episodes, tuple(items)) + decision = resolve(chain, self.identity, "devin-bot[bot]", ["builder:codex"]) + self.assertTrue(admit(decision, "claude")) + with self.assertRaisesRegex(ContractError, "arrival budget"): + Chain.from_arrivals(target(head_sha=head(32)), + concatenate(parse_markers(history, self.authorities), items, [items[0]])) + conflicting = episode(resulting_head=head(99)) + with self.assertRaisesRegex(ContractError, "conflicting duplicate"): + Chain.from_arrivals(target(head_sha=head(32)), + concatenate(parse_markers(history, self.authorities), [conflicting])) + + def test_current_head_is_judged_after_combining_public_and_private(self): + items = episodes(2) + public = History([comment(render(Chain.from_arrivals(target(head_sha=head(1)), items[:1])))]) + bound = target(head_sha=head(2)) + stale = Chain.from_arrivals(bound, parse_markers(public, self.authorities)) + self.assertEqual(resolve(stale, self.identity, "", []).status, "waiting") + combined = Chain.from_arrivals(bound, concatenate(parse_markers(public, self.authorities), items[1:])) + self.assertEqual(resolve(combined, self.identity, "", []).status, "ready") + + def test_render_roundtrip_one_and_32_episodes(self): + for count in (1, 32): + bound = target(head_sha=head(count)) + original = Chain.from_arrivals(bound, episodes(count)) + body = render(original) + for field in ("user", "author"): + restored = Chain.from_arrivals(bound, parse_markers( + History([comment(body, f" {AUTHORITY.upper()} ", field)]), self.authorities)) + self.assertEqual(restored, original) + self.assertEqual(render(restored), body) + + def test_arbitrary_empty_oversized_malformed_or_mixed_input_cannot_render(self): + for value in (None, False, {}, [], (), [episode()], episodes(33), + [episode(), episode(repo="owner/other-repo")], Chain.from_arrivals(target(), [])): + with self.subTest(value=type(value)), self.assertRaises(ContractError): + render(value) + with self.assertRaises(ContractError): + Chain.from_arrivals(target(), [{"source_lane": "devin"}]) + with self.assertRaises(ContractError): + Chain.from_arrivals(target(), episodes(33)) + + +if __name__ == "__main__": + unittest.main() diff --git a/tools/builder_lineage.py b/tools/builder_lineage.py new file mode 100644 index 00000000..5726985d --- /dev/null +++ b/tools/builder_lineage.py @@ -0,0 +1,495 @@ +"""Pure, explicit builder lineage contracts. + +All invalid contract inputs raise ContractError. Target and Episode constructors +own canonicalization; their mapping factories use those same constructors. +Repo, SHA, lane, account, label and prefix signals are trimmed/lowercased. +Branch bindings are validated verbatim (including case), never normalized. + +Only Chain.from_arrivals validates/deduplicates episode streams. Feed public +parse_markers arrivals and private arrivals together, for example with +itertools.chain, exactly once. History([]) means a successful empty fetch; +unavailable history is not an input to this module. No operation performs I/O. +""" +from __future__ import annotations + +from collections.abc import Iterable, Mapping +from dataclasses import dataclass +import json +import re + +MAX_EPISODES = 32 +MAX_RAW_ARRIVALS = 560 +LINEAGE_SCHEMA = "code_mower.builderLineage.v1" +LINEAGE_MARKER = "CODE_MOWER_BUILDER_LINEAGE" +CONTINUATION_WRITER_STATE = "same_writer" +HANDOFF_WRITER_STATES = frozenset({"terminated", "completed", "cancelled"}) + + +class ContractError(ValueError): + """An explicit input violates the lineage contract; no fallback is made.""" + + +def _text(value: object, name: str) -> str: + if not isinstance(value, str) or not value.strip(): + raise ContractError(f"{name} must be a nonempty string") + return value.strip().lower() + + +def _lane(value: object) -> str: + value = _text(value, "lane") + if not re.fullmatch(r"[a-z0-9][a-z0-9_-]*", value): + raise ContractError("invalid lane") + return value + + +def _account(value: object) -> str: + value = _text(value, "account") + if not re.fullmatch(r"[a-z0-9][a-z0-9-]*(?:\[bot\])?", value): + raise ContractError("invalid account") + return value + + +def _repo(value: object) -> str: + value = _text(value, "repo") + if not re.fullmatch(r"[a-z0-9][a-z0-9_.-]*/[a-z0-9][a-z0-9_.-]*", value): + raise ContractError("invalid repo") + return value + + +def _sha(value: object) -> str: + value = _text(value, "SHA") + if not re.fullmatch(r"[0-9a-f]{40}", value): + raise ContractError("SHA must contain 40 hex characters") + return value + + +def _positive(value: object, name: str) -> int: + if type(value) is not int or value <= 0: + raise ContractError(f"{name} must be a positive non-boolean integer") + return value + + +def _branch(value: object) -> str: + if (not isinstance(value, str) or not 0 < len(value) <= 200 + or not re.fullmatch(r"[A-Za-z0-9][A-Za-z0-9/_.-]*", value) + or any(part in value for part in ("..", "//", "@{")) + or value.endswith("/") + or any(p.startswith(".") or p.endswith((".", ".lock")) for p in value.split("/"))): + raise ContractError("invalid exact branch") + return value + + +def _mapping(value: object, allowed: set[str]) -> Mapping: + if not isinstance(value, Mapping) or set(value) - allowed: + raise ContractError("expected mapping with supported fields only") + return value + + +def _unique_pairs(pairs: list[tuple[str, object]]) -> dict: + result = {} + for key, value in pairs: + if key in result: + raise ContractError("duplicate JSON key") + result[key] = value + return result + + +def _json(text: object) -> object: + if not isinstance(text, str): + raise ContractError("expected JSON text") + try: + return json.loads(text, object_pairs_hook=_unique_pairs, + parse_constant=lambda _: _bad_json_constant()) + except (ValueError, RecursionError) as exc: + raise ContractError("invalid or non-unique JSON") from exc + + +def _bad_json_constant() -> None: + raise ContractError("non-finite JSON constant") + + +@dataclass(frozen=True, slots=True, init=False) +class Target: + repo: str + pr_number: int + branch: str + head_sha: str + + def __init__(self, repo: object = None, pr_number: object = None, + branch: object = None, head_sha: object = None): + object.__setattr__(self, "repo", _repo(repo)) + object.__setattr__(self, "pr_number", _positive(pr_number, "PR number")) + object.__setattr__(self, "branch", _branch(branch)) + object.__setattr__(self, "head_sha", _sha(head_sha)) + + @classmethod + def from_mapping(cls, value: object) -> Target: + return cls(**_mapping(value, {"repo", "pr_number", "branch", "head_sha"})) + + +_EPISODE_FIELDS = frozenset({"sequence", "repo", "pr_number", "branch", "source_lane", + "destination_lane", "expected_head", "resulting_head", + "writer_state", "kind"}) + + +@dataclass(frozen=True, slots=True, init=False) +class Episode: + sequence: int + repo: str + pr_number: int + branch: str + source_lane: str + destination_lane: str + expected_head: str + resulting_head: str + writer_state: str + kind: str + + def __init__(self, sequence: object = None, repo: object = None, + pr_number: object = None, branch: object = None, + source_lane: object = None, destination_lane: object = None, + expected_head: object = None, resulting_head: object = None, + writer_state: object = None, kind: object = "handoff"): + target = Target(repo, pr_number, branch, resulting_head) + values = dict(sequence=_positive(sequence, "sequence"), repo=target.repo, + pr_number=target.pr_number, branch=target.branch, + source_lane=_lane(source_lane), destination_lane=_lane(destination_lane), + expected_head=_sha(expected_head), resulting_head=target.head_sha, + writer_state=_text(writer_state, "writer state"), kind=_text(kind, "kind")) + if values["kind"] == "handoff": + if (values["source_lane"] == values["destination_lane"] + or values["writer_state"] not in HANDOFF_WRITER_STATES): + raise ContractError("handoff requires distinct lanes and a verified stopped writer") + elif values["kind"] == "continuation": + if (values["source_lane"] != values["destination_lane"] + or values["writer_state"] != CONTINUATION_WRITER_STATE): + raise ContractError("continuation requires the same writer and lane") + else: + raise ContractError("unknown episode kind") + for key, value in values.items(): + object.__setattr__(self, key, value) + + @classmethod + def from_mapping(cls, value: object) -> Episode: + return cls(**_mapping(value, _EPISODE_FIELDS)) + + def to_mapping(self) -> dict: + return {key: getattr(self, key) for key in sorted(_EPISODE_FIELDS)} + + +def _aliases(value: object, section: str) -> tuple[tuple[str, str], ...]: + if not isinstance(value, Mapping): + raise ContractError(f"{section} must be a mapping") + result = {} + for key, lane in value.items(): + key = _account(key) if section == "authors" else _text(key, section) + lane = _lane(lane) + if key in result and result[key] != lane: + raise ContractError(f"conflicting normalized {section} aliases") + result[key] = lane + return tuple(sorted(result.items())) + + +@dataclass(frozen=True, slots=True, init=False) +class Identity: + """Canonical identity policy; unknown fields reject rather than disappear. + + Prefixes supply provenance only when require_verified_lineage is true. + The longest matching prefix wins. An absent branch contract preserves + identity-only behavior, while all explicit policy fields remain stored. + """ + enabled: bool + labels: tuple[tuple[str, str], ...] + authors: tuple[tuple[str, str], ...] + branch_prefixes: tuple[tuple[str, str], ...] + require_verified_lineage: bool + + def __init__(self, value: object): + value = _mapping(value, {"enabled", "labels", "authors", "branch_prefixes", + "require_verified_lineage"}) + for key in ("enabled", "require_verified_lineage"): + flag = value.get(key, False) + if type(flag) is not bool: + raise ContractError(f"{key} must be boolean") + object.__setattr__(self, key, flag) + for key in ("labels", "authors", "branch_prefixes"): + object.__setattr__(self, key, _aliases(value.get(key, {}), key)) + + @classmethod + def from_mapping(cls, value: object) -> Identity: + return cls(value) + + @classmethod + def from_text(cls, value: object) -> Identity: + return cls(_json(value)) + + def to_mapping(self) -> dict: + return dict(enabled=self.enabled, labels=dict(self.labels), authors=dict(self.authors), + branch_prefixes=dict(self.branch_prefixes), + require_verified_lineage=self.require_verified_lineage) + + def with_reviewer_floor(self, reviewer: object, accounts: object) -> Identity: + """Add explicit own-reviewer accounts and builder label; remapping rejects. + + The floor enables exclusion and retains every declared mapping and + branch policy field. It does not grant marker trust to these accounts. + """ + lane = _lane(reviewer) + accounts = Authorities(accounts).accounts + value = self.to_mapping() + value["enabled"] = True + for section, keys in (("authors", accounts), ("labels", (f"builder:{lane}",))): + for key in keys: + previous = value[section].get(key) + if previous is not None and previous != lane: + raise ContractError("own-reviewer floor cannot remap a declared identity") + value[section][key] = lane + return Identity(value) + + +@dataclass(frozen=True, slots=True, init=False) +class Authorities: + """An explicit immutable account set, independent of builder identities.""" + accounts: frozenset[str] + + def __init__(self, accounts: object): + if not isinstance(accounts, (list, tuple, set, frozenset)): + raise ContractError("authorities must be an explicit account collection") + object.__setattr__(self, "accounts", frozenset(_account(a) for a in accounts)) + + +@dataclass(frozen=True, slots=True) +class _Comment: + body: str + account: str | None + + +def _comment(value: object) -> _Comment: + if not isinstance(value, Mapping): + raise ContractError("comment must be a mapping") + body = value.get("body", "") + if not isinstance(body, str): + raise ContractError("present comment body must be text") + accounts = set() + for field in ("user", "author"): + if field not in value or value[field] is None: + continue + user = value[field] + if not isinstance(user, Mapping): + raise ContractError("present comment user/author must be an object or null") + if "login" in user: + accounts.add(_account(user["login"])) + if len(accounts) > 1: + raise ContractError("conflicting comment authors") + return _Comment(body, next(iter(accounts), None)) + + +@dataclass(frozen=True, slots=True, init=False) +class History: + comments: tuple[_Comment, ...] + + def __init__(self, comments: object): + if not isinstance(comments, list): + raise ContractError("history must be an explicit list of raw comments") + object.__setattr__(self, "comments", tuple(_comment(c) for c in comments)) + + @classmethod + def from_pages(cls, pages: object) -> History: + if not isinstance(pages, list) or any(not isinstance(p, list) for p in pages): + raise ContractError("slurped history must be a list of list pages") + return cls([comment for page in pages for comment in page]) + + +@dataclass(frozen=True, slots=True, init=False) +class Chain: + """A validated Target-bound chain. Use from_arrivals, including for []. + + At most 560 arrivals are consumed, plus the first disallowed probe. Budget + precedes canonicalization and deduplication. At most 32 episodes are kept. + Sequence order is canonical, regardless of replay or arrival order. + """ + target: Target + episodes: tuple[Episode, ...] + raw_arrival_count: int + + def __init__(self, *args, **kwargs): + raise ContractError("use Chain.from_arrivals(target, raw_arrivals)") + + @classmethod + def from_arrivals(cls, target: Target, arrivals: Iterable[Episode | Mapping]) -> Chain: + if not isinstance(target, Target): + raise ContractError("Chain requires an exact Target") + if isinstance(arrivals, (str, bytes, Mapping, Chain)): + raise ContractError("arrivals must be an episode iterable") + try: + iterator = iter(arrivals) + except TypeError as exc: + raise ContractError("arrivals must be an episode iterable") from exc + episodes: dict[int, Episode] = {} + count = 0 + for raw in iterator: + count += 1 + if count > MAX_RAW_ARRIVALS: + raise ContractError("raw episode arrival budget exceeded") + episode = raw if isinstance(raw, Episode) else Episode.from_mapping(raw) + if (episode.repo, episode.pr_number, episode.branch) != ( + target.repo, target.pr_number, target.branch): + raise ContractError("episode does not match the exact Target") + previous = episodes.get(episode.sequence) + if previous is not None and previous != episode: + raise ContractError("conflicting duplicate episode") + episodes[episode.sequence] = episode + if len(episodes) > MAX_EPISODES: + raise ContractError("distinct episode budget exceeded") + ordered = tuple(episodes[key] for key in sorted(episodes)) + for index, episode in enumerate(ordered, 1): + if episode.sequence != index: + raise ContractError("episode sequences must be contiguous from 1") + if index == 1: + if episode.kind != "handoff": + raise ContractError("first episode must be a handoff") + else: + previous = ordered[index - 2] + if (episode.expected_head, episode.source_lane) != ( + previous.resulting_head, previous.destination_lane): + raise ContractError("episode head/lane continuity mismatch") + chain = object.__new__(cls) + object.__setattr__(chain, "target", target) + object.__setattr__(chain, "episodes", ordered) + object.__setattr__(chain, "raw_arrival_count", count) + return chain + + +def parse_markers(history: History, authorities: Authorities) -> Iterable[Mapping]: + """Yield raw trusted arrivals for a single subsequent Chain factory. + + Trust is checked only after History validates every transport record. This + lazy stream never deduplicates, judges current heads, or resets a budget. + Consume it with Chain.from_arrivals; do not materialize unbounded histories. + An announced marker must be one complete, unique-key, nonempty JSON chain. + """ + if not isinstance(history, History) or not isinstance(authorities, Authorities): + raise ContractError("parsing requires History and Authorities") + return _marker_arrivals(history, authorities) + + +def _marker_arrivals(history: History, authorities: Authorities) -> Iterable[Mapping]: + for comment in history.comments: + if comment.account not in authorities.accounts or LINEAGE_MARKER not in comment.body: + continue + if comment.body.count(LINEAGE_MARKER) != 1: + raise ContractError("multiple announced lineage markers") + match = re.search(r"", + comment.body, re.DOTALL) + if match is None: + raise ContractError("malformed or unterminated lineage marker") + payload = _mapping(_json(match.group(1)), {"schema", "episodes"}) + if payload.get("schema") != LINEAGE_SCHEMA: + raise ContractError("unsupported lineage schema") + episodes = payload.get("episodes") + if not isinstance(episodes, list) or not episodes: + raise ContractError("marker must contain a nonempty episode list") + # Never slice a snapshot. Even repeated arrivals count at the Chain. + yield from episodes + + +def render(chain: Chain) -> str: + """Render all episodes of a validated nonempty Chain, without truncation.""" + if not isinstance(chain, Chain) or not chain.episodes: + raise ContractError("render requires a validated nonempty Chain") + payload = dict(schema=LINEAGE_SCHEMA, episodes=[e.to_mapping() for e in chain.episodes]) + return f"" + + +@dataclass(frozen=True, slots=True, init=False) +class Lineage: + """Resolved decision; only resolve/resolve_identity_only construct decisions. + + ready admits unrelated reviewers; waiting and conflict never admit anyone. + A target of None identifies the separate, explicit identity-only decision. + """ + target: Target | None + contributors: tuple[str, ...] + current_writer: str | None + status: str + reason: str + owner_action: str + + def __init__(self, *args, **kwargs): + raise ContractError("Lineage decisions must be resolved") + + +def _decision(target: Target | None, contributors: Iterable[str], writer: str | None, + status: str, reason: str, action: str = "") -> Lineage: + decision = object.__new__(Lineage) + for key, value in dict(target=target, contributors=tuple(sorted(set(contributors))), + current_writer=writer, status=status, reason=reason, + owner_action=action).items(): + object.__setattr__(decision, key, value) + return decision + + +def _signals(identity: Identity, author: object, labels: object, branch: object) -> tuple[set, str | None]: + if not isinstance(identity, Identity): + raise ContractError("resolution requires Identity") + # Empty author explicitly means no known author; malformed types still fail. + account = None if author == "" else _account(author) + if not isinstance(labels, (list, tuple, set, frozenset)): + raise ContractError("labels must be an explicit collection") + label_keys = tuple(_text(label, "label") for label in labels) + branch = _branch(branch) + if not identity.enabled: + return set(), None + label_map, author_map = dict(identity.labels), dict(identity.authors) + lanes = {label_map[label] for label in label_keys if label in label_map} + if account in author_map: + lanes.add(author_map[account]) + prefixes = [(prefix, lane) for prefix, lane in identity.branch_prefixes + if branch.lower().startswith(prefix)] if identity.require_verified_lineage else [] + branch_lane = max(prefixes, key=lambda item: len(item[0]))[1] if prefixes else None + return lanes, branch_lane + + +def _identity_decision(target: Target | None, identity: Identity, author: object, + labels: object, branch: object) -> Lineage: + lanes, branch_lane = _signals(identity, author, labels, branch) + if branch_lane: + lanes.add(branch_lane) + if len(lanes) > 1: + return _decision(target, lanes, None, "conflict", "identity_branch_conflict", + "Provide verified recorded lineage or correct identity metadata.") + writer = next(iter(lanes), None) + return _decision(target, lanes, writer, "ready", "identity_matched" if writer else "no_identity") + + +def resolve_identity_only(identity: Identity, author: object, labels: object, branch: object) -> Lineage: + """Explicit identity-only control; accepts no Target, History or evidence.""" + return _identity_decision(None, identity, author, labels, branch) + + +def resolve(chain: Chain, identity: Identity, author: object, labels: object) -> Lineage: + """Resolve against the Chain's own exact Target; stale final heads wait.""" + if not isinstance(chain, Chain): + raise ContractError("exact resolution requires a validated Chain") + if not chain.episodes: + return _identity_decision(chain.target, identity, author, labels, chain.target.branch) + lanes, branch_lane = _signals(identity, author, labels, chain.target.branch) + contributors = {lane for e in chain.episodes for lane in (e.source_lane, e.destination_lane)} + if branch_lane: + lanes.add(branch_lane) + writer = chain.episodes[-1].destination_lane + if lanes - contributors: + return _decision(chain.target, contributors | lanes, writer, "conflict", + "unrecorded_contributor", "Provide verified lineage for every contributor.") + if chain.episodes[-1].resulting_head != chain.target.head_sha: + return _decision(chain.target, contributors, writer, "waiting", "lineage_head_pending", + "Wait for verified lineage at the current PR head.") + return _decision(chain.target, contributors, writer, "ready", "verified_lineage") + + +def admit(lineage: Lineage, reviewer: object) -> bool: + """Only a ready full decision permits a reviewer outside all contributors.""" + lane = _lane(reviewer) + if not isinstance(lineage, Lineage): + raise ContractError("admission requires a resolved Lineage decision") + return lineage.status == "ready" and lane not in lineage.contributors