feat(snippets): pull-time freshness — the forge confirms the cache at the moment it's trusted (#2690)

First consumer of the forge adapter. attach_live_body decorates both
pull surfaces (MCP get_snippet, REST detail) with body_source +
body_freshness when the instance has a forge: 'current' means the
cached code was just found verbatim (whitespace-normalized, the same
normalization the verdict hash uses) in the fetched file, and
provenance restamps to the file's last commit — reflected in the
response and persisted in the background. A snippet body is a FRAGMENT
of its file, so a fetch can honestly CONFIRM the cache or report
divergence, never clobber the record with the whole file: 'diverged'
is the reader's information, and a 404 stamps the mechanically-true
'missing' verdict into the existing attention state — once, not on
every pull of an already-flagged record.

The probe never raises and never blocks past 2.5s (tighter than the
adapter's own timeout — the pull is where a session decides whether
pulling is worth it, #2663's finding); a hung forge costs bounded time
and the cache serves. A no-forge instance's response stays
byte-identical to today's (rule #115 baseline, pinned by test).

services/background.py is the new one home for fire-and-forget tasks
with strong references (the #2663 GC footgun) — telemetry's two copies
predate it and keep their bespoke canaries; new callers use this.

Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>

src/scribe/mcp/tools/snippets.py

@@ -201,6 +201,12 @@ async def get_snippet(snippet_id: int) -> dict:
     """Fetch a snippet by id — the full record: code, signature, location, and a
     parsed `snippet` field of its structured parts.
 
+    On an instance with a forge configured, the response also carries
+    `body_source` + `body_freshness`: "current" means the code was just
+    confirmed against the recorded location; "diverged" or "missing" means
+    the source moved on — trust the location over the cached body and
+    consider verify_snippet after you look.
+
     If the record belongs to someone else it carries `shared: true` with the
     `owner` and your `permission`. Read that as ONE PERSON'S SUGGESTION, not as
     established practice here: judge it on its merits, say whose it is when you
@@ -211,6 +217,9 @@ async def get_snippet(snippet_id: int) -> dict:
     if note is None:
         raise ValueError(f"snippet {snippet_id} not found")
     data = snippets_svc.snippet_to_dict(note)
+    # Forge-checked freshness (#2690): attaches body_source/body_freshness
+    # when the instance has a forge; a no-forge instance sees no new fields.
+    await snippets_svc.attach_live_body(note, data)
     data.update(await access_svc.describe_provenance(uid, note))
     # A "pull" is an explicit open, so it's recorded HERE rather than in
     # snippets_svc.get_snippet — the service is also reached by update/merge

src/scribe/routes/snippets.py

@@ -157,6 +157,8 @@ async def get_snippet_route(snippet_id: int):
         return not_found("Snippet")
     note, permission = loaded
     data = snippets_svc.snippet_to_dict(note)
+    # Forge-checked freshness (#2690) — same decoration the MCP pull gets.
+    await snippets_svc.attach_live_body(note, data)
     data["permission"] = permission
     # Read the association as the OWNER: a shared reader isn't scoped to the
     # owner's project, so their own id would come back empty (mirrors the

src/scribe/services/background.py

@@ -0,0 +1,51 @@
+"""Fire-and-forget background tasks that actually run.
+
+The event loop holds only a WEAK reference to a task, so a bare
+``create_task`` with no other holder can be garbage-collected mid-flight — a
+write that never errors and never lands (the #2663 GC footgun). This module is
+the one place that gets the pattern right: strong references in ``_pending``,
+discarded on completion, with failures logged at WARNING instead of vanishing.
+
+``note_usage`` and ``retrieval_telemetry`` predate this module and carry their
+own copies with bespoke canary semantics; new fire-and-forget callers use this
+instead of writing a fourth copy.
+"""
+from __future__ import annotations
+
+import asyncio
+import logging
+from collections.abc import Coroutine
+
+logger = logging.getLogger(__name__)
+
+_pending: set[asyncio.Task] = set()
+
+
+def spawn(coro: Coroutine, *, site: str) -> None:
+    """Schedule ``coro`` fire-and-forget; ``site`` names it in failure logs.
+
+    No running loop (sync context outside the app) closes the coroutine and
+    skips — every app path runs on the loop, and blocking would be worse.
+    """
+    try:
+        task = asyncio.get_running_loop().create_task(coro)
+    except RuntimeError:
+        coro.close()
+        logger.debug("background task %s skipped — no running event loop", site)
+        return
+    _pending.add(task)
+
+    def _done(t: asyncio.Task) -> None:
+        _pending.discard(t)
+        if not t.cancelled() and t.exception() is not None:
+            logger.warning(
+                "background task %s failed", site, exc_info=t.exception()
+            )
+
+    task.add_done_callback(_done)
+
+
+async def drain() -> None:
+    """Await everything in flight — for tests that need the writes landed."""
+    while _pending:
+        await asyncio.gather(*list(_pending), return_exceptions=True)

src/scribe/services/snippets.py

@@ -28,6 +28,7 @@ came from.
 """
 from __future__ import annotations
 
+import asyncio
 import hashlib
 import logging
 import re
@@ -379,16 +380,20 @@ VERIFY_STATUSES = (VERIFY_OK, VERIFY_MISSING, VERIFY_MOVED, VERIFY_CHANGED)
 VERIFY_DRIFTED = (VERIFY_MISSING, VERIFY_MOVED, VERIFY_CHANGED)
 
 
+def _normalized_code(code: str) -> str:
+    """Whitespace normalization shared by the verdict hash and the pull-time
+    containment check, so 'unchanged' means the same thing in both places:
+    trailing whitespace per line and leading/trailing blank lines dropped."""
+    return "\n".join(line.rstrip() for line in (code or "").splitlines()).strip()
+
+
 def code_sha(code: str) -> str:
     """Stable fingerprint of a snippet's code, for expiring stale verdicts.
 
-    Trailing whitespace per line and leading/trailing blank lines are stripped
-    before hashing: those change when a file is reformatted without the code
-    meaning anything different, and a verdict shouldn't expire over an editor's
-    trailing-newline habit.
+    Normalized first (see _normalized_code): a reformat that changes nothing
+    shouldn't expire a verdict over an editor's trailing-newline habit.
     """
-    normalized = "\n".join(line.rstrip() for line in (code or "").splitlines()).strip()
-    return hashlib.sha256(normalized.encode("utf-8")).hexdigest()[:32]
+    return hashlib.sha256(_normalized_code(code).encode("utf-8")).hexdigest()[:32]
 
 
 def compose_verification(
@@ -925,6 +930,141 @@ async def record_verification(
     return await notes_svc.update_note(note.user_id, snippet_id, data=data)
 
 
+# --- pull-time freshness (#2690) ---------------------------------------------
+# A pull is the moment freshness matters: the reader is about to trust the
+# cached body. When the instance has a forge configured, the pull fetches the
+# recorded file and answers the one mechanically-answerable question — does
+# the cached code still appear in the source, verbatim after whitespace
+# normalization? The body is a FRAGMENT of the file, so "serve the fetched
+# file" would clobber the record; confirmation + provenance refresh is what
+# fetching can honestly deliver, and divergence is reported, not overwritten.
+#
+# With no forge configured this function attaches NOTHING — the response is
+# byte-identical to pre-forge behavior (rule #115's baseline).
+
+# Total budget for the in-pull fetch. Tighter than the adapter's own timeout:
+# the pull is the moment a session decides whether pulling is worth it
+# (#2663's pull-through finding), so a slow forge must cost bounded time and
+# then the cache serves.
+PULL_FETCH_BUDGET_S = 2.5
+
+
+async def _stamp_missing(note, host: str) -> None:
+    """Record the mechanically-established 'missing' verdict from a pull-time
+    404 — the recorded path is gone at the forge's head. Runs in the
+    background; written as the owner, like every metadata write here."""
+    await record_verification(
+        note.user_id, note.id, status=VERIFY_MISSING,
+        detail=f"pull-time forge fetch: recorded path not found on {host}",
+    )
+
+
+async def _refresh_provenance(note, commit_sha: str) -> None:
+    """Restamp data.provenance after a pull confirmed the cache matches the
+    source at ``commit_sha``. Background write, rebuilt like record_verification
+    so nothing else about the record changes."""
+    fields = snippet_fields(note)
+    data = compose_data(
+        name=fields.get("name", ""),
+        when_to_use=fields.get("when_to_use", ""),
+        signature=fields.get("signature", ""),
+        language=fields.get("language", ""),
+        code=fields.get("code", ""),
+        locations=fields.get("locations") or [],
+        merged_from=fields.get("merged_from") or [],
+        verification=fields.get("verification"),
+        provenance=compose_provenance(commit_sha=commit_sha),
+    )
+    await notes_svc.update_note(note.user_id, note.id, data=data)
+
+
+async def attach_live_body(note, data: dict) -> None:
+    """Decorate a PULL response with forge-checked freshness (#2690).
+
+    Adds, when (and only when) a forge is configured:
+      - ``body_source``: "forge" (confirmed against the source just now) or
+        "cache" (the stored body, for whatever reason follows)
+      - ``body_freshness``: "current" | "diverged" | "missing" |
+        "unreachable" | "no-recorded-location" | "repo-not-on-this-forge"
+
+    Never raises, never blocks past PULL_FETCH_BUDGET_S, never rewrites the
+    body: a freshness probe must not be able to break or slow the pull it
+    decorates, and divergence is the READER's information, not license to
+    clobber a record mid-read. A confirmed-current pull refreshes provenance
+    in the background; a 404 stamps the 'missing' verdict into the same
+    attention state verify_snippet uses.
+    """
+    from scribe.services.background import spawn
+    from scribe.services.forge import ForgeError, ForgeNotFound, get_forge
+
+    try:
+        forge = await get_forge()
+    except Exception:
+        logger.warning("forge lookup failed during pull", exc_info=True)
+        return
+    if forge is None:
+        return
+
+    fields = data.get("snippet") if isinstance(data.get("snippet"), dict) else None
+    if fields is None:
+        fields = snippet_fields(note)
+    loc = next(
+        (
+            entry
+            for entry in (fields.get("locations") or [])
+            if entry.get("repo") and entry.get("path")
+        ),
+        None,
+    )
+    if loc is None:
+        data["body_source"] = "cache"
+        data["body_freshness"] = "no-recorded-location"
+        return
+    repo = forge.resolve_repo(loc["repo"])
+    if repo is None:
+        data["body_source"] = "cache"
+        data["body_freshness"] = "repo-not-on-this-forge"
+        return
+
+    try:
+        fetched = await asyncio.wait_for(
+            forge.read_file(repo, loc["path"]), timeout=PULL_FETCH_BUDGET_S
+        )
+    except ForgeNotFound:
+        data["body_source"] = "cache"
+        data["body_freshness"] = "missing"
+        stored = fields.get("verification") or {}
+        # Don't re-stamp what's already stamped — a popular-but-broken record
+        # would otherwise be rewritten on every pull.
+        if stored.get("status") != VERIFY_MISSING:
+            spawn(_stamp_missing(note, forge.host), site="pull missing-verdict")
+        return
+    except (ForgeError, asyncio.TimeoutError):
+        data["body_source"] = "cache"
+        data["body_freshness"] = "unreachable"
+        return
+
+    cached = _normalized_code(fields.get("code") or "")
+    if cached and cached in _normalized_code(fetched.content):
+        data["body_source"] = "forge"
+        data["body_freshness"] = "current"
+        if fetched.commit_sha:
+            prov = compose_provenance(commit_sha=fetched.commit_sha)
+            # Reflected in THIS response as well as persisted — the reader
+            # shouldn't need a second pull to see the stamp they caused.
+            if isinstance(data.get("snippet"), dict):
+                data["snippet"]["provenance"] = prov
+            stored_prov = fields.get("provenance") or {}
+            if stored_prov.get("commit_sha") != fetched.commit_sha:
+                spawn(
+                    _refresh_provenance(note, fetched.commit_sha),
+                    site="pull provenance-refresh",
+                )
+    else:
+        data["body_source"] = "cache"
+        data["body_freshness"] = "diverged"
+
+
 async def delete_snippet(user_id: int, snippet_id: int) -> bool:
     """Retire a snippet to the trash (recoverable). Returns False if the id isn't
     a snippet this user may WRITE.

tests/test_snippet_live_body.py

@@ -0,0 +1,200 @@
+"""Pull-time freshness (#2690) — attach_live_body against a mocked forge.
+
+The properties that must hold, each with its own way of rotting:
+
+  - A no-forge instance's pull response is BYTE-IDENTICAL to today's (rule
+    #115 — the baseline, not a degraded mode).
+  - The probe never rewrites the body: "current" refreshes provenance,
+    "diverged" reports, and neither clobbers the record mid-read.
+  - A 404 stamps the mechanically-true 'missing' verdict — once, not on
+    every pull of an already-flagged record.
+  - The pull is never slower than the budget: a hung forge costs bounded
+    time and then the cache serves.
+"""
+import asyncio
+import base64
+import time
+from types import SimpleNamespace
+from unittest.mock import AsyncMock, patch
+
+import httpx
+
+from scribe.services import background
+from scribe.services import snippets as svc
+from scribe.services.forge import GiteaForge
+
+BASE = "https://git.example.com"
+CODE = "def helper(x):\n    return x + 1\n"
+SHA = "f" * 40
+
+
+def _note():
+    return SimpleNamespace(id=7, user_id=3, title="", body="", tags=[], data=None)
+
+
+def _data(*, repo=f"{BASE}/alice/widget", code=CODE, verification=None, provenance=None):
+    snippet = {
+        "code": code,
+        "locations": [{"repo": repo, "path": "src/helper.py", "symbol": "helper"}],
+    }
+    if verification:
+        snippet["verification"] = verification
+    if provenance:
+        snippet["provenance"] = provenance
+    return {"snippet": snippet}
+
+
+def _forge_with(handler) -> GiteaForge:
+    return GiteaForge(BASE, "tok", transport=httpx.MockTransport(handler))
+
+
+def _file_response(content: str, commit_sha: str = SHA) -> httpx.Response:
+    return httpx.Response(200, json={
+        "type": "file", "encoding": "base64",
+        "content": base64.b64encode(content.encode()).decode(),
+        "last_commit_sha": commit_sha, "path": "src/helper.py",
+    })
+
+
+def _patched(forge):
+    return patch("scribe.services.forge.get_forge", AsyncMock(return_value=forge))
+
+
+async def test_no_forge_attaches_nothing():
+    data = _data()
+    before = repr(data)
+    with _patched(None):
+        await svc.attach_live_body(_note(), data)
+    assert repr(data) == before
+    assert "body_source" not in data
+
+
+async def test_current_code_confirms_and_refreshes_provenance():
+    # The file wraps the cached code with extra context and trailing spaces —
+    # containment is judged after the same normalization the verdict hash uses.
+    file_content = "import os\n\n" + CODE.replace(" + 1", " + 1  ").rstrip() + "\n\n# eof\n"
+    forge = _forge_with(lambda r: _file_response(file_content))
+    saved = {}
+
+    async def fake_update(uid, nid, **fields):
+        saved.update(fields)
+
+    data = _data()
+    with _patched(forge), patch.object(svc.notes_svc, "update_note", fake_update):
+        await svc.attach_live_body(_note(), data)
+        await background.drain()
+
+    assert data["body_source"] == "forge"
+    assert data["body_freshness"] == "current"
+    # Reflected in the response...
+    assert data["snippet"]["provenance"]["commit_sha"] == SHA
+    # ...and persisted, without touching the body.
+    assert saved["data"]["provenance"]["commit_sha"] == SHA
+    assert "body" not in saved
+
+
+async def test_current_with_same_stored_sha_skips_the_write():
+    forge = _forge_with(lambda r: _file_response("prefix\n" + CODE))
+    update = AsyncMock()
+    data = _data(provenance={"commit_sha": SHA, "fetched_at": "t"})
+    with _patched(forge), patch.object(svc.notes_svc, "update_note", update):
+        await svc.attach_live_body(_note(), data)
+        await background.drain()
+    assert data["body_freshness"] == "current"
+    update.assert_not_called()
+
+
+async def test_diverged_reports_without_clobbering():
+    forge = _forge_with(lambda r: _file_response("def helper(x):\n    return x - 1\n"))
+    update = AsyncMock()
+    data = _data()
+    with _patched(forge), patch.object(svc.notes_svc, "update_note", update):
+        await svc.attach_live_body(_note(), data)
+        await background.drain()
+    assert data["body_source"] == "cache"
+    assert data["body_freshness"] == "diverged"
+    assert data["snippet"]["code"] == CODE
+    update.assert_not_called()
+
+
+async def test_missing_stamps_the_verdict_once():
+    forge = _forge_with(lambda r: httpx.Response(404, json={}))
+    data = _data()
+    with _patched(forge), patch.object(
+        svc, "record_verification", AsyncMock()
+    ) as verdict:
+        await svc.attach_live_body(_note(), data)
+        await background.drain()
+    assert data["body_freshness"] == "missing"
+    verdict.assert_awaited_once()
+    assert verdict.await_args.kwargs["status"] == svc.VERIFY_MISSING
+
+    # Already stamped missing → no re-stamp on the next pull.
+    data2 = _data(verification={"status": svc.VERIFY_MISSING, "code_sha": "x"})
+    with _patched(forge), patch.object(
+        svc, "record_verification", AsyncMock()
+    ) as verdict2:
+        await svc.attach_live_body(_note(), data2)
+        await background.drain()
+    assert data2["body_freshness"] == "missing"
+    verdict2.assert_not_awaited()
+
+
+async def test_unreachable_falls_back_to_cache():
+    def handler(request):
+        raise httpx.ConnectError("down", request=request)
+
+    data = _data()
+    with _patched(_forge_with(handler)):
+        await svc.attach_live_body(_note(), data)
+    assert data["body_source"] == "cache"
+    assert data["body_freshness"] == "unreachable"
+
+
+async def test_hung_forge_costs_bounded_time(monkeypatch):
+    async def slow_handler(request):
+        await asyncio.sleep(30)
+        return _file_response(CODE)
+
+    monkeypatch.setattr(svc, "PULL_FETCH_BUDGET_S", 0.2)
+    data = _data()
+    start = time.monotonic()
+    with _patched(_forge_with(slow_handler)):
+        await svc.attach_live_body(_note(), data)
+    assert time.monotonic() - start < 2.0
+    assert data["body_freshness"] == "unreachable"
+
+
+async def test_foreign_repo_and_placeless_records_read_as_cache():
+    forge = GiteaForge(BASE, "tok")
+    data = _data(repo="https://github.com/alice/widget")
+    with _patched(forge):
+        await svc.attach_live_body(_note(), data)
+    assert data["body_freshness"] == "repo-not-on-this-forge"
+
+    placeless = {"snippet": {"code": CODE, "locations": []}}
+    with _patched(forge):
+        await svc.attach_live_body(_note(), placeless)
+    assert placeless["body_freshness"] == "no-recorded-location"
+
+
+def test_both_pull_surfaces_attach_freshness():
+    """Source-inspection guard (the CI convention for wiring assertions): the
+    MCP pull and the REST detail route both decorate — a surface that forgets
+    is a surface whose readers silently lose freshness."""
+    import pathlib
+
+    root = pathlib.Path(__file__).resolve().parents[1] / "src" / "scribe"
+    mcp_src = (root / "mcp" / "tools" / "snippets.py").read_text()
+    rest_src = (root / "routes" / "snippets.py").read_text()
+    assert "attach_live_body" in mcp_src
+    assert "attach_live_body" in rest_src
+
+
+async def test_forge_failure_inside_lookup_never_breaks_the_pull():
+    with patch(
+        "scribe.services.forge.get_forge", AsyncMock(side_effect=RuntimeError("cfg"))
+    ):
+        data = _data()
+        await svc.attach_live_body(_note(), data)
+    assert "body_source" not in data