diff --git a/CHANGELOG.md b/CHANGELOG.md index 66237ba..f8dc950 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -1,5 +1,34 @@ # Changelog +## v1.3.0 - 2026-10-14 + +### Added + +- Boundary register (`boundary_register.yaml`, format `boundary-register/1`, + shared with the factory and Constrain) plus `ComponentContract.boundary_identifiers`. + Every cross-boundary identifier (service name, audience, credential, principal, + grant/role, table/column, metric type, log field, env/config key, header, + claim) must cite file or observation records bound by SHA-256. Verification + is mechanical: exact `equal` values with near-miss hints, `absent` isolation + proofs, token matching, stale-digest detection, placeholder refusal, and + project-root path confinement. +- `pact boundary verify [project]` verifies the register and exits non-zero on + any error. +- `pact boundary cite --kind file|observation --path --value ` + produces a verified record (sha256, line or observed_at) as JSON and refuses + when the literal is not where the record would claim. +- Interview, shaper, and contract-author prompts require citing a record or + raising an open question; implementer handoffs render registered literals. + +### Changed + +- `validate_all_contracts` now fails when a contract uses a cross-boundary + dependency (`external:` names, `network_call`, `database_*`/`http_*`/ + `external_*` data-access side effects) that the register does not cover, or + when any register entry fails verification. Projects with no cross-boundary + usage are unaffected. +- `pact validate` now exits 1 when the contract gate fails. + ## v1.2.0 - 2026-07-17 ### Added diff --git a/CLAUDE.md b/CLAUDE.md index 18b63e4..474f83b 100644 --- a/CLAUDE.md +++ b/CLAUDE.md @@ -20,6 +20,8 @@ pact checklist # Requirements quality checklist pact production init # Scaffold optional production-readiness pack pact production fingerprint # Print the source fingerprint for evidence pact production validate # Validate production-readiness gate for a Pact-managed project +pact boundary verify [project-dir] # Verify boundary register records (exit 1 on any error) +pact boundary cite --kind file --path --value [--line N] [project-dir] # Produce a record (JSON) pact assess # Architectural assessment (any codebase) pact export-tasks # Export TASKS.md pact handoff # Render/validate handoff brief @@ -87,6 +89,30 @@ passing after source or Pact artifact changes, or after the evidence ages out. This is a deployment gate, not a runtime substitute; `live_validation.yaml` must still prove the running environment matches the evidence being claimed. +### Boundary Register (never assume an identifier; cite the record) + +Every identifier two parties must agree on (service name, audience, credential, +principal, grant/role, table/column, metric type, log field, env/config key, +header, claim) lives in `boundary_register.yaml` (format `boundary-register/1`, +shared with the factory and Constrain) or a contract's `boundary_identifiers`. +Each side (`party`: component id or `external:`; `role`: +defines|grants|consumes|presents|denies; exact `value`) cites >=1 record: a +`file` record (path, line, sha256 of the whole file) or an `observation` record +(path, sha256, command, observed_at). `src/pact/boundary_register.py` verifies +them mechanically (token match, digest, placeholders, path containment, +`equal` vs `absent` relations); `validate_boundary_register()` in +`contracts.py` runs inside `validate_all_contracts()` as hard errors and also +fails any contract that uses an unregistered boundary (`external:` names, +`network_call`, `database_*`/`http_*`/`external_*` data-access side effects). +A stale digest means refresh the register from the source, never hand-edit it. +Prompts (interview, shaper, contract author) share `BOUNDARY_RULE_PROMPT`: +cite the record or raise an open question. Implementer handoffs render the +registered literals in tier 1. `pact boundary verify [project]` exits non-zero +on any error; `pact validate` now exits 1 on a failed gate. Records come from +`pact boundary cite --kind file|observation --path --value +[--line N] [--command ""] [project]` (JSON out; exit 1 if the literal is +not where the record would claim), never typed by hand. + ### Production-Ready Minimalism Pact supports a lean/simplification review lens, but it is not a readiness @@ -271,6 +297,7 @@ src/pact/ mcp_server.py # MCP server (FastMCP transport + PactMCPServer handlers) tool_index.py # External tool enrichment (ctags, cscope, tree-sitter, kindex) production.py # Optional production-readiness pack scaffold + validation + boundary_register.py # boundary-register/1: cross-boundary identifiers verified against records schemas_production.py # Production-readiness artifact schemas # Spec-kit capabilities (task list, analysis, checklist) diff --git a/README.md b/README.md index 85c4181..35f81ab 100644 --- a/README.md +++ b/README.md @@ -291,6 +291,60 @@ types: Tests automatically verify both acceptance and rejection: valid instances pass, invalid inputs fail with appropriate errors, and boundary values behave correctly. Implementations render these as Pydantic models (Python), Zod schemas (TypeScript), or class constructors with validation (JavaScript). +### Boundary Register: Never Assume an Identifier + +Names, tokens, roles, grants, metric types, log fields, headers and service +names that two parties must agree on are boundary identifiers. Pact never +assumes one. Each is entered in `boundary_register.yaml` at the project root +(format `boundary-register/1`, shared with the factory and Constrain), or in a +contract's `boundary_identifiers`, and every side cites a record that the gate +re-verifies mechanically: + +```yaml +schema: "boundary-register/1" +entries: + - id: log-level-field + kind: field # name|endpoint|credential|audience|principal|grant|role|table|column|field|metric|label|env|config-key|topic|queue|bucket|secret|service|header|claim|enum|other + relation: equal # equal | absent (isolation: the subject must NOT appear) + description: The log platform only derives severity from this JSON key + sides: + - party: external:cloud-logging # component id or external: + role: defines # defines|grants|consumes|presents|denies + value: severity # exact literal, case-sensitive + records: + - {kind: observation, path: evidence/log_entry.json, sha256: , + command: "gcloud logging read ... --limit 1 --format json", + observed_at: "2026-10-14T12:00:00Z"} + - party: api + role: consumes + value: severity + records: + - {kind: file, path: src/api/logging.py, line: 12, sha256: } +``` + +Every violation is an error, never a warning: mismatched values (with a +"near-miss" note when they differ only by case or separators), a side with no +records (UNVERIFIED), a stale digest (refresh the register from the source; +never hand-edit it to fit), a cited line that does not contain the value as a +token (`mea-auth` does not match `mea-auth-runtime`), a placeholder value, a +path outside the project root, and, for `absent`, a denies record that +contains the subject. A boundary that is used must be registered: contracts +that declare `external:` dependencies/requires/targets, `network_call` +side effects, or `database_*`/`http_*`/`external_*` data-access side effects +fail `pact validate` (and the decomposer gate and MCP `pact_validate`) until a +register entry covers them. Projects with no cross-boundary usage are +unaffected. Contract and interview prompts tell the model to raise an open +question rather than guess; registered literals are rendered into every +implementer handoff. Run `pact boundary verify [project]` to check on its own. + +Records are produced, never typed. `pact boundary cite --kind file|observation +--path --value [--line N] [--command ""] [project]` reads +the file under the project root (same path safety as the verifier), hashes the +whole file, finds the first line containing the value as a token (or confirms +`--line`), and for observations requires `--command` and stamps `observed_at` +in UTC. It prints the record as JSON and exits 1, writing nothing, when the +literal is not where the record would claim. + ## Audit Repo Separation Pact supports a two-repo separation-of-privilege model where the coding agent and auditing agent operate in different repositories: @@ -388,6 +442,8 @@ Either, neither, or both. Defaults: both off (sequential, single-attempt). | `pact production status ` | Show production-readiness status without blocking | | `pact production fingerprint ` | Print the current source fingerprint for production evidence | | `pact production validate ` | Validate a Pact-managed production gate and exit non-zero on blockers | +| `pact boundary verify [project]` | Verify the boundary register against its records; exit non-zero on any error | +| `pact boundary cite --kind file\|observation --path --value [project]` | Produce a verified record (sha256, line or observed_at) as JSON; exit 1 if the literal is not there | | `pact spec apply ` | Apply an AI-authored JSON/YAML build spec | | `pact spec show ` | Normalize and inspect an AI-authored build spec | | `pact agent spec-author [--apply]` | Run one constrained spec-author agent | diff --git a/REVIEW.md b/REVIEW.md new file mode 100644 index 0000000..1911dee --- /dev/null +++ b/REVIEW.md @@ -0,0 +1,37 @@ +# REVIEW.md -- Pact + +Guidance for automated review of Pact changes. Architecture and commands: `CLAUDE.md`. + +## Always check + +- Mechanical gates stay mechanical: no LLM call inside `contracts.py`, + `boundary_register.py`, `production_validation.py`, `analyzer.py`, or `checklist_gen.py`. +- New validation that is meant to block returns errors through `validate_all_contracts` + (or the production gate); it is never downgraded to a logged warning. +- Cross-boundary identifiers (names, tokens, roles, grants, metric/log fields, headers) + are never assumed: they come from `boundary_register.yaml` / `boundary_identifiers` + with records that verify. A register edited to fit a stale digest is a defect. +- Schema changes to `ComponentContract` and other persisted models load existing + `interface.json` files (new fields have defaults; malformed LLM data must not make a + contract unloadable). +- New behavior has pytest coverage under `tests/` that fails without the change; CLI + commands that gate exit non-zero on failure. +- Agent prompts keep the context fence as the first line and stay concise (no rule lists + appended to every prompt without need). +- Record/evidence paths are confined to the project root (no absolute paths, no `..`, + no symlink escape). + +## Style + +- Several source files use CRLF or mixed line endings; edits must preserve them (no + whole-file EOL churn in diffs). +- Prefer stdlib; runtime dependencies are pydantic, pyyaml, pytest (see `pyproject.toml`). +- Docstrings state the why and the invariant a module upholds. +- Error messages name the artifact, entry, and reason so an operator can act without + reading code. + +## Skip + +- `dist/`, `.pact/`, `.kin/`, generated `standards.json`, `tasks.json`, `TASKS.md`. +- `tests/smoke/` (generated by `pact adopt`). +- Local `config.yaml` operator tuning. diff --git a/pyproject.toml b/pyproject.toml index dde06ab..28d74c1 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -1,6 +1,6 @@ [project] name = "pact-agents" -version = "1.2.0" +version = "1.3.0" description = "Contract-first multi-agent software engineering. Contracts before code. Tests as law. Agents that can't cheat." readme = "README.md" license = {text = "MIT"} diff --git a/src/pact/__init__.py b/src/pact/__init__.py index f740a57..1cb694c 100644 --- a/src/pact/__init__.py +++ b/src/pact/__init__.py @@ -1,3 +1,3 @@ """Pact — contract-first multi-agent software engineering.""" -__version__ = "1.2.0" +__version__ = "1.3.0" diff --git a/src/pact/agents/code_author.py b/src/pact/agents/code_author.py index 0470a82..1487dd0 100644 --- a/src/pact/agents/code_author.py +++ b/src/pact/agents/code_author.py @@ -165,6 +165,7 @@ async def author_code( language: str = "python", strategic_context: str = "", processing_register: str = "", + boundary_entries: list | None = None, ) -> ImplementationResult: """Generate implementation code following the Research-First Protocol. @@ -269,6 +270,7 @@ async def author_code( log_key_preamble=log_preamble, strategic_context=strategic_context, processing_register=processing_register, + boundary_entries=boundary_entries, ) # The handoff brief is the largest cacheable block diff --git a/src/pact/agents/contract_author.py b/src/pact/agents/contract_author.py index 82ce4fd..496e028 100644 --- a/src/pact/agents/contract_author.py +++ b/src/pact/agents/contract_author.py @@ -12,6 +12,7 @@ from typing import TYPE_CHECKING from pact.agents.base import AgentBase +from pact.boundary_register import BOUNDARY_RULE_PROMPT from pact.readiness import ReadinessProfile from pact.agents.research import plan_and_evaluate, research_phase from pact.schemas import ( @@ -46,7 +47,9 @@ classification and affected fields. - authority: declare data domain patterns this component owns (empty list if none). If domains is non-empty, rationale is required explaining why this component is - authoritative for those domains.""" + authoritative for those domains. + +""" + BOUNDARY_RULE_PROMPT async def author_contract( @@ -62,9 +65,14 @@ async def author_contract( processing_register: str = "", type_registry: "TypeRegistry | None" = None, readiness_profile: ReadinessProfile | None = None, + boundary_register_context: str = "", ) -> tuple[ComponentContract, ResearchReport, PlanEvaluation]: """Generate a ComponentContract following the Research-First Protocol. + ``boundary_register_context`` is the rendered project boundary register + (see ``pact.boundary_register.render_register_for_prompt``): the only + source of identifier values the author may cite. + Returns: Tuple of (contract, research_report, plan_evaluation). """ @@ -172,6 +180,8 @@ async def author_contract( cache_parts.append(dep_stubs) if decisions_summary: cache_parts.append(decisions_summary) + if boundary_register_context: + cache_parts.append(boundary_register_context) cache_prefix = "\n\n".join(cache_parts) # Dynamic prompt (component-specific) @@ -202,7 +212,14 @@ async def author_contract( (not vague phrases like "handles data" — describe the exact data and purpose), and list side_effects with type, classification, fields, and rationale - authority: set domains this component owns (empty list if non-authoritative), - with rationale if domains is non-empty""" + with rationale if domains is non-empty +- boundary_identifiers: for every identifier this component shares with another + party (other components or external:), add a boundary-register/1 entry + whose sides cite verifiable records. Use only values you can cite from the + context above; never invent one. If you cannot cite it, do not guess: leave + the entry out (the validation gate then refuses the unregistered boundary until + a human supplies the record) and state what is missing in an invariant that + starts with 'OPEN QUESTION:'""" contract, in_tok, out_tok = await agent.assess_cached( ComponentContract, prompt, CONTRACT_SYSTEM, cache_prefix=cache_prefix, diff --git a/src/pact/agents/shaper.py b/src/pact/agents/shaper.py index f606b60..f6ab262 100644 --- a/src/pact/agents/shaper.py +++ b/src/pact/agents/shaper.py @@ -16,6 +16,7 @@ import logging from pact.agents.base import AgentBase +from pact.boundary_register import BOUNDARY_RULE_PROMPT from pact.schemas_shaping import ( Appetite, ShapingPitch, @@ -294,7 +295,12 @@ async def close(self) -> None: await self._agent.close() -SYSTEM_PROMPT = "You are a product shaper following Shape Up methodology." +SYSTEM_PROMPT = ( + "You are a product shaper following Shape Up methodology.\n\n" + + BOUNDARY_RULE_PROMPT + + "\nTreat every uncited boundary identifier as a rabbit hole: name it and " + "the system that owns it, never fill it in." +) SYSTEM_PROMPT_ENRICHMENT = ( "You are a product shaper performing deep analysis. " diff --git a/src/pact/boundary_register.py b/src/pact/boundary_register.py new file mode 100644 index 0000000..c47da2c --- /dev/null +++ b/src/pact/boundary_register.py @@ -0,0 +1,712 @@ +"""Boundary register: cross-boundary identifiers backed by verifiable records. + +A boundary identifier is any literal two parties must agree on for a call +across a boundary to work: a service name, an audience, a database role, a +table, a metric resource type, a log field, a header, a grant. When one side +writes ``"level"`` and the other reads ``"severity"``, or a gateway queries a +table its role was never granted, nothing inside either component is wrong; +the defect lives on the boundary. Pact refuses to assume such values. + +The register (format ``boundary-register/1``, shared with the factory and +Constrain) names every such identifier once, states which party defines, +grants, consumes, presents, or denies it, and pins each side to a record the +gate can re-verify mechanically: a file line or a captured observation, both +bound by SHA-256 of the file bytes. No LLM participates in verification. + +Invariants upheld here: + +- Every side cites at least one record; an unverified side is an error. +- A record is verified against the bytes on disk. A digest mismatch is a stale + record, reported by file name: refresh the register from the source; never + hand-edit it to fit. +- ``relation: equal`` requires one authority side (defines|grants) and one + user side (consumes|presents), with byte-identical values. +- ``relation: absent`` proves isolation: the subject value must NOT occur as + a token in any denies-side record. +- Placeholder values are refused; a value nobody can cite is an open question, + not a register entry. +""" + +from __future__ import annotations + +import hashlib +import re +from datetime import datetime, timezone +from pathlib import Path, PurePosixPath, PureWindowsPath +from typing import Any + +import yaml +from pydantic import BaseModel, ConfigDict, Field + +SCHEMA_ID = "boundary-register/1" +REGISTER_FILENAME = "boundary_register.yaml" + +ENTRY_KINDS = frozenset({ + "name", "endpoint", "credential", "audience", "principal", "grant", "role", + "table", "column", "field", "metric", "label", "env", "config-key", "topic", + "queue", "bucket", "secret", "service", "header", "claim", "enum", "other", +}) +RELATIONS = frozenset({"equal", "absent"}) +SIDE_ROLES = frozenset({"defines", "grants", "consumes", "presents", "denies"}) +AUTHORITY_ROLES = frozenset({"defines", "grants"}) +USER_ROLES = frozenset({"consumes", "presents"}) +RECORD_KINDS = frozenset({"file", "observation"}) + +ID_RE = re.compile(r"^[A-Za-z0-9][A-Za-z0-9._-]*$") +SHA256_RE = re.compile(r"^[0-9a-fA-F]{64}$") +RFC3339_RE = re.compile( + r"^\d{4}-\d{2}-\d{2}[Tt ]\d{2}:\d{2}:\d{2}(?:\.\d+)?(?:[Zz]|[+-]\d{2}:\d{2})$" +) +EXTERNAL_PREFIX = "external:" + +# A value nobody looked up (boundary-register/1 shared rule, parity with the +# factory and Constrain): the whole value is a stand-in word, or it contains an +# angle placeholder. "unknown", "none", "null", "n/a" are real enum members. +_PLACEHOLDER_RE = re.compile( + r"^(?:todo|tbd|tbc|assumed|placeholder|changeme|x{3,}|\?+)$", + re.IGNORECASE, +) +_ANGLE_PLACEHOLDER_RE = re.compile(r"<[^<>\s]+>") + +_TOKEN_CHARS = frozenset("ABCDEFGHIJKLMNOPQRSTUVWXYZabcdefghijklmnopqrstuvwxyz0123456789_-") +_SEPARATORS_RE = re.compile(r"[-_.\s]") +MAX_RECORD_BYTES = 16 * 1024 * 1024 + + +# ── Schema (lenient models; the validator owns semantics) ───────────── +# +# Contracts embed entries under ``boundary_identifiers``. The models accept +# loosely-typed data so a malformed LLM entry never makes the whole contract +# unloadable; ``verify_entries`` then reports every defect as an error. + + +class BoundaryRecord(BaseModel): + """Evidence that a side's value is what the source says it is.""" + model_config = ConfigDict(extra="allow") + kind: str = "" + path: str = "" + line: int | str | None = None + sha256: str = "" + command: str = "" + observed_at: str = "" + + +class BoundarySide(BaseModel): + """One party's view of the identifier, with its records.""" + model_config = ConfigDict(extra="allow") + party: str = "" + role: str = "" + value: str = "" + records: list[BoundaryRecord] = [] + + +class BoundaryEntry(BaseModel): + """One identifier that must agree (or must be absent) across a boundary.""" + model_config = ConfigDict(extra="allow") + id: str = "" + kind: str = "" + relation: str = "" + description: str = "" + sides: list[BoundarySide] = [] + + +class BoundaryRegister(BaseModel): + """The project-level register file.""" + model_config = ConfigDict(extra="allow", populate_by_name=True) + schema_id: str = Field(default=SCHEMA_ID, alias="schema") + entries: list[BoundaryEntry] = [] + + +# ── Matching helpers ────────────────────────────────────────────────── + + +def contains_token(haystack: str, value: str) -> bool: + """True if ``value`` occurs in ``haystack`` as a whole token. + + A value that begins (ends) with an identifier character ([A-Za-z0-9_-]) must + not be preceded (followed) by one, so ``svc-auth`` is not found inside + ``svc-auth-runtime`` and ``level`` is found inside ``"level":``. A value that + itself contains ``.`` or ``/`` is dotted or path-shaped, so that separator + followed by an identifier character continues it: ``foo.bar`` is not found + inside ``foo.bar.baz``, nor ``/health`` inside ``/health/ready``. An undotted + name still matches after an attribute dot (``entry.severity``). An edge that + is punctuation in the value matches unconditionally at that edge. + """ + if not value: + return False + joiners = {char for char in "./" if char in value} + + def continues(at: int, step: int) -> bool: + if not 0 <= at < len(haystack): + return False + char = haystack[at] + if char in _TOKEN_CHARS: + return True + beyond = at + step + return char in joiners and 0 <= beyond < len(haystack) and haystack[beyond] in _TOKEN_CHARS + + start = 0 + while True: + index = haystack.find(value, start) + if index < 0: + return False + end = index + len(value) + before = value[0] not in _TOKEN_CHARS or not continues(index - 1, -1) + after = value[-1] not in _TOKEN_CHARS or not continues(end, 1) + if before and after: + return True + start = index + 1 + + +def is_placeholder(value: str) -> bool: + """True if ``value`` is a stand-in for a literal nobody looked up.""" + text = value.strip() + return bool(_PLACEHOLDER_RE.match(text) or _ANGLE_PLACEHOLDER_RE.search(text)) + + +def near_miss(a: str, b: str) -> str: + """Describe how two differing values nearly match, or return ``""``.""" + if a == b: + return "" + if a.lower() == b.lower(): + return "differ only by case" + sa, sb = _SEPARATORS_RE.sub("", a), _SEPARATORS_RE.sub("", b) + if sa == sb: + return "differ only by separators (-, _, ., space)" + if sa.lower() == sb.lower(): + return "differ only by case and separators (-, _, ., space)" + return "" + + +def is_external_party(party: str) -> bool: + return party.startswith(EXTERNAL_PREFIX) and len(party) > len(EXTERNAL_PREFIX) + + +# ── Loading ─────────────────────────────────────────────────────────── + + +def register_path(project_dir: str | Path) -> Path: + return Path(project_dir) / REGISTER_FILENAME + + +def load_register(project_dir: str | Path) -> tuple[list[dict], list[str]]: + """Load ``boundary_register.yaml`` from the project root. + + Returns ``(entries, errors)``. A missing file is not an error here (the + gate decides whether a register is required); an unreadable or malformed + file is. + """ + path = register_path(project_dir) + if not path.exists(): + return [], [] + label = REGISTER_FILENAME + try: + data = yaml.safe_load(path.read_text(encoding="utf-8")) + except (OSError, UnicodeDecodeError, yaml.YAMLError) as exc: + return [], [f"{label}: could not be read as YAML: {exc}"] + if data is None: + data = {} + if not isinstance(data, dict): + return [], [f"{label}: top level must be a mapping with 'schema' and 'entries'"] + errors: list[str] = [] + schema = data.get("schema") + if schema != SCHEMA_ID: + errors.append(f"{label}: schema must be '{SCHEMA_ID}' (got {schema!r})") + entries = data.get("entries", []) + if entries is None: + entries = [] + if not isinstance(entries, list): + return [], errors + [f"{label}: 'entries' must be a list"] + return entries, errors + + +def load_register_entries(project_dir: str | Path | None) -> list[BoundaryEntry]: + """Best-effort typed load for rendering. Never raises; the gate reports defects.""" + if project_dir is None: + return [] + raw, _ = load_register(project_dir) + out: list[BoundaryEntry] = [] + for item in raw: + if isinstance(item, dict): + try: + out.append(BoundaryEntry.model_validate(item)) + except Exception: + continue + return out + + +def _as_dict(entry: Any) -> Any: + if isinstance(entry, BaseModel): + return entry.model_dump(by_alias=True, exclude_none=True) + return entry + + +# ── Verification ────────────────────────────────────────────────────── + + +class _FileCache: + """Reads each cited file once per verification run.""" + + def __init__(self, root: Path | None) -> None: + self.root = root.resolve() if root is not None else None + self._cache: dict[str, tuple[bytes | None, str]] = {} + + def read(self, rel: str) -> tuple[bytes | None, str]: + """Return ``(bytes, "")`` or ``(None, reason)``.""" + if rel in self._cache: + return self._cache[rel] + result = self._read(rel) + self._cache[rel] = result + return result + + def _read(self, rel: str) -> tuple[bytes | None, str]: + if self.root is None: + return None, "no project root is available to verify records against" + if not isinstance(rel, str) or not rel.strip(): + return None, "record path is empty" + posix, win = PurePosixPath(rel), PureWindowsPath(rel) + if posix.is_absolute() or win.is_absolute() or win.drive or rel.startswith("~"): + return None, f"record path '{rel}' must be relative to the project root" + if ".." in posix.parts or ".." in win.parts: + return None, f"record path '{rel}' must not contain '..'" + candidate = self.root / rel + try: + resolved = candidate.resolve(strict=True) + except (OSError, RuntimeError): + return None, f"record file '{rel}' does not exist" + if resolved != self.root and self.root not in resolved.parents: + return None, f"record path '{rel}' resolves outside the project root" + if not resolved.is_file(): + return None, f"record path '{rel}' is not a regular file" + try: + with resolved.open("rb") as fh: + data = fh.read(MAX_RECORD_BYTES + 1) + except OSError as exc: + return None, f"record file '{rel}' could not be read: {exc}" + if len(data) > MAX_RECORD_BYTES: + return None, f"record file '{rel}' exceeds {MAX_RECORD_BYTES // (1024 * 1024)} MiB" + return data, "" + + +def _verify_record( + rec: Any, + *, + where: str, + value: str, + files: _FileCache, + mode: str, +) -> list[str]: + """Verify one record. + + ``mode`` is ``"present"`` (the side's value must occur as a token at the + cited place) or ``"absent"`` (``value`` is the subject and must NOT occur + anywhere in the record's bytes). + """ + if not isinstance(rec, dict): + return [f"{where}: record must be a mapping"] + errors: list[str] = [] + kind = rec.get("kind") + if kind not in RECORD_KINDS: + return [f"{where}: record kind must be one of {sorted(RECORD_KINDS)} (got {kind!r})"] + path = rec.get("path") + sha = rec.get("sha256") + if not isinstance(sha, str) or not SHA256_RE.match(sha): + errors.append(f"{where}: sha256 must be 64 hex characters (got {sha!r})") + + line_no: int | None = None + if kind == "file": + line = rec.get("line") + if isinstance(line, bool) or not isinstance(line, int) or line < 1: + errors.append(f"{where}: file record needs 'line' as a 1-based integer (got {line!r})") + else: + line_no = line + else: + command = rec.get("command") + if not isinstance(command, str) or not command.strip(): + errors.append(f"{where}: observation record needs the exact 'command' that produced it") + observed_at = rec.get("observed_at") + if not _valid_rfc3339(observed_at): + errors.append( + f"{where}: observation record needs 'observed_at' as an RFC3339 timestamp " + f"(got {observed_at!r})" + ) + + data, reason = files.read(path if isinstance(path, str) else "") + if data is None: + errors.append(f"{where}: {reason}") + return errors + + if isinstance(sha, str) and SHA256_RE.match(sha): + actual = hashlib.sha256(data).hexdigest() + if actual != sha.lower(): + errors.append( + f"{where}: STALE record for file '{path}': sha256 on disk is {actual}, " + f"register says {sha.lower()}. Refresh the register from the source; " + f"never hand-edit it to fit" + ) + return errors + else: + return errors + + text = data.decode("utf-8", errors="replace") + if mode == "absent": + if contains_token(text, value): + errors.append( + f"{where}: isolation violated: subject {value!r} occurs as a token in " + f"'{path}', which this side cites as proof of absence" + ) + if kind == "file" and line_no is not None: + count = len(text.split("\n")) + if line_no > count: + errors.append(f"{where}: line {line_no} is beyond the end of '{path}' ({count} lines)") + return errors + + if kind == "file": + if line_no is None: + return errors + lines = text.split("\n") + if line_no > len(lines): + errors.append(f"{where}: line {line_no} is beyond the end of '{path}' ({len(lines)} lines)") + return errors + cited = lines[line_no - 1].rstrip("\r") + if not contains_token(cited, value): + errors.append( + f"{where}: '{path}' line {line_no} does not contain {value!r} as a token " + f"(line reads: {cited.strip()[:160]!r})" + ) + else: + if not contains_token(text, value): + errors.append(f"{where}: observation '{path}' does not contain {value!r} as a token") + return errors + + +def _valid_rfc3339(value: Any) -> bool: + if not isinstance(value, str) or not RFC3339_RE.match(value): + return False + normalized = value.replace("z", "Z").replace("Z", "+00:00").replace(" ", "T").replace("t", "T") + try: + datetime.fromisoformat(normalized) + except ValueError: + return False + return True + + +def verify_entries( + entries: list[Any], + project_root: str | Path | None, + *, + known_components: set[str] | None = None, + origin: str = "", +) -> list[str]: + """Verify register entries. Returns errors (empty list = every entry holds). + + Each error names the entry id, the side (index, party, role), and the + reason. ``known_components`` (when given) rejects non-external parties + that are not components of the project: a misspelled party is itself a + boundary-name mismatch. + """ + files = _FileCache(Path(project_root) if project_root is not None else None) + errors: list[str] = [] + seen: dict[str, str] = {} + prefix = f"{origin}: " if origin else "" + + for index, raw_entry in enumerate(entries): + entry = _as_dict(raw_entry) + if not isinstance(entry, dict): + errors.append(f"{prefix}boundary entry #{index}: must be a mapping") + continue + eid = entry.get("id") + label = f"{prefix}boundary[{eid if isinstance(eid, str) and eid else f'#{index}'}]" + if not isinstance(eid, str) or not ID_RE.match(eid): + errors.append(f"{label}: id must match [A-Za-z0-9][A-Za-z0-9._-]* (got {eid!r})") + elif eid in seen: + errors.append(f"{label}: duplicate id (also defined in {seen[eid]})") + else: + seen[eid] = origin or "register" + + kind = entry.get("kind") + if kind not in ENTRY_KINDS: + errors.append(f"{label}: kind must be one of {sorted(ENTRY_KINDS)} (got {kind!r})") + relation = entry.get("relation") + if relation not in RELATIONS: + errors.append(f"{label}: relation must be 'equal' or 'absent' (got {relation!r})") + description = entry.get("description") + if not isinstance(description, str) or not description.strip(): + errors.append(f"{label}: description must say why the sides must agree") + + sides = entry.get("sides") + if not isinstance(sides, list) or not sides: + errors.append(f"{label}: needs at least one side") + continue + + well_formed: list[tuple[int, dict, str]] = [] # (index, side, side label) + for s_index, side in enumerate(sides): + if not isinstance(side, dict): + errors.append(f"{label} side[{s_index}]: must be a mapping") + continue + party = side.get("party") + role = side.get("role") + value = side.get("value") + s_label = f"{label} side[{s_index} {party}/{role}]" + ok = True + if not isinstance(party, str) or not party.strip(): + errors.append(f"{s_label}: party is required (component id or 'external:')") + ok = False + elif party.startswith(EXTERNAL_PREFIX): + if not is_external_party(party): + errors.append(f"{s_label}: external party must name the system ('external:')") + ok = False + elif known_components is not None and party not in known_components: + errors.append( + f"{s_label}: party {party!r} is not a component of this project " + f"(known: {sorted(known_components)}); use 'external:' for outside systems" + ) + if role not in SIDE_ROLES: + errors.append(f"{s_label}: role must be one of {sorted(SIDE_ROLES)} (got {role!r})") + ok = False + if not isinstance(value, str) or not value: + errors.append(f"{s_label}: value must be the exact literal (non-empty string)") + ok = False + elif is_placeholder(value): + errors.append( + f"{s_label}: value {value!r} is a placeholder. Never assume an identifier: " + f"cite the record, or raise an open question" + ) + ok = False + records = side.get("records") + if not isinstance(records, list) or not records: + errors.append( + f"{s_label}: UNVERIFIED: no records. Every side must cite at least one " + f"file or observation record" + ) + ok = False + if ok: + well_formed.append((s_index, side, s_label)) + + if relation == "equal": + errors.extend(_check_equal(label, sides, well_formed, files)) + elif relation == "absent": + errors.extend(_check_absent(label, sides, well_formed, files)) + + return errors + + +def _roles(sides: list) -> list[str]: + return [s.get("role") for s in sides if isinstance(s, dict)] + + +def _check_equal(label: str, sides: list, well_formed: list, files: _FileCache) -> list[str]: + errors: list[str] = [] + roles = _roles(sides) + if "denies" in roles: + errors.append(f"{label}: role 'denies' is only valid with relation 'absent'") + if not any(r in AUTHORITY_ROLES for r in roles): + errors.append(f"{label}: relation 'equal' needs a side with role defines or grants") + if not any(r in USER_ROLES for r in roles): + errors.append(f"{label}: relation 'equal' needs a side with role consumes or presents") + + values = [ + (s.get("party"), s.get("role"), s.get("value")) + for s in sides + if isinstance(s, dict) and isinstance(s.get("value"), str) and s.get("value") + ] + distinct = sorted({v for _, _, v in values}) + if len(distinct) > 1: + detail = "; ".join(f"{p}/{r}={v!r}" for p, r, v in values) + msg = f"{label}: MISMATCH: sides disagree ({detail})" + hints = {near_miss(distinct[0], other) for other in distinct[1:]} + hints.discard("") + if hints: + msg += f". near-miss: values {', '.join(sorted(hints))}" + errors.append(msg) + + for s_index, side, s_label in well_formed: + for r_index, rec in enumerate(side["records"]): + errors.extend(_verify_record( + rec, where=f"{s_label} record[{r_index}]", + value=side["value"], files=files, mode="present", + )) + return errors + + +def _check_absent(label: str, sides: list, well_formed: list, files: _FileCache) -> list[str]: + errors: list[str] = [] + roles = _roles(sides) + presents = [s for s in sides if isinstance(s, dict) and s.get("role") == "presents"] + if len(presents) != 1: + errors.append( + f"{label}: relation 'absent' needs exactly one 'presents' side (the subject); " + f"found {len(presents)}" + ) + if "denies" not in roles: + errors.append(f"{label}: relation 'absent' needs at least one 'denies' side") + others = sorted({r for r in roles if r not in {"presents", "denies"} and isinstance(r, str)}) + if others: + errors.append(f"{label}: relation 'absent' allows only presents and denies sides (got {others})") + + subject = presents[0].get("value") if len(presents) == 1 else None + for s_index, side, s_label in well_formed: + role = side.get("role") + for r_index, rec in enumerate(side["records"]): + where = f"{s_label} record[{r_index}]" + if role == "presents": + errors.extend(_verify_record( + rec, where=where, value=side["value"], files=files, mode="present", + )) + elif role == "denies" and isinstance(subject, str) and subject: + errors.extend(_verify_record( + rec, where=where, value=subject, files=files, mode="absent", + )) + return errors + + +# ── Producing records (pact boundary cite) ─────────────────────────── + + +class CiteError(ValueError): + """The literal is not where the record would claim; no record is produced.""" + + +def cite_record( + project_root: str | Path, + *, + kind: str, + path: str, + value: str, + line: int | None = None, + command: str | None = None, + now: datetime | None = None, +) -> dict: + """Produce a record from the source so nobody types a digest or line. + + Reads ``path`` under ``project_root`` with the verifier's path safety, + hashes the whole file, and confirms ``value`` is where the record will + claim: on the given line, or (``line`` omitted) on the first line that + contains it as a token; for observations, anywhere in the file. The + returned record round-trips through ``verify_entries``. Raises + ``CiteError`` instead of producing a record that would not verify. + """ + if kind not in RECORD_KINDS: + raise CiteError(f"kind must be one of {sorted(RECORD_KINDS)} (got {kind!r})") + if not isinstance(value, str) or not value: + raise CiteError("value must be the exact non-empty literal") + if is_placeholder(value): + raise CiteError(f"value {value!r} is a placeholder; cite the real literal") + if kind == "observation" and (not command or not command.strip()): + raise CiteError("observation records need --command: the exact command or query that produced the file") + + data, reason = _FileCache(Path(project_root)).read(path) + if data is None: + raise CiteError(reason) + digest = hashlib.sha256(data).hexdigest() + text = data.decode("utf-8", errors="replace") + + if kind == "file": + lines = text.split("\n") + if line is None: + for number, content in enumerate(lines, start=1): + if contains_token(content.rstrip("\r"), value): + line = number + break + else: + raise CiteError(f"{value!r} does not occur as a token on any line of '{path}'") + else: + if line < 1 or line > len(lines): + raise CiteError(f"line {line} is outside '{path}' ({len(lines)} lines)") + if not contains_token(lines[line - 1].rstrip("\r"), value): + raise CiteError(f"'{path}' line {line} does not contain {value!r} as a token") + return {"kind": "file", "path": path, "line": line, "sha256": digest} + + if not contains_token(text, value): + raise CiteError(f"observation '{path}' does not contain {value!r} as a token") + stamp = (now or datetime.now(timezone.utc)).astimezone(timezone.utc) + return { + "kind": "observation", + "path": path, + "sha256": digest, + "command": command.strip(), + "observed_at": stamp.strftime("%Y-%m-%dT%H:%M:%SZ"), + } + + +# ── Rendering for implementers ──────────────────────────────────────── + + +def entries_for_party(entries: list[Any], party: str) -> list[BoundaryEntry]: + out: list[BoundaryEntry] = [] + for raw in entries: + try: + entry = raw if isinstance(raw, BoundaryEntry) else BoundaryEntry.model_validate(raw) + except Exception: + continue + if any(side.party == party for side in entry.sides): + out.append(entry) + return out + + +def render_boundary_identifiers(component_id: str, entries: list[Any]) -> str: + """Render the registered literals a component must use, with their records.""" + mine = entries_for_party(entries, component_id) + if not mine: + return "" + lines = [ + f"# Boundary identifiers for: {component_id}", + "# Registered and verified against records. Use these exact literals;", + "# never rename, re-case, or guess. If one looks wrong, stop and raise it.", + ] + for entry in mine: + lines.append(f"- {entry.id} ({entry.kind}, {entry.relation}): {entry.description}") + for side in entry.sides: + cite = "" + if side.records: + rec = side.records[0] + cite = f"{rec.path}:{rec.line}" if rec.kind == "file" else f"{rec.path} [observed]" + marker = " <- you" if side.party == component_id else "" + lines.append(f" {side.party} {side.role}: {side.value!r} ({cite}){marker}") + return "\n".join(lines) + + +# ── Prompt rule (one authority for the wording every author sees) ───── + +BOUNDARY_RULE_PROMPT = """Cross-boundary identifiers: any literal another party must agree on +(service name, URL/endpoint, audience, credential/secret name, principal, IAM +grant or role, table/column, log/metric field or label, metric resource type, +env var, config key, topic/queue/bucket, header, claim, enum value) is a +boundary identifier. Never invent or assume one. Record each in the boundary +register (boundary_register.yaml, format boundary-register/1, or the contract's +boundary_identifiers) with: id, kind, relation (equal, or absent for isolation), +description, and one side per party (party = component id or external:; +role = defines|grants|consumes|presents|denies; value = the exact literal), +each side citing a record (kind file: path, line, sha256; or kind observation: +path, sha256, command, observed_at) that proves the value. For relation absent, +the presents side's value is the subject, and each denies side's value names the +policy or resource whose record must exclude the subject. Records are produced +by `pact boundary cite --kind file|observation --path --value +[--line N] [--command ""]`, never typed: never write a digest or line +number by hand. If you cannot cite a +record for a value, do not guess and do not write a placeholder: raise it as an +open question for the owner of that system.""" + + +def render_register_for_prompt(entries: list[Any]) -> str: + """Render the full register (with records) so authors cite, never retype. + + The YAML here is the source of truth for values and digests: an author that + needs an existing entry copies it verbatim rather than re-deriving it. + """ + dumped = [] + for raw in entries: + entry = raw if isinstance(raw, BoundaryEntry) else None + if entry is None: + try: + entry = BoundaryEntry.model_validate(raw) + except Exception: + continue + dumped.append(entry.model_dump(exclude_none=True, exclude_defaults=True)) + if not dumped: + return "" + body = yaml.safe_dump({"schema": SCHEMA_ID, "entries": dumped}, sort_keys=False) + return ( + "Registered boundary identifiers (verified against records; reuse these " + "exact values and records verbatim):\n```yaml\n" + body + "```" + ) diff --git a/src/pact/cli.py b/src/pact/cli.py index 44c98c7..93e9115 100644 --- a/src/pact/cli.py +++ b/src/pact/cli.py @@ -29,6 +29,8 @@ pact incidents List active/recent incidents pact incident Show incident details + diagnostic report pact production ... Manage production-readiness artifact pack + pact boundary verify [project-dir] Verify the boundary register against its records + pact boundary cite --kind ... [dir] Produce a verified record for a literal (JSON) pact agent ... Run constrained Pact agents pact ci Generate GitHub Actions CI workflow pact deploy Generate baton.yaml topology config @@ -306,6 +308,32 @@ def main() -> None: ) p_production_fingerprint.add_argument("project_dir", help="Project directory path") + # boundary register (cross-boundary identifiers backed by records) + p_boundary = subparsers.add_parser( + "boundary", + help="Verify cross-boundary identifiers against their records", + ) + boundary_sub = p_boundary.add_subparsers(dest="boundary_command") + p_boundary_verify = boundary_sub.add_parser( + "verify", + help="Verify boundary_register.yaml and contract boundary_identifiers (exit 1 on any error)", + ) + p_boundary_verify.add_argument( + "project_dir", nargs="?", default=".", help="Project directory path (default: .)", + ) + p_boundary_cite = boundary_sub.add_parser( + "cite", + help="Produce a verified record (sha256, line) for a literal; prints JSON, exit 1 if not found", + ) + p_boundary_cite.add_argument("--kind", required=True, choices=["file", "observation"]) + p_boundary_cite.add_argument("--path", required=True, help="File path relative to the project root") + p_boundary_cite.add_argument("--value", required=True, help="Exact literal the record must contain") + p_boundary_cite.add_argument("--line", type=int, default=None, help="1-based line (file kind; auto-detected if omitted)") + p_boundary_cite.add_argument("--command", dest="record_command", default=None, help="Exact command/query that produced the observation file") + p_boundary_cite.add_argument( + "project_dir", nargs="?", default=".", help="Project directory path (default: .)", + ) + # directive p_directive = subparsers.add_parser("directive", help="Send structured directive to daemon") p_directive.add_argument("project_dir", help="Project directory path") @@ -544,6 +572,8 @@ def main() -> None: cmd_checklist(args) elif args.command == "production": cmd_production(args) + elif args.command == "boundary": + cmd_boundary(args) elif args.command == "directive": cmd_directive(args) elif args.command == "export-tasks": @@ -1596,7 +1626,7 @@ def cmd_validate(args: argparse.Namespace) -> None: contracts = project.load_all_contracts() test_suites = project.load_all_test_suites() - gate = validate_all_contracts(tree, contracts, test_suites) + gate = validate_all_contracts(tree, contracts, test_suites, project.project_dir) if gate.passed: print("Validation PASSED") @@ -1604,6 +1634,7 @@ def cmd_validate(args: argparse.Namespace) -> None: print(f"Validation FAILED: {gate.reason}") for detail in gate.details: print(f" - {detail}") + raise SystemExit(1) def cmd_design(args: argparse.Namespace) -> None: @@ -2024,7 +2055,9 @@ def cmd_doctor(args: argparse.Namespace) -> None: if contracts: from pact.contracts import validate_all_contracts test_suites = project.load_all_test_suites() - gate = validate_all_contracts(tree, contracts, test_suites) + gate = validate_all_contracts( + tree, contracts, test_suites, project.project_dir, + ) if gate.passed: checks.append(("OK", "All contracts validated", "")) else: @@ -2429,6 +2462,57 @@ def cmd_production(args: argparse.Namespace) -> None: raise SystemExit("Usage: pact production {init,status,validate,fingerprint} ...") +def cmd_boundary(args: argparse.Namespace) -> None: + """Verify the boundary register mechanically; exit non-zero on any error. + + Checks every entry in boundary_register.yaml and every contract's + boundary_identifiers against the cited records, and (when contracts exist) + that every cross-boundary usage the contracts declare is registered. + """ + from pact.boundary_register import REGISTER_FILENAME, CiteError, cite_record, load_register + from pact.contracts import validate_boundary_register + + subcommand = getattr(args, "boundary_command", "") + if subcommand == "cite": + try: + record = cite_record( + Path(args.project_dir).resolve(), + kind=args.kind, + path=args.path, + value=args.value, + line=args.line, + command=args.record_command, + ) + except CiteError as exc: + print(f"Refusing to cite: {exc}", file=sys.stderr) + raise SystemExit(1) from None + print(json.dumps(record, indent=2)) + return + if subcommand != "verify": + raise SystemExit("Usage: pact boundary {verify,cite} ...") + + project_dir = Path(args.project_dir).resolve() + if not project_dir.is_dir(): + raise SystemExit(f"Not a directory: {project_dir}") + project = ProjectManager(project_dir) + tree = project.load_tree() + contracts = project.load_all_contracts() + file_entries, _ = load_register(project_dir) + embedded = sum(len(c.boundary_identifiers) for c in contracts.values()) + + errors = validate_boundary_register(contracts, project_dir, tree) + total = len(file_entries) + embedded + if errors: + print(f"Boundary register FAILED: {len(errors)} error(s) across {total} entr{'y' if total == 1 else 'ies'}") + for error in errors: + print(f" - {error}") + raise SystemExit(1) + if total == 0: + print(f"Boundary register PASSED: no {REGISTER_FILENAME} entries and no cross-boundary usage declared") + else: + print(f"Boundary register PASSED: {total} entr{'y' if total == 1 else 'ies'} verified against records") + + def cmd_assess(args: argparse.Namespace) -> None: """Assess codebase architecture for structural friction.""" from pathlib import Path @@ -2615,6 +2699,7 @@ def cmd_handoff(args: argparse.Namespace) -> None: """Render and optionally validate the handoff brief for a component.""" import json as json_mod + from pact.boundary_register import load_register_entries from pact.interface_stub import render_handoff_brief from pact.project import ProjectManager @@ -2678,6 +2763,7 @@ def cmd_handoff(args: argparse.Namespace) -> None: learnings=learnings, standards_brief=standards_brief, max_context_tokens=args.max_tokens, + boundary_entries=load_register_entries(project.project_dir), ) if args.validate: diff --git a/src/pact/contracts.py b/src/pact/contracts.py index 8908c5f..e8d43dc 100644 --- a/src/pact/contracts.py +++ b/src/pact/contracts.py @@ -8,6 +8,7 @@ import ast import logging +import re from pathlib import Path from pact.schemas import ( @@ -404,6 +405,7 @@ def validate_all_contracts( tree: DecompositionTree, contracts: dict[str, ComponentContract], test_suites: dict[str, ContractTestSuite], + project_dir: str | Path | None = None, ) -> GateResult: """Full mechanical validation gate. No LLM, no persuasion. @@ -413,6 +415,9 @@ def validate_all_contracts( 3. Dependency graph is acyclic 4. All contracts have test suites 5. Test code parses + 6. Boundary register: every cross-boundary usage is registered and every + registered side verifies against its records (``project_dir`` is the + root that ``boundary_register.yaml`` and record paths resolve against) """ all_errors: list[str] = [] @@ -476,6 +481,9 @@ def validate_all_contracts( cid, dep_id, ) + # Boundary register — hard errors (never assume an identifier) + all_errors.extend(validate_boundary_register(contracts, project_dir, tree)) + # Cross-component interface compatibility (warnings, not blocking) interface_warnings = validate_cross_component_interfaces(contracts) for w in interface_warnings: @@ -954,3 +962,152 @@ def validate_decomposition_coverage( ) return warnings + + +# ── Boundary register gate ─────────────────────────────────────────── +# +# A boundary that is used must be registered, and every registered side must +# cite a record that still verifies. Mechanical; no LLM; errors, never warnings. + +# DataAccessSideEffect.type prefixes that cross a process/system boundary. +_BOUNDARY_SIDE_EFFECT_RE = re.compile( + r"^(?:database|db|http|https|external|network|api|rpc|grpc|queue|topic|" + r"pubsub|bucket|storage|secret|auth|iam)[_-]", + re.IGNORECASE, +) + + +def find_boundary_usages(contract: ComponentContract) -> list[tuple[str, str]]: + """List the cross-boundary usages a contract declares. + + Returns ``(external_party_or_empty, description)`` pairs. A usage that + names ``external:`` must be covered by an entry naming both this + component and that system; any other usage must be covered by an entry in + which this component is a party opposite some other party. + """ + from pact.boundary_register import is_external_party + + cid = contract.component_id + usages: list[tuple[str, str]] = [] + + def _ext(text: str) -> str: + text = (text or "").strip() + return text if is_external_party(text) else "" + + for dep in contract.dependencies: + if _ext(dep): + usages.append((dep.strip(), f"'{cid}' depends on '{dep.strip()}'")) + for dep in contract.dependency_contracts: + if _ext(dep): + usages.append((dep.strip(), f"'{cid}' dependency_contracts names '{dep.strip()}'")) + for req in contract.requires: + if _ext(req): + usages.append((req.strip(), f"'{cid}' requires '{req.strip()}'")) + + for func in contract.functions: + for se in func.structured_side_effects: + kind = str(se.kind) + target = se.target.strip() + if _ext(target): + usages.append(( + target, + f"'{cid}'.{func.name} side effect {kind} targets '{target}'", + )) + elif kind == "network_call": + usages.append(( + "", + f"'{cid}'.{func.name} side effect network_call" + + (f" on '{target}'" if target else ""), + )) + + for se in contract.data_access.side_effects: + se_type = (se.type or "").strip() + if _ext(se_type): + usages.append((se_type, f"'{cid}' data_access side effect '{se_type}'")) + elif _BOUNDARY_SIDE_EFFECT_RE.match(se_type): + fields = f" (fields: {', '.join(se.fields)})" if se.fields else "" + usages.append(("", f"'{cid}' data_access side effect '{se_type}'{fields}")) + + seen: set[tuple[str, str]] = set() + unique: list[tuple[str, str]] = [] + for usage in usages: + if usage not in seen: + seen.add(usage) + unique.append(usage) + return unique + + +def _entry_parties(entry: object) -> set[str]: + sides = entry.get("sides") if isinstance(entry, dict) else None + if not isinstance(sides, list): + return set() + return { + s.get("party") for s in sides + if isinstance(s, dict) and isinstance(s.get("party"), str) + } + + +def validate_boundary_register( + contracts: dict[str, ComponentContract], + project_dir: str | Path | None, + tree: DecompositionTree | None = None, +) -> list[str]: + """Verify the boundary register and that every used boundary is registered. + + Sources: ``boundary_register.yaml`` at the project root plus every + contract's ``boundary_identifiers``. Every entry is verified against its + records (see ``pact.boundary_register.verify_entries``). Every usage found + by ``find_boundary_usages`` must be covered by an entry. A project with no + register and no cross-boundary usage passes untouched. + """ + from pact.boundary_register import ( + REGISTER_FILENAME, + load_register, + verify_entries, + ) + + errors: list[str] = [] + file_entries: list = [] + if project_dir is not None: + file_entries, load_errors = load_register(project_dir) + errors.extend(load_errors) + + contract_entries: list[tuple[str, list]] = [] + for cid, contract in contracts.items(): + if contract.boundary_identifiers: + contract_entries.append(( + cid, + [e.model_dump(by_alias=True, exclude_none=True) for e in contract.boundary_identifiers], + )) + + known: set[str] | None = None + if tree is not None and tree.nodes: + known = set(tree.nodes.keys()) | set(contracts.keys()) + elif contracts: + known = set(contracts.keys()) + + # Verify all entries together so ids are unique across file and contracts. + all_entries = list(file_entries) + for _, entries in contract_entries: + all_entries.extend(entries) + if all_entries: + errors.extend(verify_entries(all_entries, project_dir, known_components=known)) + + # Coverage: a boundary that is used must be registered. + party_sets = [_entry_parties(e) for e in all_entries] + for cid, contract in contracts.items(): + for external, description in find_boundary_usages(contract): + if external: + covered = any(cid in ps and external in ps for ps in party_sets) + need = f"an entry with sides for both '{cid}' and '{external}'" + else: + covered = any(cid in ps and len(ps) > 1 for ps in party_sets) + need = f"an entry in which '{cid}' is a party opposite another party" + if not covered: + errors.append( + f"UNREGISTERED boundary: {description}, but no boundary register " + f"entry covers it (need {need} in {REGISTER_FILENAME} or the " + f"contract's boundary_identifiers, each side citing a verifiable record). " + f"Never assume an identifier; cite the record" + ) + return errors diff --git a/src/pact/decomposer.py b/src/pact/decomposer.py index d34195d..c749c40 100644 --- a/src/pact/decomposer.py +++ b/src/pact/decomposer.py @@ -26,6 +26,11 @@ author_tests, generate_emission_compliance_test, ) +from pact.boundary_register import ( + BOUNDARY_RULE_PROMPT, + load_register_entries, + render_register_for_prompt, +) from pact.contracts import validate_all_contracts, validate_decomposition_coverage from pact.project import ProjectManager from pact.readiness import ( @@ -98,7 +103,9 @@ async def run_register_establishment( You are a cynical senior architect reviewing a task specification. Find risks, ambiguities, and missing decisions BEFORE any work begins. -Focus on issues that would cause implementation failures.""" +Focus on issues that would cause implementation failures. + +""" + BOUNDARY_RULE_PROMPT async def run_interview( @@ -150,6 +157,12 @@ async def run_interview( verifiable — not vague qualities but observable outcomes. If the task spec already states acceptance criteria, extract them. If it doesn't, propose what "done" should mean based on the task description. +6. Boundary identifiers: every name, endpoint, credential, principal, grant, + role, table, metric/log field, env/config key, header or claim this task + shares with another component or an external system. For each one the task + or SOPs do not cite with a verifiable record, ask a question naming the + identifier and the system that owns it. Never list a boundary identifier as + an assumption: an assumed identifier is the defect this step exists to stop. Be specific and actionable. Focus on issues that would cause different engineers to implement incompatible solutions.""" @@ -598,6 +611,11 @@ async def decompose_and_contract( f"{len(coverage_warnings)} warnings: {'; '.join(coverage_warnings[:3])}", ) + # Boundary register — the only citable source of cross-boundary identifiers + boundary_register_context = render_register_for_prompt( + load_register_entries(project.project_dir), + ) + # Generate or load type registry — canonical shared types type_registry = project.load_type_registry() if type_registry is None: @@ -652,6 +670,7 @@ async def decompose_and_contract( processing_register=processing_register, type_registry=type_registry, readiness_profile=readiness_profile, + boundary_register_context=boundary_register_context, ) # Mechanical correction: replace shared types with registry versions. @@ -785,7 +804,9 @@ async def decompose_and_contract( project.append_audit("type_reconciliation", f"{reconciled} contracts reconciled") # Validate (mechanical gate) - gate = validate_all_contracts(decomp_tree, contracts, test_suites) + gate = validate_all_contracts( + decomp_tree, contracts, test_suites, project.project_dir, + ) project.append_audit( "validation", f"{'PASSED' if gate.passed else 'FAILED'}: {gate.reason}", diff --git a/src/pact/implementer.py b/src/pact/implementer.py index 4c86c25..5673507 100644 --- a/src/pact/implementer.py +++ b/src/pact/implementer.py @@ -25,6 +25,7 @@ from pact.agents.base import AgentBase from pact.agents.code_author import author_code from pact.backends.agent_runtime import provider_for_backend, session_id_for_project +from pact.boundary_register import load_register_entries from pact.interface_stub import get_required_exports from pact.project import ProjectManager from pact.resolution import ScoredAttempt, format_resolution_summary, select_winner @@ -631,6 +632,7 @@ async def implement_component( strategic_context=_build_strategic_context(project, contract), language=language, processing_register=contract.processing_register, + boundary_entries=load_register_entries(project.project_dir), ) # Save implementation files @@ -799,6 +801,7 @@ async def implement_component_iterative( strategic_context=strategic_ctx, processing_register=contract.processing_register, include_test_code=False, # Agent reads test file directly + boundary_entries=load_register_entries(project.project_dir), ) # Write test file so the agent can run it @@ -1018,6 +1021,7 @@ async def implement_component_interactive( strategic_context=strategic_ctx, processing_register=contract.processing_register, include_test_code=False, # Agent reads test file directly + boundary_entries=load_register_entries(project.project_dir), ) # Write test file so the agent can run it @@ -1154,6 +1158,7 @@ async def _run_one_competitor( language=language, strategic_context=_build_strategic_context(project, contract), processing_register=contract.processing_register, + boundary_entries=load_register_entries(project.project_dir), ) # Save to attempt directory (not main src) diff --git a/src/pact/interface_stub.py b/src/pact/interface_stub.py index 9847b4f..576da1d 100644 --- a/src/pact/interface_stub.py +++ b/src/pact/interface_stub.py @@ -1267,9 +1267,16 @@ def render_log_key_preamble_rust(key: str) -> str: def render_dependency_map( component_id: str, contracts: dict[str, ComponentContract], + boundary_entries: list | None = None, + include_boundaries: bool = True, ) -> str: """Render a compact reference of all dependencies' interfaces. + Also renders the registered boundary identifiers this component is a party + to (from ``boundary_entries``, i.e. the project's boundary_register.yaml, + plus every contract's ``boundary_identifiers``), so the implementer uses the + registered literal rather than a plausible-looking guess. + This gives agents working on `component_id` a quick reference for every function they can call on their dependencies, without seeing the full contract details. It's a "what can I use?" cheat sheet. @@ -1319,9 +1326,33 @@ def render_dependency_map( lines.append("") + if include_boundaries: + boundary_block = _render_component_boundaries(component_id, contracts, boundary_entries) + if boundary_block: + lines.append(boundary_block) + lines.append("") + return "\n".join(lines) +def _render_component_boundaries( + component_id: str, + contracts: dict[str, ComponentContract], + boundary_entries: list | None = None, +) -> str: + """Registered boundary literals for ``component_id`` (register + contracts).""" + from pact.boundary_register import render_boundary_identifiers + + entries: list = list(boundary_entries or []) + seen_ids = {getattr(e, "id", None) or (e.get("id") if isinstance(e, dict) else None) for e in entries} + for c in contracts.values(): + for entry in c.boundary_identifiers: + if entry.id not in seen_ids: + entries.append(entry) + seen_ids.add(entry.id) + return render_boundary_identifiers(component_id, entries) + + def render_compact_deps(contracts: dict[str, ComponentContract]) -> str: """Compact dependency reference: function signatures + type shapes only. @@ -1454,6 +1485,7 @@ def render_handoff_brief( max_context_tokens: int = 0, tool_index_context: str = "", language: str = "python", + boundary_entries: list | None = None, ) -> str: """Render a complete handoff document for a fresh agent. @@ -1509,6 +1541,16 @@ def render_handoff_brief( tier1_lines.append("```") tier1_lines.append("") + # Registered boundary identifiers are part of the contract: the literal a + # peer or external system expects. Tier 1 so compression never drops them. + boundary_block = _render_component_boundaries(component_id, contracts, boundary_entries) + if boundary_block: + tier1_lines.append("Registered boundary identifiers you must use verbatim:") + tier1_lines.append("```") + tier1_lines.append(boundary_block) + tier1_lines.append("```") + tier1_lines.append("") + tier1 = "\n".join(tier1_lines) used_tokens = _estimate_tokens(tier1) @@ -1526,11 +1568,13 @@ def render_handoff_brief( tier2_lines.append(tool_index_context) tier2_lines.append("") - # Dependencies + # Dependencies (boundary identifiers are rendered in tier 1, never trimmed) if contract.dependencies: tier2_lines.append("Your available dependencies:") tier2_lines.append("```") - tier2_lines.append(render_dependency_map(component_id, contracts)) + tier2_lines.append(render_dependency_map( + component_id, contracts, boundary_entries, include_boundaries=False, + )) tier2_lines.append("```") tier2_lines.append("") diff --git a/src/pact/mcp_server.py b/src/pact/mcp_server.py index 8057b90..62fbb11 100644 --- a/src/pact/mcp_server.py +++ b/src/pact/mcp_server.py @@ -212,7 +212,7 @@ def tool_validate(self) -> dict[str, Any]: contracts = pm.load_all_contracts() test_suites = pm.load_all_test_suites() - gate = validate_all_contracts(tree, contracts, test_suites) + gate = validate_all_contracts(tree, contracts, test_suites, pm.project_dir) return { "passed": gate.passed, "reason": gate.reason, diff --git a/src/pact/schemas.py b/src/pact/schemas.py index cd0ad34..2c01188 100644 --- a/src/pact/schemas.py +++ b/src/pact/schemas.py @@ -15,6 +15,7 @@ from pydantic import BaseModel, Field, field_validator, model_validator +from pact.boundary_register import BoundaryEntry from pact.readiness import ReadinessProfile @@ -208,6 +209,10 @@ class ComponentContract(BaseModel): processing_register: str = "" data_access: DataAccessDeclaration = DataAccessDeclaration() authority: AuthorityDeclaration = AuthorityDeclaration() + # Cross-boundary identifiers this component defines/consumes, each backed by + # verifiable records (format: boundary-register/1 entries). See + # pact.boundary_register; validated mechanically by validate_all_contracts. + boundary_identifiers: list[BoundaryEntry] = [] # ── Preflight Models ───────────────────────────────────────────────── diff --git a/tests/test_boundary_register.py b/tests/test_boundary_register.py new file mode 100644 index 0000000..8f0e46d --- /dev/null +++ b/tests/test_boundary_register.py @@ -0,0 +1,710 @@ +"""Tests for the boundary register (boundary-register/1). + +Every cross-boundary identifier must be registered with a record the gate can +re-verify; values are never assumed. These tests model the production defects +that motivated the feature: a log formatter writing "level" while the platform +reads "severity", a service name that does not exist, a project-wide grant +that breaks an isolation claim. +""" + +from __future__ import annotations + +import argparse +import hashlib +from pathlib import Path + +import pytest +import yaml + +from pact.boundary_register import ( + contains_token, + is_placeholder, + load_register_entries, + render_register_for_prompt, + verify_entries, +) +from pact.contracts import ( + find_boundary_usages, + validate_all_contracts, + validate_boundary_register, +) +from pact.interface_stub import render_dependency_map, render_handoff_brief +from pact.project import ProjectManager +from pact.schemas import ( + ComponentContract, + ContractTestSuite, + DataAccessDeclaration, + DataAccessSideEffect, + DecompositionNode, + DecompositionTree, + FieldSpec, + FunctionContract, + SideEffect, + SideEffectKind, + TestCase, +) + +OBSERVED_AT = "2026-10-14T12:00:00Z" + + +# ── helpers ─────────────────────────────────────────────────────────── + + +def _write(root: Path, rel: str, text: str) -> str: + path = root / rel + path.parent.mkdir(parents=True, exist_ok=True) + path.write_text(text) + return hashlib.sha256(path.read_bytes()).hexdigest() + + +def _file_record(root: Path, rel: str, line: int) -> dict: + return { + "kind": "file", + "path": rel, + "line": line, + "sha256": hashlib.sha256((root / rel).read_bytes()).hexdigest(), + } + + +def _obs_record(root: Path, rel: str, command: str = "gcloud monitoring uptime list-configs") -> dict: + return { + "kind": "observation", + "path": rel, + "sha256": hashlib.sha256((root / rel).read_bytes()).hexdigest(), + "command": command, + "observed_at": OBSERVED_AT, + } + + +def _equal_entry(root: Path, *, producer_value: str = "severity", reader_value: str = "severity") -> dict: + """Log field the app writes vs. the field the log platform reads.""" + _write(root, "app/logging.py", f'FORMAT = {{"{producer_value}": record.levelname}}\n') + _write(root, "ops/log_platform.yaml", f"level_field: {reader_value}\n") + return { + "id": "log-level-field", + "kind": "field", + "relation": "equal", + "description": "The log platform only derives severity from this JSON key", + "sides": [ + { + "party": "external:cloud-logging", + "role": "defines", + "value": reader_value, + "records": [_file_record(root, "ops/log_platform.yaml", 1)], + }, + { + "party": "api", + "role": "consumes", + "value": producer_value, + "records": [_file_record(root, "app/logging.py", 1)], + }, + ], + } + + +def _write_register(root: Path, entries: list[dict], schema: str = "boundary-register/1") -> None: + (root / "boundary_register.yaml").write_text( + yaml.safe_dump({"schema": schema, "entries": entries}, sort_keys=False) + ) + + +def _contract(cid: str = "api", **kwargs) -> ComponentContract: + return ComponentContract( + component_id=cid, + name=cid.title(), + description="Component", + functions=kwargs.pop("functions", [ + FunctionContract( + name="handle", + description="Handles a request", + inputs=[FieldSpec(name="x", type_ref="str")], + output_type="str", + ), + ]), + data_access=kwargs.pop("data_access", DataAccessDeclaration( + reads=["PUBLIC"], rationale="Reads public request input only", + )), + **kwargs, + ) + + +def _suite(cid: str) -> ContractTestSuite: + return ContractTestSuite( + component_id=cid, + contract_version=1, + test_cases=[TestCase(id="t1", description="t", function="handle", category="happy_path")], + generated_code="def test_ok(): pass\n", + ) + + +def _tree(*cids: str) -> DecompositionTree: + return DecompositionTree( + root_id=cids[0], + nodes={c: DecompositionNode(component_id=c, name=c.title(), description=c) for c in cids}, + ) + + +def _db_contract(cid: str = "api") -> ComponentContract: + return _contract(cid, data_access=DataAccessDeclaration( + reads=["PII"], + rationale="Reads user rows from the accounts database for login", + side_effects=[DataAccessSideEffect(type="database_read", classification="PII", fields=["users.email"])], + )) + + +# ── token matching and placeholders ─────────────────────────────────── + + +class TestTokenRule: + def test_token_does_not_match_inside_longer_identifier(self): + assert not contains_token("service: mea-auth-runtime", "mea-auth") + assert not contains_token("xmea-auth", "mea-auth") + assert not contains_token("mea_auth_x", "mea_auth") + + def test_token_matches_with_punctuation_neighbours(self): + assert contains_token('{"level": "INFO"}', "level") + assert contains_token("mea-auth", "mea-auth") + assert contains_token("url: https://mea-auth.run.app", "mea-auth") + + def test_later_occurrence_can_match(self): + assert contains_token("mea-auth-runtime and mea-auth", "mea-auth") + + @pytest.mark.parametrize(("haystack", "value", "expected"), [ + ("metric foo.bar.baz", "foo.bar", False), + ("a.foo.bar", "foo.bar", False), + ("type=foo.bar.", "foo.bar", True), + ("GET /health/ready", "/health", False), + ("GET /health HTTP/1.1", "/health", True), + ("entry.severity = level", "severity", True), + ('"level": "INFO"', "level", True), + ("svc-auth-runtime", "svc-auth", False), + ("serviceAccount:svc-auth@p.iam", "svc-auth", True), + ]) + def test_shared_format_vectors(self, haystack, value, expected): + """boundary-register/1 token vectors shared with the factory and Constrain.""" + assert contains_token(haystack, value) is expected + + @pytest.mark.parametrize("value", [ + "TODO", "tbd", "TBC", "assumed", "Placeholder", "changeme", "xxx", "XXXX", + "?", "???", " todo ", "", "svc-", + ]) + def test_placeholders_detected(self, value): + assert is_placeholder(value) + + @pytest.mark.parametrize("value", [ + "severity", "cloud_run_revision", "roles/run.invoker", + "unknown", "none", "null", "n/a", "UNKNOWN", "xx", "todo-list", "a < b", + ]) + def test_real_values_are_not_placeholders(self, value): + assert not is_placeholder(value) + + +# ── relation: equal ─────────────────────────────────────────────────── + + +class TestEqualRelation: + def test_matching_values_with_valid_records_pass(self, tmp_path): + assert verify_entries([_equal_entry(tmp_path)], tmp_path) == [] + + def test_value_mismatch_fails(self, tmp_path): + errors = verify_entries([_equal_entry(tmp_path, producer_value="level")], tmp_path) + joined = "\n".join(errors) + assert "MISMATCH" in joined + assert "log-level-field" in joined + assert "'level'" in joined and "'severity'" in joined + assert "near-miss" not in joined + + def test_near_miss_by_separators_is_named(self, tmp_path): + errors = verify_entries( + [_equal_entry(tmp_path, producer_value="uptime_url", reader_value="uptime-url")], tmp_path, + ) + assert any("near-miss" in e and "separators" in e for e in errors), errors + + def test_near_miss_by_case_is_named(self, tmp_path): + errors = verify_entries( + [_equal_entry(tmp_path, producer_value="Severity", reader_value="severity")], tmp_path, + ) + assert any("near-miss" in e and "case" in e for e in errors), errors + + def test_side_without_records_is_unverified(self, tmp_path): + entry = _equal_entry(tmp_path) + entry["sides"][1]["records"] = [] + errors = verify_entries([entry], tmp_path) + assert any("UNVERIFIED" in e and "side[1 api/consumes]" in e for e in errors), errors + + def test_stale_digest_fails_naming_the_file(self, tmp_path): + entry = _equal_entry(tmp_path) + (tmp_path / "app/logging.py").write_text('FORMAT = {"severity": record.levelname} # edited\n') + errors = verify_entries([entry], tmp_path) + assert any("STALE" in e and "app/logging.py" in e and "never hand-edit" in e for e in errors), errors + + def test_cited_line_must_contain_value(self, tmp_path): + entry = _equal_entry(tmp_path) + _write(tmp_path, "app/logging.py", "import json\nFORMAT = {\"severity\": 1}\n") + entry["sides"][1]["records"] = [_file_record(tmp_path, "app/logging.py", 1)] + errors = verify_entries([entry], tmp_path) + assert any("line 1 does not contain 'severity'" in e for e in errors), errors + + def test_line_beyond_end_of_file_fails(self, tmp_path): + entry = _equal_entry(tmp_path) + entry["sides"][1]["records"][0]["line"] = 99 + errors = verify_entries([entry], tmp_path) + assert any("beyond the end" in e for e in errors), errors + + def test_token_boundary_rejects_longer_service_name(self, tmp_path): + _write(tmp_path, "infra/uptime.tf", 'service = "mea-auth-runtime"\n') + _write(tmp_path, "infra/services.tf", 'name = "mea-auth"\n') + entry = { + "id": "uptime-target", + "kind": "service", + "relation": "equal", + "description": "Uptime check must target a service that exists", + "sides": [ + {"party": "external:cloud-run", "role": "defines", "value": "mea-auth", + "records": [_file_record(tmp_path, "infra/services.tf", 1)]}, + {"party": "monitoring", "role": "consumes", "value": "mea-auth", + "records": [_file_record(tmp_path, "infra/uptime.tf", 1)]}, + ], + } + errors = verify_entries([entry], tmp_path) + assert len(errors) == 1 and "infra/uptime.tf" in errors[0] and "does not contain" in errors[0], errors + + def test_observation_record_passes(self, tmp_path): + _write(tmp_path, "obs/uptime.json", '{"monitoredResource": {"type": "cloud_run_revision"}}\n') + _write(tmp_path, "ops/alert.yaml", 'filter: resource.type = "cloud_run_revision"\n') + entry = { + "id": "uptime-resource-type", + "kind": "metric", + "relation": "equal", + "description": "Alert filter must match the resource type uptime checks emit", + "sides": [ + {"party": "external:cloud-monitoring", "role": "defines", "value": "cloud_run_revision", + "records": [_obs_record(tmp_path, "obs/uptime.json")]}, + {"party": "alerts", "role": "consumes", "value": "cloud_run_revision", + "records": [_file_record(tmp_path, "ops/alert.yaml", 1)]}, + ], + } + assert verify_entries([entry], tmp_path) == [] + + def test_observation_needs_command_and_timestamp(self, tmp_path): + _write(tmp_path, "obs/uptime.json", '{"type": "cloud_run_revision"}\n') + rec = _obs_record(tmp_path, "obs/uptime.json") + rec["command"] = "" + rec["observed_at"] = "yesterday" + entry = _equal_entry(tmp_path, producer_value="cloud_run_revision", reader_value="cloud_run_revision") + entry["sides"][0]["records"] = [rec] + errors = verify_entries([entry], tmp_path) + assert any("command" in e for e in errors), errors + assert any("observed_at" in e for e in errors), errors + + def test_observation_must_contain_value(self, tmp_path): + _write(tmp_path, "obs/uptime.json", '{"type": "cloud_run_revision"}\n') + entry = _equal_entry(tmp_path, producer_value="uptime_url", reader_value="uptime_url") + entry["sides"][0]["records"] = [_obs_record(tmp_path, "obs/uptime.json")] + errors = verify_entries([entry], tmp_path) + assert any("observation 'obs/uptime.json' does not contain 'uptime_url'" in e for e in errors), errors + + def test_equal_needs_authority_and_user_sides(self, tmp_path): + entry = _equal_entry(tmp_path) + entry["sides"][0]["role"] = "consumes" + errors = verify_entries([entry], tmp_path) + assert any("defines or grants" in e for e in errors), errors + + @pytest.mark.parametrize("value", ["TODO", "", "???", "changeme", "assumed"]) + def test_placeholder_value_fails(self, tmp_path, value): + entry = _equal_entry(tmp_path) + entry["sides"][1]["value"] = value + errors = verify_entries([entry], tmp_path) + assert any("placeholder" in e and "side[1 api/consumes]" in e for e in errors), errors + + +# ── structural rules ────────────────────────────────────────────────── + + +class TestStructure: + def test_duplicate_ids_fail(self, tmp_path): + entry = _equal_entry(tmp_path) + errors = verify_entries([entry, dict(entry)], tmp_path) + assert any("duplicate id" in e for e in errors), errors + + def test_bad_id_kind_relation_role_fail(self, tmp_path): + entry = _equal_entry(tmp_path) + entry.update({"id": "-bad id", "kind": "widget", "relation": "similar", "description": ""}) + entry["sides"][0]["role"] = "owns" + joined = "\n".join(verify_entries([entry], tmp_path)) + for fragment in ("id must match", "kind must be one of", "relation must be", "description", "role must be one of"): + assert fragment in joined, fragment + + def test_unknown_party_fails_when_components_known(self, tmp_path): + errors = verify_entries([_equal_entry(tmp_path)], tmp_path, known_components={"gateway"}) + assert any("'api' is not a component" in e for e in errors), errors + + @pytest.mark.parametrize("bad_path", ["../outside.txt", "/etc/hosts", "app/../../outside.txt"]) + def test_path_traversal_refused(self, tmp_path, bad_path): + project = tmp_path / "project" + project.mkdir() + (tmp_path / "outside.txt").write_text("severity\n") + entry = _equal_entry(project) + entry["sides"][1]["records"] = [{ + "kind": "file", "path": bad_path, "line": 1, + "sha256": hashlib.sha256(b"severity\n").hexdigest(), + }] + errors = verify_entries([entry], project) + assert any(("must be relative" in e or "must not contain '..'" in e) for e in errors), errors + + def test_symlink_escaping_root_refused(self, tmp_path): + project = tmp_path / "project" + project.mkdir() + (tmp_path / "outside.txt").write_text("severity\n") + (project / "link.txt").symlink_to(tmp_path / "outside.txt") + entry = _equal_entry(project) + entry["sides"][1]["records"] = [{ + "kind": "file", "path": "link.txt", "line": 1, + "sha256": hashlib.sha256(b"severity\n").hexdigest(), + }] + errors = verify_entries([entry], project) + assert any("resolves outside the project root" in e for e in errors), errors + + def test_missing_record_file_fails(self, tmp_path): + entry = _equal_entry(tmp_path) + (tmp_path / "app/logging.py").unlink() + errors = verify_entries([entry], tmp_path) + assert any("'app/logging.py' does not exist" in e for e in errors), errors + + def test_wrong_schema_in_register_file_fails(self, tmp_path): + _write_register(tmp_path, [_equal_entry(tmp_path)], schema="boundary-register/0") + errors = validate_boundary_register({}, tmp_path) + assert any("schema must be 'boundary-register/1'" in e for e in errors), errors + + +# ── relation: absent (isolation) ────────────────────────────────────── + + +def _absent_entry(root: Path, policy_text: str) -> dict: + subject = "serviceAccount:gateway@proj.iam.gserviceaccount.com" + _write(root, "infra/gateway_sa.tf", f'member = "{subject}"\n') + _write(root, "obs/project_iam.json", policy_text) + return { + "id": "isolated-svc-no-gateway-invoker", + "kind": "grant", + "relation": "absent", + "description": "The gateway must not be able to invoke the isolated service", + "sides": [ + {"party": "gateway", "role": "presents", "value": subject, + "records": [_file_record(root, "infra/gateway_sa.tf", 1)]}, + {"party": "external:gcp-iam", "role": "denies", "value": "roles/run.invoker", + "records": [_obs_record(root, "obs/project_iam.json", "gcloud projects get-iam-policy proj")]}, + ], + } + + +class TestAbsentRelation: + def test_passes_when_subject_absent(self, tmp_path): + entry = _absent_entry(tmp_path, '{"bindings": [{"role": "roles/run.invoker", "members": ["serviceAccount:ops@proj.iam.gserviceaccount.com"]}]}\n') + assert verify_entries([entry], tmp_path) == [] + + def test_fails_when_subject_present(self, tmp_path): + entry = _absent_entry(tmp_path, '{"bindings": [{"role": "roles/run.invoker", "members": ["serviceAccount:gateway@proj.iam.gserviceaccount.com"]}]}\n') + errors = verify_entries([entry], tmp_path) + assert any("isolation violated" in e and "obs/project_iam.json" in e for e in errors), errors + + def test_denies_record_must_verify_by_digest(self, tmp_path): + entry = _absent_entry(tmp_path, '{"bindings": []}\n') + (tmp_path / "obs/project_iam.json").write_text('{"bindings": [{"members": []}]}\n') + errors = verify_entries([entry], tmp_path) + assert any("STALE" in e and "obs/project_iam.json" in e for e in errors), errors + + def test_needs_exactly_one_presents_and_a_denies(self, tmp_path): + entry = _absent_entry(tmp_path, '{"bindings": []}\n') + entry["sides"] = [entry["sides"][0], dict(entry["sides"][0])] + errors = verify_entries([entry], tmp_path) + assert any("exactly one 'presents'" in e for e in errors), errors + assert any("at least one 'denies'" in e for e in errors), errors + + +# ── gate wiring: validate_all_contracts ─────────────────────────────── + + +class TestGateWiring: + def test_no_boundaries_no_register_passes(self, tmp_path): + gate = validate_all_contracts(_tree("api"), {"api": _contract()}, {"api": _suite("api")}, tmp_path) + assert gate.passed, gate.details + + def test_unregistered_database_usage_fails_and_names_it(self, tmp_path): + gate = validate_all_contracts(_tree("api"), {"api": _db_contract()}, {"api": _suite("api")}, tmp_path) + assert not gate.passed + joined = "\n".join(gate.details) + assert "UNREGISTERED boundary" in joined + assert "data_access side effect 'database_read'" in joined + assert "users.email" in joined + + def test_unregistered_external_dependency_fails(self, tmp_path): + contract = _contract(requires=["external:stripe"]) + gate = validate_all_contracts(_tree("api"), {"api": contract}, {"api": _suite("api")}, tmp_path) + assert not gate.passed + assert any("requires 'external:stripe'" in d for d in gate.details), gate.details + + def test_network_call_side_effect_is_a_boundary(self): + contract = _contract(functions=[FunctionContract( + name="probe", description="Calls a peer", inputs=[], output_type="bool", + structured_side_effects=[SideEffect(kind=SideEffectKind.NETWORK_CALL, target="billing /healthz")], + )]) + usages = find_boundary_usages(contract) + assert usages and usages[0][0] == "" and "network_call on 'billing /healthz'" in usages[0][1] + + def test_registered_usage_passes(self, tmp_path): + _write_register(tmp_path, [_equal_entry(tmp_path)]) + contract = _contract(requires=["external:cloud-logging"]) + gate = validate_all_contracts(_tree("api"), {"api": contract}, {"api": _suite("api")}, tmp_path) + assert gate.passed, gate.details + + def test_registered_entry_that_does_not_cover_named_system_fails(self, tmp_path): + _write_register(tmp_path, [_equal_entry(tmp_path)]) + contract = _contract(requires=["external:cloud-logging", "external:postgres"]) + gate = validate_all_contracts(_tree("api"), {"api": contract}, {"api": _suite("api")}, tmp_path) + assert not gate.passed + assert any("external:postgres" in d and "UNREGISTERED" in d for d in gate.details), gate.details + assert not any("external:cloud-logging'" in d and "UNREGISTERED" in d for d in gate.details) + + def test_register_errors_are_hard_errors(self, tmp_path): + _write_register(tmp_path, [_equal_entry(tmp_path, producer_value="level")]) + gate = validate_all_contracts(_tree("api"), {"api": _contract()}, {"api": _suite("api")}, tmp_path) + assert not gate.passed + assert any("MISMATCH" in d for d in gate.details), gate.details + assert not any(d.startswith("[warning]") and "MISMATCH" in d for d in gate.details) + + def test_contract_embedded_entries_are_verified(self, tmp_path): + bad = _equal_entry(tmp_path) + bad["sides"][1]["records"][0]["sha256"] = "0" * 64 + contract = _db_contract().model_copy(update={ + "boundary_identifiers": ComponentContract.model_validate( + {"component_id": "api", "name": "Api", "description": "d", "boundary_identifiers": [bad]} + ).boundary_identifiers, + }) + gate = validate_all_contracts(_tree("api"), {"api": contract}, {"api": _suite("api")}, tmp_path) + assert not gate.passed + assert any("STALE" in d and "app/logging.py" in d for d in gate.details), gate.details + assert not any("UNREGISTERED" in d for d in gate.details) + + def test_duplicate_id_across_register_and_contract_fails(self, tmp_path): + entry = _equal_entry(tmp_path) + _write_register(tmp_path, [entry]) + contract = ComponentContract.model_validate({ + **_contract().model_dump(), "boundary_identifiers": [entry], + }) + errors = validate_boundary_register({"api": contract}, tmp_path, _tree("api")) + assert any("duplicate id" in e for e in errors), errors + + def test_project_validate_cli_and_mcp_enforce(self, tmp_path, capsys): + from pact.cli import cmd_boundary, cmd_validate + from pact.mcp_server import PactMCPServer + + pm = ProjectManager(tmp_path) + pm.save_tree(_tree("api")) + pm.save_contract(_db_contract()) + pm.save_test_suite(_suite("api")) + + with pytest.raises(SystemExit) as exc: + cmd_validate(argparse.Namespace(project_dir=str(tmp_path))) + assert exc.value.code == 1 + assert "UNREGISTERED boundary" in capsys.readouterr().out + + with pytest.raises(SystemExit) as exc: + cmd_boundary(argparse.Namespace(boundary_command="verify", project_dir=str(tmp_path))) + assert exc.value.code == 1 + assert "UNREGISTERED boundary" in capsys.readouterr().out + + (tmp_path / ".pact").mkdir(exist_ok=True) + result = PactMCPServer(tmp_path).tool_validate() + assert result.get("passed") is False, result + assert any("UNREGISTERED" in e for e in result["errors"]) + + +class TestBoundaryCli: + def test_verify_prints_entry_side_and_reason(self, tmp_path, capsys): + from pact.cli import cmd_boundary + + _write_register(tmp_path, [_equal_entry(tmp_path, producer_value="level")]) + with pytest.raises(SystemExit) as exc: + cmd_boundary(argparse.Namespace(boundary_command="verify", project_dir=str(tmp_path))) + assert exc.value.code == 1 + out = capsys.readouterr().out + assert "boundary[log-level-field]" in out + assert "MISMATCH" in out + + def test_verify_passes_on_valid_register(self, tmp_path, capsys): + from pact.cli import cmd_boundary + + _write_register(tmp_path, [_equal_entry(tmp_path)]) + cmd_boundary(argparse.Namespace(boundary_command="verify", project_dir=str(tmp_path))) + assert "PASSED: 1 entry verified" in capsys.readouterr().out + + def test_main_dispatches_boundary_verify(self, tmp_path, monkeypatch): + from pact.cli import main + + _write_register(tmp_path, [_equal_entry(tmp_path, producer_value="level")]) + monkeypatch.setattr("sys.argv", ["pact", "boundary", "verify", str(tmp_path)]) + with pytest.raises(SystemExit) as exc: + main() + assert exc.value.code == 1 + + +# ── rendering for implementers and authors ──────────────────────────── + + +class TestRendering: + def test_dependency_map_shows_registered_values(self, tmp_path): + _write_register(tmp_path, [_equal_entry(tmp_path)]) + entries = load_register_entries(tmp_path) + rendered = render_dependency_map("api", {"api": _contract()}, entries) + assert "log-level-field" in rendered + assert "'severity'" in rendered + assert "app/logging.py:1" in rendered + assert "<- you" in rendered + + def test_dependency_map_omits_entries_for_other_components(self, tmp_path): + _write_register(tmp_path, [_equal_entry(tmp_path)]) + rendered = render_dependency_map("worker", {"worker": _contract("worker")}, load_register_entries(tmp_path)) + assert "log-level-field" not in rendered + + def test_handoff_brief_shows_registered_values_even_when_compressed(self, tmp_path): + _write_register(tmp_path, [_equal_entry(tmp_path)]) + contract = _contract() + brief = render_handoff_brief( + "api", contract, {"api": contract}, + boundary_entries=load_register_entries(tmp_path), + sops="x " * 5000, max_context_tokens=50, + ) + assert "Registered boundary identifiers you must use verbatim" in brief + assert "'severity'" in brief + + def test_contract_embedded_entries_render(self, tmp_path): + contract = ComponentContract.model_validate({ + **_contract().model_dump(), "boundary_identifiers": [_equal_entry(tmp_path)], + }) + rendered = render_dependency_map("api", {"api": contract}) + assert "log-level-field" in rendered and "'severity'" in rendered + + def test_register_rendered_for_contract_author(self, tmp_path): + _write_register(tmp_path, [_equal_entry(tmp_path)]) + text = render_register_for_prompt(load_register_entries(tmp_path)) + assert "boundary-register/1" in text + assert "severity" in text and "sha256" in text + + +class TestPrompts: + def test_contract_author_interview_and_shaper_forbid_invented_identifiers(self): + from pact.agents.contract_author import CONTRACT_SYSTEM + from pact.agents.shaper import SYSTEM_PROMPT + from pact.decomposer import INTERVIEW_SYSTEM + + for prompt in (CONTRACT_SYSTEM, INTERVIEW_SYSTEM, SYSTEM_PROMPT): + assert "Never invent or assume one" in prompt + assert "boundary-register/1" in prompt + assert "open question" in prompt + assert "pact boundary cite" in prompt + assert "denies side's value names the" in prompt + + +# ── producing records: pact boundary cite ───────────────────────────── + + +class TestCite: + def test_file_record_round_trips_through_verifier(self, tmp_path): + from pact.boundary_register import cite_record + + _write(tmp_path, "ops/platform.yaml", "# header\nlevel_field: severity\n") + _write(tmp_path, "app/logging.py", 'FORMAT = {"severity": record.levelname}\n') + entry = { + "id": "log-level-field", "kind": "field", "relation": "equal", + "description": "The log platform reads severity from this key", + "sides": [ + {"party": "external:cloud-logging", "role": "defines", "value": "severity", + "records": [cite_record(tmp_path, kind="file", path="ops/platform.yaml", value="severity")]}, + {"party": "api", "role": "consumes", "value": "severity", + "records": [cite_record(tmp_path, kind="file", path="app/logging.py", value="severity", line=1)]}, + ], + } + assert verify_entries([entry], tmp_path) == [] + + def test_line_auto_detect_uses_first_token_match(self, tmp_path): + from pact.boundary_register import cite_record + + _write(tmp_path, "infra/svc.tf", 'a = "mea-auth-runtime"\nb = "mea-auth"\nc = "mea-auth"\n') + record = cite_record(tmp_path, kind="file", path="infra/svc.tf", value="mea-auth") + assert record["line"] == 2 + assert record["sha256"] == hashlib.sha256((tmp_path / "infra/svc.tf").read_bytes()).hexdigest() + + def test_refuses_when_value_absent_or_not_on_line(self, tmp_path): + from pact.boundary_register import CiteError, cite_record + + _write(tmp_path, "infra/svc.tf", 'a = "mea-auth-runtime"\nb = "other"\n') + with pytest.raises(CiteError, match="does not occur as a token"): + cite_record(tmp_path, kind="file", path="infra/svc.tf", value="mea-auth") + with pytest.raises(CiteError, match="line 2 does not contain"): + cite_record(tmp_path, kind="file", path="infra/svc.tf", value="other-x", line=2) + with pytest.raises(CiteError, match="does not contain"): + cite_record(tmp_path, kind="observation", path="infra/svc.tf", value="mea-auth", command="cat infra/svc.tf") + + def test_refuses_path_escape(self, tmp_path): + from pact.boundary_register import CiteError, cite_record + + with pytest.raises(CiteError, match="must not contain '..'"): + cite_record(tmp_path, kind="file", path="../x", value="v") + + def test_observation_needs_command_and_stamps_utc(self, tmp_path): + from pact.boundary_register import CiteError, cite_record + + _write(tmp_path, "obs/uptime.json", '{"type": "cloud_run_revision"}\n') + with pytest.raises(CiteError, match="--command"): + cite_record(tmp_path, kind="observation", path="obs/uptime.json", value="cloud_run_revision") + record = cite_record( + tmp_path, kind="observation", path="obs/uptime.json", value="cloud_run_revision", + command="gcloud monitoring uptime list-configs --format json", + ) + assert record["observed_at"].endswith("Z") + entry = _equal_entry(tmp_path, producer_value="cloud_run_revision", reader_value="cloud_run_revision") + entry["sides"][0]["records"] = [record] + assert verify_entries([entry], tmp_path) == [] + + def test_cli_prints_json_and_exits_1_on_refusal(self, tmp_path, monkeypatch, capsys): + import json + + from pact.cli import main + + _write(tmp_path, "ops/platform.yaml", "level_field: severity\n") + monkeypatch.setattr("sys.argv", [ + "pact", "boundary", "cite", "--kind", "file", "--path", "ops/platform.yaml", + "--value", "severity", str(tmp_path), + ]) + main() + record = json.loads(capsys.readouterr().out) + assert record == { + "kind": "file", "path": "ops/platform.yaml", "line": 1, + "sha256": hashlib.sha256(b"level_field: severity\n").hexdigest(), + } + + monkeypatch.setattr("sys.argv", [ + "pact", "boundary", "cite", "--kind", "file", "--path", "ops/platform.yaml", + "--value", "level", str(tmp_path), + ]) + with pytest.raises(SystemExit) as exc: + main() + assert exc.value.code == 1 + assert "Refusing to cite" in capsys.readouterr().err + + def test_cli_observation_with_command(self, tmp_path, monkeypatch, capsys): + import json + + from pact.cli import main + + _write(tmp_path, "obs/uptime.json", '{"type": "cloud_run_revision"}\n') + monkeypatch.setattr("sys.argv", [ + "pact", "boundary", "cite", "--kind", "observation", "--path", "obs/uptime.json", + "--value", "cloud_run_revision", "--command", "gcloud monitoring uptime list-configs", + str(tmp_path), + ]) + main() + record = json.loads(capsys.readouterr().out) + assert record["command"] == "gcloud monitoring uptime list-configs" + assert record["kind"] == "observation" and record["observed_at"].endswith("Z")