Asset record support: budget-neutral reader, declared repo_meta + assessment block, trust-rule tests

assessor/assessment/assemble.py

@@ -3,6 +3,7 @@
 
 from . import content as content_mod
 from . import coords, license as license_mod, probes, purpose, subject, versions
+from . import record as record_mod
 from .shapes import fact, unknown
 from .fingerprint import build_payload, compute_fingerprint
 from .registry import (TYPE_MODULES, TIER_CONF, REGISTRY_VERSION, FULL_TAXONOMY,
@@ -52,6 +53,11 @@ def build(repo_url: str, content: dict[str, str], commit_sha: str | None,
     if subdir:
         content = subject.filter_content_to_subdir(content, subdir)
         paths = subject.filter_paths_to_subdir(paths, subdir)
+    # `declared` reads `content` AFTER the subdir filter above, so a subdir subject
+    # reads its own re-rooted asset-record.json (record.py's parse_record docstring).
+    # Deliberately NOT threaded into build_payload/compute_fingerprint below — it is
+    # served-only, never hashed (spec §7 trust rule: nothing in record.py classifies).
+    declared = record_mod.declared_block(content)
     b = bucket_b or {}
     meta = {"description": b.get("description"),
             "topics": [t.lower() for t in (b.get("topics") or []) if isinstance(t, str)],
@@ -237,7 +243,7 @@ def build(repo_url: str, content: dict[str, str], commit_sha: str | None,
         "license": {"spdx": lic["spdx"], "caveat": lic["caveat"]},
         "coordinates": coordinates, "versions": versions.build(releases, commit_sha),
         "risk": risk_flags, "known_unknowns": known, "topics": topics,
-        "coverage_probes": coverage_probes,
+        "coverage_probes": coverage_probes, "declared": declared,
         # Audit trail for invariant-gated promotions (Task 7): which type was promoted,
         # from which probe state, on which LLM citation, at which confidence. Empty for
         # every repo that had no citation-backed candidate — i.e. almost all of them.

assessor/assessment/content.py

@@ -11,6 +11,7 @@
 from collections import Counter
 
 from ..vendored import _valid_slug
+from .record import RECORD_BASENAME, RECORD_MAX_BYTES
 
 _PHASE1 = ("README.md", "README.rst", "package.json", "pyproject.toml", "mcp.json",
            "server.json", "smithery.yaml", "smithery.json", "Dockerfile",
@@ -115,11 +116,12 @@ def _get(path):
 _SRC_MAX_FILES = 120            # was 60
 _SRC_MAX_BYTES = 1_600 * 1024   # was 800 * 1024
 
-ALLOWLIST_VERSION = 7   # bump on every allowlist widening; acquire stores it and
+ALLOWLIST_VERSION = 8   # bump on every allowlist widening; acquire stores it and
                         # SHA-reuse requires a match so stale snapshots refetch (§4.2)
                         # 4->5: dependency manifests in subdirs + non-Python/JS ecosystems.
                         # 5->6: marker-driven selection + raised source caps.
                         # 6->7: agent_run_shape v3 co-occurrence changes marker-driven selection.
+                        # 7->8: asset-record.json budget-neutral selection (authoring MCP spec §6).
 _AGENT_MANIFESTS = ("agents.yaml", "tasks.yaml", "langgraph.json", "agent.json", "agent-card.json")
 _MANIFEST_MAX_FILES, _MANIFEST_MAX_BYTES = 8, 64 * 1024
 
@@ -154,6 +156,16 @@ def dep_manifest_paths(paths) -> list[str]:
 _NEWTYPE_MANIFESTS = ("SKILL.md", "dataset_infos.json")
 _SKILL_MAX_FILES = 40
 
+# asset-record.json (spec 2026-08-09-authoring-mcp §7 channel 4): selected OUTSIDE
+# every shared budget — never counts toward total/mtotal/src_files, never sets
+# `capped` (which asserts SOURCE truncation), never displaces any other file.
+# Oversized bodies are skipped (the reader would reject them anyway); overflow
+# beyond the file cap is dropped silently — matches `_SKILL_MAX_FILES` (a legitimate
+# multi-asset monorepo can legitimately have >20 subdir records, so the cap isn't
+# about rejecting that case). The per-file RECORD_MAX_BYTES cap bounds the worst
+# case at 640KiB (40 * 16KiB) of budget-neutral reads.
+_RECORD_MAX_FILES = 40
+
 # Tool-definition trees (e.g. a TS MCP server's src/tools/*.ts, spread across many
 # per-tool subdirectories) and package/module entrypoints (e.g. a monorepo package's
 # packages/<name>/src/index.ts) routinely lose the density sort to a single denser,
@@ -319,6 +331,10 @@ def select_source_paths(candidates: dict, *, skip=frozenset(), hits=()) -> tuple
     `_AGENT_MANIFESTS`. Without the dedup a source-file entrypoint consumes a cap
     slot/byte budget and manifests double-count, dropping a different tail file than
     REST would.
+
+    A terminal pass then adds any asset-record.json candidates (`_RECORD_MAX_FILES`,
+    `RECORD_MAX_BYTES`) outside every budget above — see the comment at their
+    definitions for why that pass never counts toward SOURCE totals or `capped`.
     """
     blob_paths = [p for p in candidates if p not in skip and _is_source(p)]
 
@@ -480,4 +496,11 @@ def _lift(names):
         src_files += 1
         total += size
 
+    record_paths = sorted(
+        p for p in candidates
+        if p.rsplit("/", 1)[-1] == RECORD_BASENAME
+        and p not in skip and p not in selected
+        and not any(x in p.lower() for x in _SRC_EXCLUDE)
+        and _candidate_size(candidates, p) <= RECORD_MAX_BYTES)
+    selected.extend(record_paths[:_RECORD_MAX_FILES])
     return selected, capped

assessor/assessment/git_fetch.py

@@ -24,6 +24,7 @@
 
 from . import content
 from . import markers
+from . import record
 from ..vendored import _REPO_KEY_RE, _SHA_RE
 
 _ALLOWED_HOSTS = frozenset({"github.com"})
@@ -175,13 +176,28 @@ def _read_blobs(git_dir: Path, shas: list[str], *, run_bytes: BytesRunner, timeo
 _MAX_MARKER_HITS = 500
 
 
+def _skip_marker_scan(path: str) -> bool:
+    """asset-record.json is excluded from marker scanning (spec §7 channel 2):
+    a record naming provider env vars or 'mcp server' must not generate
+    persisted marker_hits from its own text."""
+    return path.rsplit("/", 1)[-1] == record.RECORD_BASENAME
+
+
 def _scan_present_blobs(bodies: dict[str, str]) -> list[dict]:
     """Marker hits across already-read blob bodies, in path order. Pure over its argument:
-    the caller decides which blobs are present and affordable to read."""
+    the caller decides which blobs are present and affordable to read.
+
+    This is the SOLE production call site of `markers.scan_text` in this module — every
+    scan pass in `GitContentFetcher.fetch` (manifest pre-scan, source pre-scan, and the
+    final re-scan over the fully-selected `files`) routes through here, so the
+    `_skip_marker_scan` guard below covers all of them without needing a second guard
+    elsewhere."""
     hits: list[dict] = []
     for path in sorted(bodies):
         if not markers.is_scannable(path):
             continue
+        if _skip_marker_scan(path):
+            continue
         for h in markers.scan_text(path, bodies[path]):
             hits.append(h)
             if len(hits) >= _MAX_MARKER_HITS:
@@ -393,10 +409,10 @@ def _ls_tree(self, git_dir: Path, sha: str) -> tuple[list[tuple[str, str, str]],
         )
         entries: list[tuple[str, str, str]] = []
         capped = False
-        for record in out.split("\0"):
-            if not record:
+        for entry in out.split("\0"):
+            if not entry:
                 continue
-            meta, sep, path = record.partition("\t")
+            meta, sep, path = entry.partition("\t")
             if not sep:
                 continue
             mode, _type, blob_sha = meta.split(" ")

assessor/assessment/record.py

@@ -0,0 +1,115 @@
+"""Tolerant reading of asset-record.json — the authoring-time foundational record
+(spec 2026-08-09-authoring-mcp-design.md §4, §9 D1).
+
+Malformed = absent (never raises, never fails an assessment): a record-level
+defect (bad JSON, wrong/missing record_version, oversized body, non-dict top
+level) returns None; a FIELD-level defect drops that field and keeps the rest.
+
+TRUST RULE (spec §7): nothing in this module feeds classification. declared_block
+produces fact envelopes for the served `declared` section only."""
+from __future__ import annotations
+
+import json
+
+from .shapes import fact
+
+RECORD_BASENAME = "asset-record.json"
+RECORD_MAX_BYTES = 16 * 1024
+_MAX_STR = 200
+_MAX_DEP = 100
+_MAX_DEPS = 50
+_SOURCE = "declared: asset-record.json"
+
+
+def _clean_str(v) -> str | None:
+    return v if isinstance(v, str) and 0 < len(v) <= _MAX_STR else None
+
+
+def _clean_model_access(v) -> dict | None:
+    if not isinstance(v, dict):
+        return None
+    mode = v.get("mode")
+    if mode not in ("pinned", "byo"):
+        return None
+    provider, model = _clean_str(v.get("provider")), _clean_str(v.get("model"))
+    if mode == "pinned" and not (provider and model):
+        return None                     # a pin that names no model asserts nothing
+    if mode == "byo":
+        provider = model = None
+    return {"mode": mode, "provider": provider, "model": model}
+
+
+def _clean_technologies(v) -> dict | None:
+    if not isinstance(v, dict):
+        return None
+    deps_in = v.get("dependencies")
+    kept = [d for d in deps_in if isinstance(d, str) and 0 < len(d) <= _MAX_DEP] \
+        if isinstance(deps_in, list) else []
+    # A list that survives filtering to empty (every entry rejected, or the list was
+    # empty to start) collapses to None — never an affirmative empty [] that would
+    # read as a declared "zero dependencies" fact.
+    deps = (kept[:_MAX_DEPS] or None) if isinstance(deps_in, list) else None
+    out = {"language": _clean_str(v.get("language")),
+           "framework": _clean_str(v.get("framework")),
+           "runtime": _clean_str(v.get("runtime")),
+           "dependencies": deps}
+    return out if any(x is not None for x in out.values()) else None
+
+
+def _clean_source_repo(v) -> dict | None:
+    if not isinstance(v, dict):
+        return None
+    host, owner, name = (_clean_str(v.get(k)) for k in ("host", "owner", "name"))
+    return {"host": host, "owner": owner, "name": name} if host and owner and name else None
+
+
+def parse_record(content: dict[str, str]) -> dict | None:
+    """The validated record, or None. Reads content[RECORD_BASENAME] — after
+    filter_content_to_subdir a subdir's record re-roots to exactly this key, so
+    one lookup serves both whole-repo and subdir subjects."""
+    body = content.get(RECORD_BASENAME)
+    if not isinstance(body, str) or len(body.encode("utf-8")) > RECORD_MAX_BYTES:
+        return None
+    try:
+        raw = json.loads(body)
+    except (ValueError, TypeError, RecursionError):
+        # RecursionError can occur on deeply-nested JSON; treat as malformed
+        return None
+    if not isinstance(raw, dict):
+        return None
+    rv = raw.get("record_version")
+    # Exclude booleans: True == 1 but is not a valid version
+    if not (isinstance(rv, int) and not isinstance(rv, bool) and rv == 1):
+        return None
+    conf = raw.get("confirmation")
+    return {
+        "record_version": 1,
+        "created_by": _clean_str(raw.get("created_by")),
+        "created_at": _clean_str(raw.get("created_at")),
+        "maintained_by": _clean_str(raw.get("maintained_by")),
+        "source_repo": _clean_source_repo(raw.get("source_repo")),
+        "technologies": _clean_technologies(raw.get("technologies")),
+        "model_access": _clean_model_access(raw.get("model_access")),
+        "confirmation": conf if isinstance(conf, dict) else None,
+    }
+
+
+def _facts(section: str, values: dict) -> dict:
+    return {k: fact(v, _SOURCE, [{"path": RECORD_BASENAME, "marker": f"{section}.{k}"}])
+            for k, v in values.items() if v is not None}
+
+
+def declared_block(content: dict[str, str]) -> dict | None:
+    """The `declared` section of the assessment payload: technologies +
+    model_access as fact envelopes (spec §4 — these two land on the ASSESSMENT;
+    created_by/maintained_by/created_at land on creation-info via repo_meta and
+    are deliberately NOT emitted here)."""
+    rec = parse_record(content)
+    if rec is None:
+        return None
+    out: dict = {}
+    if rec["technologies"]:
+        out["technologies"] = _facts("technologies", rec["technologies"])
+    if rec["model_access"]:
+        out["model_access"] = _facts("model_access", rec["model_access"])
+    return out or None

assessor/ports/source.py

@@ -10,6 +10,7 @@
 from urllib.parse import urlsplit
 
 from ..assessment import content as content_mod
+from ..assessment import record
 from ..errors import RepoGone
 from ..vendored import _valid_slug
 
@@ -158,6 +159,18 @@ def acquire(self, repo_url: str, *, prior: Prior | None) -> AcquireResult:
         # equivalent and keeps the on-disk cache path fixed per repo.
         repo_key = hashlib.sha256(f"{owner}/{name}".encode()).hexdigest()
         files, paths, capped, hits = self.fetcher.fetch(clone_url, repo_key, sha)
+        # declared_* keys land ONLY on this branch: `unchanged` fetches no
+        # files, so the record is unreadable there, and emitting None keys on
+        # that branch would look identical to "record removed" to CodeRoot's
+        # executor (which preserves prior DB values on `unchanged`) — see
+        # Task 5 brief for the full rationale.
+        rec = record.parse_record(files)
+        repo_meta = dict(repo_meta)
+        repo_meta.update({
+            "declared_created_by": rec["created_by"] if rec else None,
+            "declared_maintained_by": rec["maintained_by"] if rec else None,
+            "declared_created_at": rec["created_at"] if rec else None,
+        })
         return {"status": "acquired", "commit_sha": sha, "metadata": meta,
                 "repo_meta": repo_meta, "allowlist_version": alv,
                 "snapshot": {"commit_sha": sha, "metadata": meta,

tests/assessment/test_assessment_content.py

@@ -165,8 +165,8 @@ def test_phase1_fetches_requirements_and_langgraph_json():
     assert "requirements.txt" in out["files"] and "langgraph.json" in out["files"]
 
 
-def test_allowlist_version_is_7():
-    assert ALLOWLIST_VERSION == 7   # 6->7: agent_run_shape v3 co-occurrence changes marker-driven selection
+def test_allowlist_version_is_8():
+    assert ALLOWLIST_VERSION == 8   # 7->8: asset-record.json budget-neutral selection (authoring MCP spec §6).
 
 
 def test_select_source_paths_selects_skill_md_bodies_independent_of_is_source():
@@ -256,11 +256,11 @@ def test_non_agent_hit_does_not_displace_a_tool_reservation():
     assert chosen.index("src/tools/a.ts") < chosen.index("cfg/settings.py")
 
 
-def test_hit_paths_that_are_not_candidates_are_ignored_and_allowlist_is_7():
+def test_hit_paths_that_are_not_candidates_are_ignored_and_allowlist_is_8():
     cands = {"lib/a.ts": 10}
     chosen, _ = select_source_paths(cands, hits=[{**_SHAPE, "path": "gone/x.py"}])
     assert chosen == ["lib/a.ts"]
-    assert ALLOWLIST_VERSION == 7
+    assert ALLOWLIST_VERSION == 8
 
 
 def test_agent_lift_is_capped_so_the_tool_reservation_always_survives():

tests/test_app_http.py

@@ -113,7 +113,7 @@ def test_version_endpoint_reports_all_three_versions():
     assert set(r.json()) == {"registry_version", "allowlist_version",
                              "marker_vocab_version"}
     assert r.json()["registry_version"] == 10
-    assert r.json()["allowlist_version"] == 7
+    assert r.json()["allowlist_version"] == 8
 
 
 def test_healthz_needs_no_auth():
@@ -174,7 +174,7 @@ def get_subject(self, repo_id, subdir=""):
         return {"commit_sha": "abc123", "description": None, "homepage": None,
                 "topics": [], "license_spdx": None, "tree_paths": list(_MCP),
                 "content_paths": list(_MCP), "tree_capped": False, "marker_hits": [],
-                "source_coverage_capped": False, "allowlist_version": 7}
+                "source_coverage_capped": False, "allowlist_version": 8}
 
     def read_files(self, repo_id, commit_sha, paths):
         return {"files": dict(_MCP), "missing": []}

tests/test_handlers_assess.py

@@ -52,12 +52,13 @@ def test_returns_the_seven_top_level_record_keys():
                       "llm_used", "assessment"}
 
 
-def test_assessment_payload_has_exactly_the_fifteen_keys():
+def test_assessment_payload_has_exactly_the_sixteen_keys():
     r = assess_handler(_Source(_MCP), NullCache(), _S, _SUBJECT)
     assert set(r["assessment"]) == {
         "classification", "purpose", "compositions", "composition", "license",
         "coordinates", "versions", "risk", "known_unknowns", "topics",
-        "coverage_probes", "promoted_types", "subdir", "asset_id", "source_url"}
+        "coverage_probes", "promoted_types", "subdir", "asset_id", "source_url",
+        "declared"}
 
 
 def test_interop_is_absent():

tests/test_ports_source.py

@@ -55,7 +55,7 @@ def test_matching_prior_performs_no_clone(monkeypatch):
     f = _Fetcher()
     src = _direct(_Http(sha="abc123"), f, monkeypatch)
     r = src.acquire("https://github.com/o/n",
-                    prior={"commit_sha": "abc123", "allowlist_version": 7})
+                    prior={"commit_sha": "abc123", "allowlist_version": 8})
     assert r["status"] == "unchanged"
     assert f.calls == 0
 
@@ -75,7 +75,7 @@ def test_unchanged_still_returns_refreshed_metadata(monkeypatch):
     http = _Http(sha="abc123", repo_obj={"description": "fresh", "topics": ["mcp"]})
     src = _direct(http, _Fetcher(), monkeypatch)
     r = src.acquire("https://github.com/o/n",
-                    prior={"commit_sha": "abc123", "allowlist_version": 7})
+                    prior={"commit_sha": "abc123", "allowlist_version": 8})
     assert r["status"] == "unchanged"
     assert r["metadata"]["description"] == "fresh"
 
@@ -196,7 +196,12 @@ def test_noassertion_license_is_normalized_to_none():
 
 def test_acquire_result_carries_repo_meta_on_both_branches(monkeypatch):
     """`unchanged` (SHA reuse) must carry refreshed metadata too — spec §5.1:
-    Bucket B is refreshed on every run including the reuse path."""
+    Bucket B is refreshed on every run including the reuse path.
+
+    The acquired branch additionally carries the three declared_* keys (Task
+    5) that `unchanged` deliberately omits — see test_record_repo_meta.py for
+    the dedicated coverage of that asymmetry; this test only pins that the
+    baseline repo_meta columns keep showing up on both branches."""
     repo_obj = {
         "default_branch": "main", "description": "d", "homepage": "h",
         "topics": ["a"], "license": {"spdx_id": "MIT"},
@@ -211,14 +216,15 @@ def test_acquire_result_carries_repo_meta_on_both_branches(monkeypatch):
     src = _direct(http, _Fetcher(), monkeypatch)
     r = src.acquire("https://github.com/o/n", prior=None)
     assert r["status"] == "acquired"
-    assert set(r["repo_meta"]) == _EXPECTED_REPO_META_KEYS
+    assert set(r["repo_meta"]) == _EXPECTED_REPO_META_KEYS | {
+        "declared_created_by", "declared_maintained_by", "declared_created_at"}
     assert r["repo_meta"]["repo_owner_login"] == "o"
 
     # unchanged branch: matching prior short-circuits the clone, metadata
     # (including repo_meta) is still refreshed from resolve_head.
     src2 = _direct(http, _Fetcher(), monkeypatch)
     r2 = src2.acquire("https://github.com/o/n",
-                       prior={"commit_sha": "abc123", "allowlist_version": 7})
+                       prior={"commit_sha": "abc123", "allowlist_version": 8})
     assert r2["status"] == "unchanged"
     assert set(r2["repo_meta"]) == _EXPECTED_REPO_META_KEYS
     assert r2["repo_meta"]["repo_owner_login"] == "o"