feat(context-engine): D3 P0-A — surface cold-item slug + record injected slugs (attribution substrate)

.console/log.md

@@ -1,4 +1,25 @@
 # Log
+## 2026-07-16 — feat: D3 P0-A — surface cold-item slug + record injected slugs (attribution substrate)
+
+The substrate for D3 "attribution scheme A" (explicit `Context-Used: <slug>`
+citation). Two focused, non-breaking changes to the context engine:
+
+- cold.py `surface_cold`: each surfaced cold line now leads with a
+  machine-parseable `[<slug>]` citation token, ahead of the unchanged human
+  `topic — glob — finding` content. An acting agent reading the injected line
+  can now cite the exact item it used. Additive only — what surfaces and when
+  is untouched.
+- route.py telemetry: `_log_injection_event` records a new `cold_slugs`
+  field — `[{"slug", "ts"}]` per surfaced cold item, ts = injection time — in
+  ADDITION to the legacy `cold_surfaced` count and every existing field (fully
+  backward compatible). Slugs are recovered from the surfaced lines via a new
+  `_cold_slug_from_line` helper, which skips the `...(N more)` truncation note.
+
+NO promotion-gate behavior change. New/updated tests: two exact-line assertions
+updated to include the token, a parse-back test in test_cold_store.py, and two
+telemetry tests in test_context_router.py (cold_slugs recorded + legacy fields
+unchanged; truncation note excluded). Full suite 412 pass; ruff clean.
+
 ## 2026-07-14 — refactor: C2 shares one public section-extractor with signing (reviewer code_quality)
 
 Addressed the C2 reviewer's code_quality concern + removed a real smell:

src/context_lifecycle/context_engine/cold.py

@@ -285,12 +285,17 @@ def surface_cold(target: str, knowledge_dir: Path, max_items: int) -> list[str]:
     """One-line cold-topic surfacing for `target` (spec §2.3).
 
     For each cold-tier item whose any() path-glob matches `target`, emit one
-    line ``'topic — <first matching path glob> — <finding>'`` (em-dash
-    separators, spec §2.3 line 198-199 verbatim format). Only
-    ``tier == 'cold'`` items surface (warm/hot are injected/anchored elsewhere,
-    avoiding double-surfacing). Sorted by topic for determinism; capped at
-    `max_items` with a non-silent ``...(N more)`` note (spec §3 parity).
-    Returns [] on any failure — fail-soft, never raises (spec §1).
+    line ``'[<slug>] topic — <first matching path glob> — <finding>'`` (em-dash
+    separators, spec §2.3 line 198-199 verbatim format for the human portion).
+    The leading ``[<slug>]`` token is a machine-parseable citation handle (D3
+    attribution scheme A): an acting agent that reads the injected line can cite
+    the item back in a ``Context-Used: <slug>`` commit trailer. The token is
+    additive — the human ``topic — glob — finding`` content after it is
+    unchanged. Only ``tier == 'cold'`` items surface (warm/hot are
+    injected/anchored elsewhere, avoiding double-surfacing). Sorted by topic for
+    determinism; capped at `max_items` with a non-silent ``...(N more)`` note
+    (spec §3 parity). Returns [] on any failure — fail-soft, never raises
+    (spec §1).
     """
     try:
         if target.startswith("./"):
@@ -307,7 +312,10 @@ def surface_cold(target: str, knowledge_dir: Path, max_items: int) -> list[str]:
             if hit_glob is None:
                 continue
             matched.append(
-                (item.topic, f"{item.topic} — {hit_glob} — {item.finding}")
+                (
+                    item.topic,
+                    f"[{item.slug}] {item.topic} — {hit_glob} — {item.finding}",
+                )
             )
 
         matched.sort(key=lambda pair: pair[0])

src/context_lifecycle/context_engine/route.py

@@ -289,6 +289,23 @@ def _surface_cold_lines(target: str, root: Path) -> list[str]:
 
 
 
