From d6938387511546d0b14136ddd8f73959a189e43a 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 1/2] 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 7910965ea2..e45d0ba933 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, @@ -873,6 +877,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) @@ -1227,6 +1260,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, @@ -1297,6 +1331,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) == ([], []) From b9f99f115fff5ae15680ae8e060b483f372079b2 Mon Sep 17 00:00:00 2001 From: huangruiteng Date: Tue, 27 Oct 2026 16:46:10 +0800 Subject: [PATCH 2/2] refactor(contract): separate explicit source pins from evolving knowledge Signed-off-by: huangruiteng --- docs/architecture.md | 34 ++ loopx/bootstrap.py | 20 +- loopx/contract.py | 20 +- .../goals/architecture_invariants.py | 381 ------------------ loopx/control_plane/goals/source_integrity.py | 56 +++ tests/test_architecture_invariants.py | 268 ------------ tests/test_authority_source_integrity.py | 148 +++++++ 7 files changed, 246 insertions(+), 681 deletions(-) delete mode 100644 loopx/control_plane/goals/architecture_invariants.py create mode 100644 loopx/control_plane/goals/source_integrity.py delete mode 100644 tests/test_architecture_invariants.py create mode 100644 tests/test_authority_source_integrity.py diff --git a/docs/architecture.md b/docs/architecture.md index c534875f05..dbe6826919 100644 --- a/docs/architecture.md +++ b/docs/architecture.md @@ -43,6 +43,40 @@ The core repository intentionally avoids domain logic. A data experiment goal, a note-maintenance goal, and a harness self-improvement goal should share the same runtime and contract, but use different adapters. +## Source content, knowledge and accepted constraints + +Repository knowledge and retrieved memory may evolve. A source path, familiar +filename or `bootstrap --goal-doc` declaration does not make its contents an +immutable contract, enable architecture scanning, or authorize new obligations. +Memory providers own retrieval and revision/freshness; existing acceptance and +authority owners decide which constraints apply. Repository memory and semantic +preference adapters remain advisory under their existing application contracts. + +For an explicitly pinned local source, the existing private Goal registry may +include `sha256` beside `path` in `authority_sources`. This is the SHA-256 of the +accepted **file bytes**, distinct from a redacted source-reference hash or the +registry witness. `loopx check` and the status health collector verify only such +pins, using the existing Goal diagnostic scope. Changed, missing or unreadable +content is reported without changing documents, registry locks or memory. +Unpinned sources retain their previous behavior. Bootstrap never generates a +pin or discovers authority from specification filenames. + +To recover, restore the accepted file bytes, or separately review the source +change and update its explicit private registry pin under the existing registry +write authority. Rerun `loopx check` and read back Goal health. Current document +text, an Agent note or recalled memory cannot approve that update. This bounded +integrity adapter does not parse source policy, infer approval, freeze knowledge, +or introduce another memory provider or a new approval workflow. + +中文：仓库知识与召回记忆可以持续更新。文件名、来源路径或 `--goal-doc` 声明不会 +自动把正文升级为不可变合同，也不启用架构扫描。检索与新鲜度由既有 memory provider +负责；哪些约束生效仍归既有验收和权限 owner。只有在私有 Goal registry 的 +`authority_sources` 中显式为 `path` 设置文件字节 `sha256` 时，check/status 才校验该 +固定版本；它与来源引用的脱敏哈希、registry witness 不同。普通来源行为不变， +bootstrap 不自动加锁或发现权限。校验不修改文件、记忆或摘要。恢复时还原已接受的 +字节，或在既有写入权限下单独评审来源变更并更新 pin，再回读健康状态；当前正文、 +Agent 声明和记忆不能自行批准换锁。此边界不新增 memory provider、源码政策或审批流。 + ## Control Plane As Effect Interpreter The six layers are not only storage surfaces. Together they interpret one diff --git a/loopx/bootstrap.py b/loopx/bootstrap.py index ab9f435d33..aac03ac5dd 100644 --- a/loopx/bootstrap.py +++ b/loopx/bootstrap.py @@ -59,11 +59,6 @@ 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 @@ -95,16 +90,9 @@ def resolve_project_path(project: Path, path: Path | None) -> Path | None: def render_authority_sources(project: Path, goal_doc: Path | None) -> str: - 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: + if not goal_doc: return "- No explicit goal document was provided during bootstrap." - return "\n".join(entries) + return f"- Primary goal document: `{rel_or_abs(goal_doc, project)}`" def repair_missing_todo_source_sections(state_text: str) -> tuple[str, list[str]]: @@ -184,7 +172,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 @@ -234,10 +222,8 @@ 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 e45d0ba933..f3f01f1fd2 100644 --- a/loopx/contract.py +++ b/loopx/contract.py @@ -20,10 +20,7 @@ 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.source_integrity import verify_authority_source_digests from .control_plane.goals.activation import ( GoalActivationState, goal_activation_state, @@ -877,10 +874,9 @@ def _active_state_projection_gap_warnings( return warnings -def _authority_and_invariant_diagnostics( +def _authority_source_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]]: @@ -897,10 +893,6 @@ def _authority_and_invariant_diagnostics( ): 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 @@ -1260,7 +1252,6 @@ 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, @@ -1331,14 +1322,13 @@ 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( + source_diagnostics, source_checks = _authority_source_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) + error_diagnostics.extend(source_diagnostics) + checks.extend(source_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 deleted file mode 100644 index f76b2f7e76..0000000000 --- a/loopx/control_plane/goals/architecture_invariants.py +++ /dev/null @@ -1,381 +0,0 @@ -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/loopx/control_plane/goals/source_integrity.py b/loopx/control_plane/goals/source_integrity.py new file mode 100644 index 0000000000..ca8cea5189 --- /dev/null +++ b/loopx/control_plane/goals/source_integrity.py @@ -0,0 +1,56 @@ +"""Read-only content verification for explicitly registered local sources. + +This filesystem adapter measures bytes and uses existing Goal health diagnostics. +It neither discovers authority from filenames nor applies software-domain rules. +""" +from __future__ import annotations + +import hashlib +from pathlib import Path +from typing import Any + +from .contract_health import contract_error_diagnostic + + +def compute_file_sha256(path: Path) -> str: + with path.open("rb") as source: + return hashlib.file_digest(source, "sha256").hexdigest() + + +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 diff --git a/tests/test_architecture_invariants.py b/tests/test_architecture_invariants.py deleted file mode 100644 index c988079da7..0000000000 --- a/tests/test_architecture_invariants.py +++ /dev/null @@ -1,268 +0,0 @@ -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) == ([], []) diff --git a/tests/test_authority_source_integrity.py b/tests/test_authority_source_integrity.py new file mode 100644 index 0000000000..8cb0a83bd7 --- /dev/null +++ b/tests/test_authority_source_integrity.py @@ -0,0 +1,148 @@ +"""Explicit document locks do not imply software policies or discover authority.""" +from __future__ import annotations + +import hashlib +import json +from pathlib import Path + +import pytest + +from loopx.cli import main as cli_main +from loopx.control_plane.goals.source_integrity import verify_authority_source_digests + + +@pytest.fixture +def connected(tmp_path, capsys): + project = tmp_path / "project" + project.mkdir() + # Filenames and executable-looking prose do not authorize a code scanner. + (project / "architecture.md").write_text( + "## Repo-Wide Invariants\n- `must_exist: absent.txt`\n" + "- `forbid_import: *.py -> pkg.db`\n" + ) + (project / "goal-and-requirements.md").write_text("# Requirements\n") + (project / "intent.md").write_text("# Explicit goal\n") + (project / "README.md").write_text("# Public fixture\n") + registry, runtime = project / ".loopx" / "registry.json", tmp_path / "runtime" + + def run(*args): + code = cli_main(["--format", "json", "--registry", str(registry), + "--runtime-root", str(runtime), *args]) + return code, json.loads(capsys.readouterr().out) + + def connect(*args): + return run("bootstrap", "--project", str(project), "--goal-id", "source-lock", + "--objective", "Preserve explicit source evidence", "--domain", "research", + "--no-global-sync", *args) + + def check(): + return run("check", "--scan-path", str(project / "README.md")) + + return project, registry, connect, check + + +def test_default_bootstrap_does_not_discover_rules_or_generate_obligations(connected): + project, registry, connect, check = connected + code, result = connect() + assert code == 0 and result["ok"] + goal = json.loads(registry.read_text())["goals"][0] + assert goal["authority_sources"] == [] + state = (project / goal["state_file"]).read_text() + assert "## Repo-Wide Architecture Invariants" not in state + assert "absent.txt" not in state and "pkg.db" not in state + assert check()[0] == 0 + # Neither direct nor from-import is inspected by ordinary health checks. + for code in ("import pkg.db\n", "from pkg import db\n", "from .pkg import db\n"): + (project / "source.py").write_text(code) + assert check()[0] == 0 + + +def test_goal_document_is_mutable_until_its_content_is_explicitly_pinned(connected): + project, registry, connect, check = connected + assert connect("--goal-doc", "intent.md")[0] == 0 + goal = json.loads(registry.read_text())["goals"][0] + assert goal["authority_sources"] == [{ + "kind": "goal_doc", "path": "intent.md", "role": "primary_goal_document", + }] + assert check()[0] == 0 + data = json.loads(registry.read_text()) + data["goals"][0]["authority_sources"][0]["sha256"] = hashlib.sha256( + (project / "intent.md").read_bytes()).hexdigest() + registry.write_text(json.dumps(data)) + before = registry.read_bytes() + (project / "intent.md").write_text("# Changed source\n") + code, failed = check() + assert code == 1 + assert any(d["code"] == "authority_source_digest_mismatch" + and d["goal_id"] == "source-lock" for d in failed["error_diagnostics"]) + assert registry.read_bytes() == before + # Restoring the accepted bytes recovers without rewriting the lock. + (project / "intent.md").write_text("# Explicit goal\n") + assert check()[0] == 0 + assert registry.read_bytes() == before + + +def test_state_prose_cannot_approve_a_source_lock_update(connected): + project, registry, connect, check = connected + assert connect("--goal-doc", "intent.md")[0] == 0 + data = json.loads(registry.read_text()) + goal = data["goals"][0] + goal["authority_sources"][0]["sha256"] = hashlib.sha256( + (project / "intent.md").read_bytes()).hexdigest() + registry.write_text(json.dumps(data)) + (project / "intent.md").write_text("# Revised requirement\n") + new_digest = hashlib.sha256((project / "intent.md").read_bytes()).hexdigest() + goal = data["goals"][0] + state = project / goal["state_file"] + state.write_text(state.read_text() + f"\nApproved source digest: {new_digest}\n") + assert check()[0] == 1 + # An explicit registry lock revision is separate from current document prose. + goal["authority_sources"][0]["sha256"] = new_digest + registry.write_text(json.dumps(data)) + assert check()[0] == 0 + assert json.loads(registry.read_text())["goals"][0]["authority_sources"][0]["sha256"] == new_digest + + +def test_missing_explicit_document_preserves_unhashed_registration(connected): + project, registry, connect, check = connected + assert connect("--goal-doc", "future.md")[0] == 0 + source = json.loads(registry.read_text())["goals"][0]["authority_sources"][0] + assert source == {"kind": "goal_doc", "path": "future.md", "role": "primary_goal_document"} + assert not (project / "future.md").exists() + assert check()[0] == 0 + + +def test_missing_and_unreadable_content_are_scoped_read_only_failures(tmp_path, monkeypatch): + source = tmp_path / "intent.md" + source.write_text("accepted") + goal = {"id": "source-lock", "repo": str(tmp_path), "authority_sources": [{ + "path": "intent.md", "sha256": hashlib.sha256(source.read_bytes()).hexdigest(), + }]} + original = Path.open + + def denied(path, *args, **kwargs): + if path == source: + raise PermissionError("unreadable source") + return original(path, *args, **kwargs) + + with monkeypatch.context() as context: + context.setattr(Path, "open", denied) + assert verify_authority_source_digests(goal)[0][0]["code"] == "authority_source_unreadable" + source.unlink() + assert verify_authority_source_digests(goal)[0][0]["code"] == "authority_source_missing" + + +def test_unhashed_sources_retain_previous_behavior(tmp_path): + goal = {"id": "unhashed", "repo": str(tmp_path), "authority_sources": [ + {"kind": "document", "path": "missing.md"}, + ]} + assert verify_authority_source_digests(goal) == ([], []) + + +def test_unpinned_goal_doc_changes_do_not_block_normal_health(connected): + project, registry, connect, check = connected + assert connect("--goal-doc", "intent.md")[0] == 0 + before = registry.read_bytes() + (project / "intent.md").write_text("# Updated repository knowledge\n") + assert check()[0] == 0 + assert registry.read_bytes() == before