feat(snippets): pull-time freshness — the forge confirms the cache at the moment it's trusted (#2690)
CI & Build / Plugin hooks (push) Successful in 8s
CI & Build / Python lint (push) Successful in 4s
CI & Build / TypeScript typecheck (push) Successful in 37s
CI & Build / integration (push) Successful in 26s
CI & Build / Python tests (push) Failing after 43s
CI & Build / Build & push image (push) Skipped

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>
This commit is contained in:
2026-08-16 12:46:45 -04:00
co-authored by Claude Fable 5
parent 13e428c596
commit 2fce57847b
5 changed files with 408 additions and 6 deletions
+9
View File
@@ -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
+2
View File
@@ -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
+51
View File
@@ -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)
+146 -6
View File
@@ -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.
+200
View File
@@ -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