+def _cold_slug_from_line(line: str) -> str | None:
+    """Extract the leading ``[<slug>]`` citation token from a surfaced cold line.
+
+    The cold emitter (`cold.surface_cold`) prefixes every real item line with a
+    machine-parseable ``[<slug>]`` handle (D3 attribution scheme A). Return the
+    slug, or None for a line that carries no token — e.g. the
+    ``...(N more cold topic(s); ...)`` truncation note, which must not be
+    recorded as an injected slug. Never raises.
+    """
+    if not line.startswith("["):
+        return None
+    end = line.find("]")
+    if end <= 1:
+        return None
+    return line[1:end]
+
+
 def _log_injection_event(
     root: Path,
     target: str,
@@ -298,6 +315,7 @@ def _log_injection_event(
     missing: list[str],
     over_budget: list[str],
     cold_count: int,
+    cold_slugs: list[str],
 ) -> None:
     """Append one JSONL line per injection to sessions/.telemetry/injection.jsonl.
 
@@ -307,18 +325,27 @@ def _log_injection_event(
     dot-dir, machine-local, covered by the fleet's sessions gitignore) and is
     strictly best-effort: a telemetry failure must never affect injection
     (spec §1: the router never raises).
+
+    `cold_slugs` additionally records the slug of each surfaced cold item with a
+    per-slug injection timestamp (D3 attribution scheme A substrate): the ledger
+    of which cold memories were put in front of an agent, and when, so a later
+    ``Context-Used: <slug>`` citation can be matched to a real injection. This is
+    purely additive — the legacy ``cold_surfaced`` count and every existing field
+    are unchanged, so existing telemetry consumers keep working.
     """
     try:
         tel_dir = root / ".context" / "sessions" / ".telemetry"
         tel_dir.mkdir(parents=True, exist_ok=True)
+        ts = datetime.now().isoformat(timespec="seconds")
         event = {
-            "ts": datetime.now().isoformat(timespec="seconds"),
+            "ts": ts,
             "target": target,
             "injected": injected,
             "empty": empty,
             "missing": missing,
             "over_budget": over_budget,
             "cold_surfaced": cold_count,
+            "cold_slugs": [{"slug": slug, "ts": ts} for slug in cold_slugs],
         }
         with (tel_dir / "injection.jsonl").open("a", encoding="utf-8") as fh:
             fh.write(json.dumps(event, ensure_ascii=False) + "\n")
@@ -380,6 +407,13 @@ def build_context(target: str, root: Path) -> str:
     if not blocks and not notes and not cold_lines:
         return ""
 
+    # Recover the citation slug from each surfaced cold line (skipping the
+    # ...(N more) truncation note, which carries no token) so the telemetry can
+    # record exactly which cold items were injected — the D3 attribution substrate.
+    cold_slugs = [
+        slug for line in cold_lines if (slug := _cold_slug_from_line(line)) is not None
+    ]
+
     # Something is being surfaced — record it (best-effort, never raises).
     _log_injection_event(
         root,
@@ -389,6 +423,7 @@ def build_context(target: str, root: Path) -> str:
         missing=missing,
         over_budget=list(over_budget),
         cold_count=len(cold_lines),
+        cold_slugs=cold_slugs,
     )
 
     note = ("\n\n" + "\n".join(notes)) if notes else ""

tests/test_cold_store.py

@@ -179,7 +179,32 @@ def test_surface_cold_matches_glob(tmp_path):
         "src/platform_manifest/projection/rules.py", kn, max_items=5
     )
     assert len(lines) == 1
+    # The line now leads with the machine-parseable [slug] citation token
+    # (D3 attribution scheme A), followed by the unchanged human content.
     assert lines[0] == (
+        "[projection] projection — src/platform_manifest/projection/** — "
+        "Redaction must run before validation."
+    )
+
+
+def test_surface_cold_line_carries_parseable_slug(tmp_path):
+    # D3 attribution scheme A substrate: the surfaced line must carry the item's
+    # slug as a leading [slug] token that parses back exactly to the slug, while
+    # the human `topic — glob — finding` content is preserved intact.
+    kn = tmp_path / "knowledge"
+    _write(kn, "projection.md", _VALID)
+    lines = cold.surface_cold(
+        "src/platform_manifest/projection/rules.py", kn, max_items=5
+    )
+    assert len(lines) == 1
+    line = lines[0]
+    # The token is greppable and parses back to the slug (== the file stem).
+    assert line.startswith("[projection] ")
+    parsed_slug = line[1 : line.index("]")]
+    assert parsed_slug == "projection"
+    # Human-readable content unchanged (present after the token).
+    human = line[line.index("]") + 2 :]
+    assert human == (
         "projection — src/platform_manifest/projection/** — "
         "Redaction must run before validation."
     )
