feat(contract): verify explicitly pinned source content

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

loopx/contract.py

@@ -20,6 +20,7 @@
     contract_error_views,
     project_contract_health_for_goal,
 )
+from .control_plane.goals.source_integrity import verify_authority_source_digests
 from .control_plane.goals.activation import (
     GoalActivationState,
     goal_activation_state,
@@ -873,6 +874,30 @@ def _active_state_projection_gap_warnings(
     return warnings
 
 
+def _authority_source_diagnostics(
+    registry: dict[str, Any],
+    *,
+    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)
+        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)
@@ -1297,6 +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)
+    source_diagnostics, source_checks = _authority_source_diagnostics(
+        registry,
+        goal_id_filter=goal_id_filter,
+        activation_state_filter=activation_state_filter,
+    )
+    error_diagnostics.extend(source_diagnostics)
+    checks.extend(source_checks)
     warnings.extend(
         _active_state_projection_gap_warnings(
             registry,

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

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