From 576af8b44def7095aaec839c5e159ed49829d88d Mon Sep 17 00:00:00 2001 From: aihop <4408344+hhyykk@users.noreply.github.com> Date: Sun, 25 Oct 2026 22:25:29 +0800 Subject: [PATCH] feat(contract): add authority source digest locks and executable invariants probe Signed-off-by: aihop <4408344+hhyykk@users.noreply.github.com> --- loopx/bootstrap.py | 20 +- loopx/contract.py | 42 ++ .../goals/architecture_invariants.py | 381 ++++++++++++++++++ tests/test_architecture_invariants.py | 268 ++++++++++++ 4 files changed, 708 insertions(+), 3 deletions(-) create mode 100644 loopx/control_plane/goals/architecture_invariants.py create mode 100644 tests/test_architecture_invariants.py diff --git a/loopx/bootstrap.py b/loopx/bootstrap.py index aac03ac5dd..ab9f435d33 100644 --- a/loopx/bootstrap.py +++ b/loopx/bootstrap.py @@ -59,6 +59,11 @@ validate_goal_id_path_segment, ) from .control_plane.goals.active_state_metadata import markdown_blockquote, markdown_frontmatter_string +from .control_plane.goals.architecture_invariants import ( + compute_file_sha256, + discover_top_level_specs, + render_architecture_invariants_markdown, +) from .registry_writability import probe_registry_write_path @@ -90,9 +95,16 @@ def resolve_project_path(project: Path, path: Path | None) -> Path | None: def render_authority_sources(project: Path, goal_doc: Path | None) -> str: - if not goal_doc: + entries: list[str] = [] + if goal_doc: + entries.append(f"- Primary goal document: `{rel_or_abs(goal_doc, project)}`") + for spec in discover_top_level_specs(project): + entries.append( + f"- Top-level specification (`{spec['kind']}`): `{spec['path']}` (sha256=`{spec['sha256'][:12]}`)" + ) + if not entries: return "- No explicit goal document was provided during bootstrap." - return f"- Primary goal document: `{rel_or_abs(goal_doc, project)}`" + return "\n".join(entries) def repair_missing_todo_source_sections(state_text: str) -> tuple[str, list[str]]: @@ -172,7 +184,7 @@ def render_state_markdown( - Do not perform irreversible production operations without explicit approval. - Do not publish private project evidence. - Do not optimize for activity if no useful artifact or decision can be produced. - +{render_architecture_invariants_markdown(project)} ## User Todo / Owner Review Reading Queue ## Agent Todo @@ -222,8 +234,10 @@ def build_goal_entry( "kind": "goal_doc", "path": rel_or_abs(goal_doc, project), "role": "primary_goal_document", + **({"sha256": compute_file_sha256(goal_doc)} if goal_doc.is_file() else {}), } ) + authority_sources.extend(discover_top_level_specs(project)) adapter = { "kind": adapter_kind, "status": adapter_status, diff --git a/loopx/contract.py b/loopx/contract.py index 1f9f53f5d3..0a8db9dd67 100644 --- a/loopx/contract.py +++ b/loopx/contract.py @@ -20,6 +20,10 @@ contract_error_views, project_contract_health_for_goal, ) +from .control_plane.goals.architecture_invariants import ( + verify_authority_source_digests, + verify_executable_architecture_invariants, +) from .control_plane.goals.activation import ( GoalActivationState, goal_activation_state, @@ -850,6 +854,35 @@ def _active_state_projection_gap_warnings( return warnings +def _authority_and_invariant_diagnostics( + registry: dict[str, Any], + *, + include_architecture_invariants: bool, + goal_id_filter: str | None = None, + activation_state_filter: GoalActivationState | str | None = None, +) -> tuple[list[dict[str, Any]], list[str]]: + diagnostics: list[dict[str, Any]] = [] + checks: list[str] = [] + for goal in registry_goals(registry): + goal_id = str(goal.get("id") or "") + if goal_id_filter and goal_id != goal_id_filter: + continue + if ( + activation_state_filter is not None + and goal_activation_state(goal) + is not normalize_goal_activation_state(activation_state_filter) + ): + continue + goal_diags, goal_checks = verify_authority_source_digests(goal) + if include_architecture_invariants: + executable_diags, executable_checks = verify_executable_architecture_invariants(goal) + goal_diags.extend(executable_diags) + goal_checks.extend(executable_checks) + diagnostics.extend(goal_diags) + checks.extend(goal_checks) + return diagnostics, checks + + def _tracked_scan_files(scan_root: Path) -> list[Path]: scan_root = scan_root.resolve() repo_root = _git_worktree_root(scan_root) @@ -1204,6 +1237,7 @@ def check_contract( goal_id_filter: str | None = None, activation_state_filter: GoalActivationState | str | None = None, include_public_boundary_scan: bool = True, + include_architecture_invariants: bool = True, history_audit: RunHistoryAudit | None = None, registry: dict[str, Any] | None = None, todo_snapshot: CanonicalTodoSnapshot | None = None, @@ -1273,6 +1307,14 @@ def add_global_error(code: str, message: str) -> None: if checked_user_gates: checks.append(f"user-gate scopes checked: {checked_user_gates} open multi-agent gates") error_diagnostics.extend(todo_contract_diagnostics) + invariant_diagnostics, invariant_checks = _authority_and_invariant_diagnostics( + registry, + include_architecture_invariants=include_architecture_invariants, + goal_id_filter=goal_id_filter, + activation_state_filter=activation_state_filter, + ) + error_diagnostics.extend(invariant_diagnostics) + checks.extend(invariant_checks) warnings.extend( _active_state_projection_gap_warnings( registry, diff --git a/loopx/control_plane/goals/architecture_invariants.py b/loopx/control_plane/goals/architecture_invariants.py new file mode 100644 index 0000000000..f76b2f7e76 --- /dev/null +++ b/loopx/control_plane/goals/architecture_invariants.py @@ -0,0 +1,381 @@ +from __future__ import annotations + +import ast +import fnmatch +import hashlib +import os +import re +from dataclasses import dataclass +from pathlib import Path +from typing import Any + +from .contract_health import contract_error_diagnostic + +INVARIANT_SECTION_PATTERN = re.compile( + r"^##\s+(?:Repo-Wide\s+Invariants|Architecture\s+Invariants|Invariants)\s*$", + re.IGNORECASE, +) +ANY_H2_PATTERN = re.compile(r"^##\s+\S") +FORBID_IMPORT_RULE_PATTERN = re.compile( + r"^(?:-\s+|\*\s+)?`?forbid_import:\s*([^\s`]+)\s*->\s*([^\s`]+)`?(?:\s*(?:—|--|#)\s*(.*))?$", + re.IGNORECASE, +) +FORBID_PATTERN_RULE_PATTERN = re.compile( + r"^(?:-\s+|\*\s+)?`?forbid_pattern:\s*([^\s`]+)\s*::\s*(.+?)`?(?:\s+#\s*(.*))?$", + re.IGNORECASE, +) +MUST_EXIST_RULE_PATTERN = re.compile( + r"^(?:-\s+|\*\s+)?`?must_exist:\s*([^\s`]+)`?(?:\s*(?:—|--|#)\s*(.*))?$", + re.IGNORECASE, +) +IMMUTABLE_PATH_RULE_PATTERN = re.compile( + r"^(?:-\s+|\*\s+)?`?immutable_path:\s*([^\s`]+)`?(?:\s*(?:—|--|#)\s*(.*))?$", + re.IGNORECASE, +) + + +@dataclass(frozen=True) +class ArchitectureInvariantRule: + kind: str # "forbid_import" | "forbid_pattern" | "must_exist" | "immutable_path" + source_glob: str + target: str + rationale: str = "" + raw_line: str = "" + + def to_dict(self) -> dict[str, str]: + return { + "kind": self.kind, + "source_glob": self.source_glob, + "target": self.target, + "rationale": self.rationale, + "raw_line": self.raw_line, + } + + +def compute_file_sha256(path: Path) -> str: + return hashlib.sha256(path.read_bytes()).hexdigest() + + +def extract_invariant_section_lines(spec_text: str) -> list[str]: + lines: list[str] = [] + in_section = False + for raw_line in spec_text.splitlines(): + stripped = raw_line.strip() + if INVARIANT_SECTION_PATTERN.match(stripped): + in_section = True + continue + if in_section and ANY_H2_PATTERN.match(stripped): + break + if in_section and stripped: + lines.append(stripped) + return lines + + +def parse_architecture_invariants(spec_text: str) -> tuple[list[ArchitectureInvariantRule], list[str]]: + """Parse structured invariant rules and free-form invariant statements.""" + section_lines = extract_invariant_section_lines(spec_text) + rules: list[ArchitectureInvariantRule] = [] + principles: list[str] = [] + + for line in section_lines: + m_import = FORBID_IMPORT_RULE_PATTERN.match(line) + if m_import: + rules.append( + ArchitectureInvariantRule( + kind="forbid_import", + source_glob=m_import.group(1).strip(), + target=m_import.group(2).strip(), + rationale=(m_import.group(3) or "").strip(), + raw_line=line, + ) + ) + continue + + m_pattern = FORBID_PATTERN_RULE_PATTERN.match(line) + if m_pattern: + rules.append( + ArchitectureInvariantRule( + kind="forbid_pattern", + source_glob=m_pattern.group(1).strip(), + target=m_pattern.group(2).strip().rstrip("`"), + rationale=(m_pattern.group(3) or "").strip(), + raw_line=line, + ) + ) + continue + + m_exist = MUST_EXIST_RULE_PATTERN.match(line) + if m_exist: + rules.append( + ArchitectureInvariantRule( + kind="must_exist", + source_glob=m_exist.group(1).strip(), + target="", + rationale=(m_exist.group(2) or "").strip(), + raw_line=line, + ) + ) + continue + + m_imm = IMMUTABLE_PATH_RULE_PATTERN.match(line) + if m_imm: + rules.append( + ArchitectureInvariantRule( + kind="immutable_path", + source_glob=m_imm.group(1).strip(), + target="", + rationale=(m_imm.group(2) or "").strip(), + raw_line=line, + ) + ) + continue + + if line.startswith(("-", "*")): + principles.append(line.lstrip("-* ").strip()) + + return rules, principles + + +def discover_top_level_specs(project: Path) -> list[dict[str, Any]]: + """Snapshot top-level specifications and their executable rules at bootstrap.""" + discovered: list[dict[str, Any]] = [] + candidates = [ + ("architecture_spec", "top_level_architecture_invariants", project / "architecture.md"), + ("architecture_spec", "top_level_architecture_invariants", project / "docs" / "architecture.md"), + ("requirements_spec", "goal_and_requirements", project / "goal-and-requirements.md"), + ("requirements_spec", "goal_and_requirements", project / "docs" / "goal-and-requirements.md"), + ] + seen_kinds: set[str] = set() + for kind, role, candidate_path in candidates: + if kind in seen_kinds or not candidate_path.is_file(): + continue + text = candidate_path.read_text(encoding="utf-8") + rules, principles = parse_architecture_invariants(text) + rel_path = str(candidate_path.relative_to(project)) + sha256_hex = compute_file_sha256(candidate_path) + entry: dict[str, Any] = { + "kind": kind, + "path": rel_path, + "role": role, + "sha256": sha256_hex, + } + if rules or principles: + entry["invariants"] = [r.to_dict() for r in rules] + for rule in entry["invariants"]: + if rule["kind"] == "immutable_path": + target = _project_rule_path(project, rule["source_glob"]) + if target is not None and target.is_file(): + rule["sha256"] = compute_file_sha256(target) + entry["principles"] = principles + discovered.append(entry) + seen_kinds.add(kind) + return discovered + + +def render_architecture_invariants_markdown(project: Path) -> str: + specs = discover_top_level_specs(project) + if not specs: + return "" + lines: list[str] = [ + "", + "## Repo-Wide Architecture Invariants", + "", + "- Do not violate or silently weaken the repository-wide invariants declared below during replans or code edits.", + "- Changed authority sources or immutable files fail verification until their registry locks are updated through a separately authorized review.", + ] + for spec in specs: + lines.append( + f"- Authority spec `{spec['path']}` (`{spec['kind']}`, sha256=`{spec['sha256'][:12]}...`)" + ) + for principle in spec.get("principles") or []: + lines.append(f" - Principle: {principle}") + for rule in spec.get("invariants") or []: + kind = rule.get("kind") + src = rule.get("source_glob") + target = rule.get("target") + rationale = f" ({rule['rationale']})" if rule.get("rationale") else "" + if kind == "forbid_import": + lines.append(f" - Invariant [`forbid_import`]: `{src} -> {target}`{rationale}") + elif kind == "forbid_pattern": + lines.append(f" - Invariant [`forbid_pattern`]: `{src} :: {target}`{rationale}") + elif kind in {"must_exist", "immutable_path"}: + lines.append(f" - Invariant [`{kind}`]: `{src}`{rationale}") + return "\n".join(lines) + "\n" + + +def _iter_project_code_files(project: Path) -> list[Path]: + ignore_dirs = {".git", ".loopx", ".codex", "__pycache__", "node_modules", ".venv", "venv"} + matched: list[Path] = [] + for root, dirs, files in os.walk(project, followlinks=False): + dirs[:] = sorted(d for d in dirs if d not in ignore_dirs) + for name in sorted(files): + path = Path(root) / name + if not path.is_symlink(): + matched.append(path) + return matched + + +def _project_rule_path(project: Path, relative: str) -> Path | None: + """Executable rules may only read paths inside the selected repository.""" + if not relative or Path(relative).is_absolute(): + return None + path = (project / relative).resolve() + return path if path.is_relative_to(project.resolve()) else None + + +def _matches_glob(rel_path: str, pattern: str) -> bool: + if fnmatch.fnmatch(rel_path, pattern): + return True + if pattern.endswith("/**"): + prefix = pattern[:-3].rstrip("/") + if rel_path == prefix or rel_path.startswith(prefix + "/"): + return True + return False + + +def _check_python_forbidden_imports( + file_path: Path, rel_path: str, forbidden_prefix: str +) -> list[str]: + try: + source = file_path.read_text(encoding="utf-8") + tree = ast.parse(source, filename=rel_path) + except (OSError, UnicodeError, SyntaxError): + return [f"{rel_path}: cannot inspect Python imports"] + + violations: list[str] = [] + for node in ast.walk(tree): + if isinstance(node, ast.Import): + for alias in node.names: + name = alias.name + if name == forbidden_prefix or name.startswith(forbidden_prefix + "."): + violations.append( + f"{rel_path}:{node.lineno}: forbidden import '{name}' " + f"(violates `forbid_import: -> {forbidden_prefix}`)" + ) + elif isinstance(node, ast.ImportFrom): + mod = node.module or "" + if mod == forbidden_prefix or mod.startswith(forbidden_prefix + "."): + violations.append( + f"{rel_path}:{node.lineno}: forbidden from-import '{mod}' " + f"(violates `forbid_import: -> {forbidden_prefix}`)" + ) + return violations + + +def verify_authority_source_digests( + goal: dict[str, Any], +) -> tuple[list[dict[str, Any]], list[str]]: + """Verify every local digest-bearing authority, independent of its domain. + + Legacy entries without a digest retain their previous behavior. This is a + read-only check: prose in active state cannot approve or replace a lock. + """ + diagnostics: list[dict[str, Any]] = [] + checks: list[str] = [] + project = Path(str(goal.get("repo") or "")).expanduser() + goal_id = str(goal.get("id") or "") + for source in goal.get("authority_sources") or []: + if not isinstance(source, dict) or not source.get("sha256"): + continue + relative = str(source.get("path") or "") + path = project / relative + code = "" + if not relative or not path.is_file(): + code = "authority_source_missing" + else: + try: + actual = compute_file_sha256(path) + except OSError: + code = "authority_source_unreadable" + else: + if actual != source["sha256"]: + code = "authority_source_digest_mismatch" + if code: + diagnostics.append(contract_error_diagnostic( + code=code, + message=f"{goal_id}: authority source `{relative}` failed verification ({code})", + goal_id=goal_id, + )) + else: + checks.append(f"{goal_id}: authority source `{relative}` digest verified") + return diagnostics, checks + + +def verify_executable_architecture_invariants( + goal: dict[str, Any], +) -> tuple[list[dict[str, Any]], list[str]]: + """Inspect bootstrap-snapshotted software rules; never infer new authority. + + Import checks cover Python static imports only. Prose principles are + guidance, and executable rules neither approve changes nor write state. + """ + diagnostics: list[dict[str, Any]] = [] + checks: list[str] = [] + goal_id = str(goal.get("id") or "") + project = Path(str(goal.get("repo") or "")).expanduser().resolve() + code_files: list[Path] | None = None + for source in goal.get("authority_sources") or []: + if not isinstance(source, dict): + continue + rules = source.get("invariants") + if not isinstance(rules, list): + continue + hits: list[str] = [] + for item in rules: + if not isinstance(item, dict): + hits.append("invalid executable invariant record") + continue + kind = str(item.get("kind") or "") + pattern = str(item.get("source_glob") or "") + target = _project_rule_path(project, pattern) + if target is None: + hits.append(f"invalid repository-relative invariant path `{pattern}`") + continue + if kind == "must_exist": + if not target.exists(): + hits.append(f"required architectural artifact `{pattern}` is missing") + elif kind == "immutable_path": + try: + digest = compute_file_sha256(target) + except OSError: + hits.append(f"immutable artifact `{pattern}` is missing or unreadable") + else: + if not item.get("sha256") or digest != item["sha256"]: + hits.append(f"immutable artifact `{pattern}` differs from its bootstrap lock") + elif kind in {"forbid_import", "forbid_pattern"}: + forbidden = str(item.get("target") or "") + try: + regex = re.compile(forbidden) if kind == "forbid_pattern" else None + except re.error: + hits.append(f"invalid regular expression in `{pattern}` invariant") + continue + if code_files is None: + code_files = _iter_project_code_files(project) + for path in code_files: + relative = path.relative_to(project).as_posix() + if not _matches_glob(relative, pattern): + continue + if kind == "forbid_import": + if path.suffix == ".py": + hits.extend(_check_python_forbidden_imports(path, relative, forbidden)) + else: + try: + lines = path.read_text(encoding="utf-8").splitlines() + except (OSError, UnicodeError): + hits.append(f"cannot inspect `{relative}` for forbidden patterns") + continue + for number, line in enumerate(lines, 1): + if regex is not None and regex.search(line): + hits.append(f"{relative}:{number}: forbidden pattern `{forbidden}`") + else: + hits.append(f"unsupported executable invariant `{kind}`") + for hit in hits: + diagnostics.append(contract_error_diagnostic( + code="architecture_invariant_violation", + message=f"{goal_id}: {hit}", goal_id=goal_id, + )) + checks.append( + f"{goal_id}: executable invariants in `{source.get('path', '')}` checked " + f"(rules={len(rules)}, violations={len(hits)})" + ) + return diagnostics, checks diff --git a/tests/test_architecture_invariants.py b/tests/test_architecture_invariants.py new file mode 100644 index 0000000000..c988079da7 --- /dev/null +++ b/tests/test_architecture_invariants.py @@ -0,0 +1,268 @@ +from __future__ import annotations + +import hashlib +import json +from pathlib import Path + +import pytest + +from loopx.bootstrap import bootstrap_project, build_goal_entry, render_state_markdown +from loopx.contract import check_contract +from loopx.control_plane.goals.architecture_invariants import ( + discover_top_level_specs, + parse_architecture_invariants, + verify_authority_source_digests, + verify_executable_architecture_invariants, +) + + +def test_parse_and_discover_top_level_specs(tmp_path: Path) -> None: + arch = tmp_path / "architecture.md" + arch.write_text( + """# Architecture + +## Repo-Wide Invariants +- Keep presentation separate from persistence. +- `forbid_import: api/** -> db` — API must not import DB +- `forbid_pattern: api/** :: \\bexecute_raw_sql\\(` # No raw SQL +- `must_exist: services/audit.py` — Audit module required +""", + encoding="utf-8", + ) + req = tmp_path / "goal-and-requirements.md" + req.write_text("# Goal and Requirements\n\n## Goal\nFast and safe.\n", encoding="utf-8") + + rules, principles = parse_architecture_invariants(arch.read_text(encoding="utf-8")) + assert len(rules) == 3 + assert principles == ["Keep presentation separate from persistence."] + + specs = discover_top_level_specs(tmp_path) + assert len(specs) == 2 + assert specs[0]["kind"] == "architecture_spec" + assert specs[0]["path"] == "architecture.md" + assert len(specs[0]["invariants"]) == 3 + assert specs[1]["kind"] == "requirements_spec" + + +def test_evaluate_goal_architecture_invariants_catches_violations_and_tampering( + tmp_path: Path, +) -> None: + (tmp_path / "api").mkdir() + (tmp_path / "services").mkdir() + (tmp_path / "db").mkdir() + + (tmp_path / "architecture.md").write_text( + """# Architecture + +## Repo-Wide Invariants +- `forbid_import: api/** -> db` +- `must_exist: services/audit.py` +""", + encoding="utf-8", + ) + (tmp_path / "services" / "audit.py").write_text("def audit(): pass\n", encoding="utf-8") + (tmp_path / "api" / "handler.py").write_text("def handle(): return 200\n", encoding="utf-8") + + goal_entry = build_goal_entry( + project=tmp_path, + goal_id="test-goal", + domain="backend", + role="primary", + parent_goal_id=None, + state_file=tmp_path / ".loopx" / "ACTIVE_GOAL_STATE.md", + goal_doc=None, + adapter_kind="read_only_project_map_v0", + adapter_status="connected-read-only", + next_probe=None, + spawn_allowed=False, + max_children=0, + allowed_domains=[], + write_scope=[], + execution_profile=None, + ) + state_md = render_state_markdown( + project=tmp_path, + goal_id="test-goal", + adapter_kind="read_only_project_map_v0", + objective="Test objective", + updated_at="2026-10-25T21:00:00+08:00", + goal_doc=None, + execution_profile=None, + ) + assert "## Repo-Wide Architecture Invariants" in state_md + + # 1. Compliant state passes cleanly + diags, checks = _verify(goal_entry) + assert diags == [] + assert len(checks) == 2 + + # 2. Introduce forbidden import and delete required file + (tmp_path / "api" / "handler.py").write_text("import db.conn\n", encoding="utf-8") + (tmp_path / "services" / "audit.py").unlink() + diags, _ = _verify(goal_entry) + codes = [d["code"] for d in diags] + assert codes == ["architecture_invariant_violation", "architecture_invariant_violation"] + + # 3. Unapproved mutation of architecture.md is caught even if rules are erased + (tmp_path / "architecture.md").write_text("# Architecture\n", encoding="utf-8") + diags_tamper, _ = _verify(goal_entry) + assert any(d["code"] == "authority_source_digest_mismatch" for d in diags_tamper) + assert sum(1 for d in diags_tamper if d["code"] == "architecture_invariant_violation") == 2 + + # 4. Prose cannot approve a lock change. A reviewed registry update can. + (tmp_path / "services" / "audit.py").write_text("def audit(): pass\n", encoding="utf-8") + (tmp_path / "api" / "handler.py").write_text("def handle(): return 200\n", encoding="utf-8") + new_sha = hashlib.sha256((tmp_path / "architecture.md").read_bytes()).hexdigest() + (tmp_path / ".loopx").mkdir() + (tmp_path / ".loopx" / "ACTIVE_GOAL_STATE.md").write_text( + state_md + f"\n- [x] USER-GATE-1 (`approved_architecture_digest: {new_sha}`)\n", + encoding="utf-8", + ) + assert verify_authority_source_digests(goal_entry)[0] + goal_entry["authority_sources"][0]["sha256"] = new_sha + assert _verify(goal_entry)[0] == [] + + +def _verify(goal): + digest_diags, digest_checks = verify_authority_source_digests(goal) + code_diags, code_checks = verify_executable_architecture_invariants(goal) + return digest_diags + code_diags, digest_checks + code_checks + + +@pytest.fixture +def bootstrapped(tmp_path): + project = tmp_path / "project" + project.mkdir() + (project / "architecture.md").write_text( + "## Repo-Wide Invariants\n- `must_exist: required.txt`\n", encoding="utf-8" + ) + (project / "goal-and-requirements.md").write_text("# Requirements\n", encoding="utf-8") + (project / "intent.md").write_text("# Explicit goal\n", encoding="utf-8") + runtime = tmp_path / "runtime" + registry = project / ".loopx" / "registry.json" + bootstrap_project( + project=project, registry_path=registry, runtime_root=runtime, + goal_id="digest-test", objective="Preserve accepted sources", domain="research", + role="primary", parent_goal_id=None, state_file=None, goal_doc=Path("intent.md"), + adapter_kind="read_only_project_map_v0", adapter_status="connected-read-only", + next_probe=None, spawn_allowed=False, max_children=0, allowed_domains=[], + write_scope=[], force=False, dry_run=False, sync_global=False, + ) + return project, registry, runtime + + +def test_bootstrap_explicit_goal_doc_and_discovered_sources_have_digests(bootstrapped): + project, registry, _ = bootstrapped + goal = json.loads(registry.read_text())["goals"][0] + assert {s["path"] for s in goal["authority_sources"]} == { + "intent.md", "architecture.md", "goal-and-requirements.md" + } + for source in goal["authority_sources"]: + assert source["sha256"] == hashlib.sha256((project / source["path"]).read_bytes()).hexdigest() + assert verify_authority_source_digests(goal)[0] == [] + + +@pytest.mark.parametrize("include_probe", [False, True]) +def test_contract_toggle_preserves_digest_lock(bootstrapped, include_probe, monkeypatch): + project, registry, runtime = bootstrapped + if not include_probe: + def unexpected_probe(_goal): + raise AssertionError("disabled executable scanning must not run") + + monkeypatch.setattr("loopx.contract.verify_executable_architecture_invariants", unexpected_probe) + + def check(): + return check_contract( + registry_path=registry, runtime_root_override=str(runtime), scan_roots=[], + limit=20, include_architecture_invariants=include_probe, + include_public_boundary_scan=False, + ) + + initial = check() + codes = {d["code"] for d in initial["error_diagnostics"]} + assert ("architecture_invariant_violation" in codes) == include_probe + assert initial["ok"] == (not include_probe), initial["errors"] + (project / "required.txt").write_text("present", encoding="utf-8") + assert check()["ok"] + (project / "intent.md").write_text("changed", encoding="utf-8") + assert "authority_source_digest_mismatch" in { + d["code"] for d in check()["error_diagnostics"] + } + (project / "intent.md").unlink() + assert "authority_source_missing" in { + d["code"] for d in check()["error_diagnostics"] + } + + +def test_legacy_unhashed_authority_is_unchanged(tmp_path): + goal = {"id": "legacy", "repo": str(tmp_path), "authority_sources": [ + {"kind": "document", "path": "missing.md"} + ]} + assert verify_authority_source_digests(goal) == ([], []) + assert verify_executable_architecture_invariants(goal) == ([], []) + + +def test_immutable_snapshot_and_pattern_rules(tmp_path): + (tmp_path / "fixed.txt").write_text("stable", encoding="utf-8") + (tmp_path / "code.py").write_text("safe = True\n", encoding="utf-8") + spec = tmp_path / "architecture.md" + spec.write_text( + "## Invariants\n- `immutable_path: fixed.txt`\n" + "- `forbid_pattern: *.py :: UNSAFE`\n", encoding="utf-8" + ) + goal = {"id": "rules", "repo": str(tmp_path), + "authority_sources": discover_top_level_specs(tmp_path)} + assert verify_executable_architecture_invariants(goal)[0] == [] + (tmp_path / "fixed.txt").write_text("changed", encoding="utf-8") + (tmp_path / "code.py").write_text("UNSAFE = True\n", encoding="utf-8") + assert len(verify_executable_architecture_invariants(goal)[0]) == 2 + spec.unlink() + # A deleted spec cannot erase its locked executable rules. + assert len(verify_executable_architecture_invariants(goal)[0]) == 2 + assert verify_authority_source_digests(goal)[0][0]["code"] == "authority_source_missing" + + +@pytest.mark.parametrize("rule", [ + {"kind": "forbid_pattern", "source_glob": "*.py", "target": "["}, + {"kind": "must_exist", "source_glob": "../outside.txt"}, + {"kind": "immutable_path", "source_glob": "missing.txt"}, +]) +def test_invalid_rules_fail_closed(tmp_path, rule): + goal = {"id": "invalid", "repo": str(tmp_path), "authority_sources": [ + {"path": "architecture.md", "invariants": [rule]} + ]} + assert verify_executable_architecture_invariants(goal)[0] + + +@pytest.mark.parametrize("filename", ["architecture.md", "goal-and-requirements.md"]) +def test_docs_discovery_locks_documents_without_executable_rules(tmp_path, filename): + docs = tmp_path / "docs" + docs.mkdir() + source = docs / filename + source.write_text("# Accepted document\n", encoding="utf-8") + entries = discover_top_level_specs(tmp_path) + assert len(entries) == 1 + assert entries[0]["path"] == f"docs/{filename}" + assert entries[0]["sha256"] == hashlib.sha256(source.read_bytes()).hexdigest() + + +def test_bootstrap_nonexistent_goal_doc_preserves_unhashed_authority(tmp_path): + project = tmp_path / "project" + project.mkdir() + registry = project / ".loopx" / "registry.json" + result = bootstrap_project( + project=project, registry_path=registry, runtime_root=tmp_path / "runtime", + goal_id="future-document", objective="Prepare a future goal document", domain="research", + role="primary", parent_goal_id=None, state_file=None, goal_doc=Path("future.md"), + adapter_kind="read_only_project_map_v0", adapter_status="connected-read-only", + next_probe=None, spawn_allowed=False, max_children=0, allowed_domains=[], + write_scope=[], force=False, dry_run=False, sync_global=False, + ) + assert result["ok"] is True + assert result["goal_doc_exists"] is False + assert not (project / "future.md").exists() + goal = json.loads(registry.read_text(encoding="utf-8"))["goals"][0] + assert goal["authority_sources"] == [{ + "kind": "goal_doc", "path": "future.md", "role": "primary_goal_document", + }] + assert verify_authority_source_digests(goal) == ([], [])