@@ -198,9 +223,10 @@ def test_surface_cold_emits_second_matching_glob(tmp_path):
         "src/platform_manifest/projection/rules.py", kn, max_items=5
     )
     assert len(lines) == 1
-    # The emitted glob is the second path entry (the matching one), with em-dash.
+    # The emitted glob is the second path entry (the matching one), with em-dash,
+    # behind the leading [slug] citation token.
     assert lines[0] == (
-        "projection — src/platform_manifest/projection/** — "
+        "[projection] projection — src/platform_manifest/projection/** — "
         "Redaction must run before validation."
     )
 

tests/test_context_router.py

@@ -322,6 +322,87 @@ def test_build_context_no_event_on_no_match(tmp_path):
     assert not (tmp_path / ".context" / "sessions" / ".telemetry").exists()
 
 
+# A valid §2.6 cold item (mirrors the cold-store test fixture) so the router's
+# lazy cold surfacing has something to inject.
+_COLD_ITEM = """---
+topic: projection
+paths: ["src/x/**"]
+created: 2026-06-02
+campaign_id: c-2026-06-02-9f3a
+consequence:
+  acted_on_commit: null
+  tests_green: unknown
+tier: cold
+pinned: false
+last_injected: null
+---
+## Finding
+Redaction must run before validation.
+## Detail
+Long reasoning here.
+"""
+
+
+def test_build_context_records_injected_cold_slugs(tmp_path):
+    # D3 attribution scheme A substrate: the injection telemetry must record the
+    # slug of each surfaced cold item with a per-slug injection ts, IN ADDITION
+    # to the legacy cold_surfaced count and every existing field (non-breaking).
+    import json as _json
+
+    ctx = tmp_path / ".context"
+    kn = ctx / "knowledge"
+    kn.mkdir(parents=True)
+    (kn / "projection.md").write_text(_COLD_ITEM)
+
+    block = route.build_context("src/x/thing.py", tmp_path)
+    # The slug token is visible in the injected block (agent can cite it).
+    assert "[projection]" in block
+
+    log = ctx / "sessions" / ".telemetry" / "injection.jsonl"
+    assert log.is_file()
+    event = _json.loads(log.read_text().splitlines()[-1])
+
+    # New field: one record per surfaced cold slug, each with slug + ts.
+    assert event["cold_slugs"] == [{"slug": "projection", "ts": event["ts"]}]
+
+    # Legacy fields unchanged: the count still reflects the surfaced cold lines,
+    # and the existing keys are all still present with their prior meaning.
+    assert event["cold_surfaced"] == 1
+    assert event["target"] == "src/x/thing.py"
+    assert event["injected"] == []  # cold-only target: no warm docs
+    assert event["empty"] == []
+    assert event["missing"] == []
+    assert event["over_budget"] == []
+    assert "ts" in event
+
+
+def test_injection_event_omits_truncation_note_from_cold_slugs(tmp_path):
+    # The ...(N more) truncation note is not a cold item and must NOT appear as a
+    # slug in cold_slugs; the count (cold_surfaced) still includes the note line.
+    import json as _json
+
+    ctx = tmp_path / ".context"
+    ctx.mkdir()
+    (ctx / "routes.yaml").write_text(
+        "engine_compat: '>=0.2 <0.3'\nbudget:\n  max_cold_surface_per_edit: 1\n"
+    )
+    kn = ctx / "knowledge"
+    kn.mkdir()
+    for i in range(3):
+        (kn / f"item{i}.md").write_text(
+            _COLD_ITEM.replace("topic: projection", f"topic: t{i}")
+        )
+
+    route.build_context("src/x/thing.py", tmp_path)
+
+    log = ctx / "sessions" / ".telemetry" / "injection.jsonl"
+    event = _json.loads(log.read_text().splitlines()[-1])
+    slugs = [rec["slug"] for rec in event["cold_slugs"]]
+    # Exactly the one surfaced item (cap=1); no synthetic slug for the note.
+    assert slugs == ["item0"]
+    assert all("more cold topic" not in s for s in slugs)
+
+
 def test_telemetry_failure_never_breaks_router(tmp_path, monkeypatch):
     ctx = tmp_path / ".context"
     ctx.mkdir()