From 7ad0d2e0e933c1bd42456a8b225d73e7d2d9de37 Mon Sep 17 00:00:00 2001 From: Jeff Huber Date: Tue, 15 Sep 2026 00:16:01 -0700 Subject: [PATCH 1/3] Add pure typed builder lineage contract Introduce immutable explicit targets, episodes, identity policy, history, authorities, target-bound chains and reviewer decisions. Validate combined raw arrivals once before deduplication and keep exact branch case separate from normalized identity signals. Include standalone init support and owning unittest coverage without activating live consumers. Builder-Provider: codex Builder-Executor: codex_cli --- CHANGELOG.md | 7 + code-mower-package-manifest.json | 10 + src/code_mower/builder_lineage.py | 495 ++++++++++++++++++++++++ src/code_mower/init.py | 6 + src/code_mower/package_manifest.py | 2 + tests/minimal_lineage_fixtures.py | 73 ++++ tests/test_minimal_lineage_contract.py | 246 ++++++++++++ tests/test_minimal_lineage_packaging.py | 97 +++++ tests/test_minimal_lineage_transport.py | 185 +++++++++ tools/builder_lineage.py | 495 ++++++++++++++++++++++++ 10 files changed, 1616 insertions(+) create mode 100644 src/code_mower/builder_lineage.py create mode 100644 tests/minimal_lineage_fixtures.py create mode 100644 tests/test_minimal_lineage_contract.py create mode 100644 tests/test_minimal_lineage_packaging.py create mode 100644 tests/test_minimal_lineage_transport.py create mode 100644 tools/builder_lineage.py 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..69df2ce2 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": "src/code_mower/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..628b8be3 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"), + ("src/code_mower/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..75393aa6 --- /dev/null +++ b/tests/test_minimal_lineage_packaging.py @@ -0,0 +1,97 @@ +"""Pure import boundary, byte parity, and real isolated init materialization.""" +import ast +import json +from pathlib import Path +import subprocess +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(["python", "-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 From bfb6ba0cf0760bbbd1075951bd39a0d380107035 Mon Sep 17 00:00:00 2001 From: Jeff Huber Date: Tue, 15 Sep 2026 00:20:42 -0700 Subject: [PATCH 2/3] Use current interpreter for isolated lineage packaging test Builder-Provider: codex Builder-Executor: codex_cli --- tests/test_minimal_lineage_packaging.py | 3 ++- 1 file changed, 2 insertions(+), 1 deletion(-) diff --git a/tests/test_minimal_lineage_packaging.py b/tests/test_minimal_lineage_packaging.py index 75393aa6..5671aae7 100644 --- a/tests/test_minimal_lineage_packaging.py +++ b/tests/test_minimal_lineage_packaging.py @@ -3,6 +3,7 @@ import json from pathlib import Path import subprocess +import sys import tempfile import unittest @@ -87,7 +88,7 @@ def test_init_materialized_helper_import_parse_resolve_without_package_or_repo(s assert importlib.util.find_spec("code_mower") is None print("isolated import/parse/resolve/admit passed") ''' - result = subprocess.run(["python", "-I", "-S", "-c", program], cwd=output, + 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") From 6b1642d561ee46987870f492f95d4222179d2374 Mon Sep 17 00:00:00 2001 From: Jeff Huber Date: Tue, 15 Sep 2026 00:45:55 -0700 Subject: [PATCH 3/3] Use committed lineage mirror as standalone package source Builder-Provider: codex Builder-Executor: codex_cli --- code-mower-package-manifest.json | 2 +- src/code_mower/package_manifest.py | 2 +- 2 files changed, 2 insertions(+), 2 deletions(-) diff --git a/code-mower-package-manifest.json b/code-mower-package-manifest.json index 69df2ce2..a0f15c9c 100644 --- a/code-mower-package-manifest.json +++ b/code-mower-package-manifest.json @@ -2064,7 +2064,7 @@ }, { "kind": "core", - "source": "src/code_mower/builder_lineage.py", + "source": "tools/builder_lineage.py", "target": "tools/builder_lineage.py" } ], diff --git a/src/code_mower/package_manifest.py b/src/code_mower/package_manifest.py index 628b8be3..0cdc6c52 100644 --- a/src/code_mower/package_manifest.py +++ b/src/code_mower/package_manifest.py @@ -10,7 +10,7 @@ PACKAGE_FILES = ( ("src/code_mower/builder_lineage.py", "src/code_mower/builder_lineage.py", "core"), - ("src/code_mower/builder_lineage.py", "tools/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"),