From cecebe285512df59b676964ded1d6776d0f6a18f Mon Sep 17 00:00:00 2001 From: Claude Date: Mon, 24 Aug 2026 14:26:49 +0000 Subject: [PATCH 1/3] fix(recall): stamp injected memories with age, provenance grade, and stale flag MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit harness-comparison rev.2 measured the ai-architect stack (Harness B) serving facts "2-4 months stale with no age signal": every recalled memory entered the model's context as bare text, so a months-old fact and a fresh one were indistinguishable. The store already tracks the freshness (created_at, the source_attribution provenance grade, is_stale) — the injection formatter just discarded it. - shared/freshness.py: pure helper. humanize_age(created, now) → compact relative age; provenance_suffix(memory, now) → "age · src= · ⚠stale" with empty parts omitted. Caller owns the clock (deterministic/testable). - hooks/auto_recall.py: carry created_at/source_attribution/is_stale through both the PG and SQLite recall paths, and append the freshness suffix per memory in _format_injection (now injectable for tests; suffix counts toward the injection budget so the receipt still mirrors exactly what is printed). - Memories without these fields render exactly as before — bare call sites and existing budget/parity tests are unaffected. Addresses fleet-watch #110 (freshness seams) for the primary UserPromptSubmit recall path. The parallel formatters (session_start banner, recall_helpers) take the same helper as a follow-up. Tests: tests_py/shared/test_freshness.py (age buckets, ISO/naive coercion, grade/stale composition) + two _format_injection cases in tests_py/hooks/test_hook_receipts_unit.py (suffix rendered; bare memory unchanged). ruff check + format clean. Co-Authored-By: Claude Fable 5 Claude-Session: https://claude.ai/code/session_017fnomaXS71hgxMw2zr7HGR --- mcp_server/hooks/auto_recall.py | 24 ++++++- mcp_server/shared/freshness.py | 81 +++++++++++++++++++++++ tests_py/hooks/test_hook_receipts_unit.py | 25 +++++++ tests_py/shared/test_freshness.py | 65 ++++++++++++++++++ 4 files changed, 193 insertions(+), 2 deletions(-) create mode 100644 mcp_server/shared/freshness.py create mode 100644 tests_py/shared/test_freshness.py diff --git a/mcp_server/hooks/auto_recall.py b/mcp_server/hooks/auto_recall.py index 57e593da..11613c6f 100644 --- a/mcp_server/hooks/auto_recall.py +++ b/mcp_server/hooks/auto_recall.py @@ -68,6 +68,7 @@ import os import re import sys +from datetime import datetime, timezone from typing import Any from mcp_server.handlers.injection_receipts import ( @@ -76,6 +77,7 @@ receipt_marker, session_id_from_transcript, ) +from mcp_server.shared.freshness import provenance_suffix _LOG_PREFIX = "[cortex-auto-recall]" _DATABASE_URL = os.environ.get("DATABASE_URL", "postgresql://localhost:5432/cortex") @@ -173,6 +175,7 @@ def _recall_memories(conn, query: str) -> list[dict]: SELECT m.id, m.content, effective_heat(m, NOW()) AS heat, m.domain, m.agent_context, m.is_protected, + m.created_at, m.source_attribution, m.is_stale, ts_rank_cd(m.content_tsv, q) AS rank -- JOIN current_memories: auto-recall injects content into the -- session context — supersession chain heads only. The join @@ -201,6 +204,9 @@ def _recall_memories(conn, query: str) -> list[dict]: "domain": r.get("domain", ""), "agent": r.get("agent_context", ""), "protected": bool(r.get("is_protected")), + "created_at": r.get("created_at"), + "source_attribution": r.get("source_attribution", ""), + "is_stale": bool(r.get("is_stale")), } ) except Exception as exc: # noqa: BLE001 — hook boundary — failure is logged to the hook log; the hook stays non-fatal @@ -276,6 +282,9 @@ def _recall_memories_sqlite(store, query: str) -> list[dict]: "domain": m.get("domain", "") or "", "agent": m.get("agent_context", "") or "", "protected": bool(m.get("is_protected")), + "created_at": m.get("created_at"), + "source_attribution": m.get("source_attribution", ""), + "is_stale": bool(m.get("is_stale")), } ) # Protected (decision) memories first; stable sort keeps FTS rank @@ -316,7 +325,9 @@ def _process_event_sqlite(event: dict[str, Any], query: str) -> None: sys.exit(0) -def _format_injection(memories: list[dict]) -> tuple[str, list[dict]]: +def _format_injection( + memories: list[dict], now: datetime | None = None +) -> tuple[str, list[dict]]: """Format memories as a compact context block for injection. Keeps total injection under _MAX_INJECTION_CHARS to avoid flooding @@ -325,7 +336,14 @@ def _format_injection(memories: list[dict]) -> tuple[str, list[dict]]: was fetched (parity invariant, decision 4255039 correction 11): entries dropped by the budget were never in context; entries printed truncated keep their id and ARE in context. + + Each memory carries a freshness suffix (age · provenance grade · stale + marker; fleet-watch #110) so a months-old fact is distinguishable from a + fresh one in context — the failure the harness-comparison rev.2 measured. + The suffix counts toward the budget, so a memory is dropped on the full + rendered line and the receipt still mirrors exactly what is printed. """ + now = now or datetime.now(timezone.utc) lines = ["**Cortex context:**"] total_chars = len(lines[0]) included: list[dict] = [] @@ -339,8 +357,10 @@ def _format_injection(memories: list[dict]) -> tuple[str, list[dict]]: agent = m.get("agent", "") prefix = f"[{agent}] " if agent else "" protected = " (decision)" if m.get("protected") else "" + suffix = provenance_suffix(m, now) + freshness = f" · {suffix}" if suffix else "" - line = f"- {prefix}{content}{protected}" + line = f"- {prefix}{content}{protected}{freshness}" if total_chars + len(line) > _MAX_INJECTION_CHARS: break diff --git a/mcp_server/shared/freshness.py b/mcp_server/shared/freshness.py new file mode 100644 index 00000000..e635861e --- /dev/null +++ b/mcp_server/shared/freshness.py @@ -0,0 +1,81 @@ +"""Freshness annotation for injected memories (fleet-watch #110). + +The harness-comparison rev.2 A/B measured the ai-architect stack (Harness B) +serving facts "2-4 months stale with no age signal": every recalled memory +entered the model's context as bare text, so a fresh fact and a months-old one +were indistinguishable. This module renders the freshness the store *already* +tracks -- ``created_at``, the ``source_attribution`` provenance grade, and +``is_stale`` -- as a compact suffix the injection formatters append per memory. + +Pure: a memory dict plus an explicit ``now`` in, an annotation string out. The +caller owns the clock, so the output is deterministic and testable. A memory +that carries none of the three signals yields "" -- callers append nothing, so +bare-memory call sites are unaffected. +""" + +from __future__ import annotations + +from datetime import datetime, timezone + +# Calendar/SI time-unit boundaries, in seconds. These are unit *definitions* +# (a minute is 60 s, a day 86400 s), not tuned parameters; month and year use +# the conventional 30-day / 365-day display approximations. +# source: calendar arithmetic (SI second; 30-day month / 365-day year display +# convention). +_MINUTE = 60 +_HOUR = 60 * _MINUTE +_DAY = 24 * _HOUR +_MONTH = 30 * _DAY +_YEAR = 365 * _DAY + +_SEP = " · " + + +def _coerce(value: object) -> datetime | None: + """A tz-aware datetime from a datetime or ISO-8601 string, else None.""" + if isinstance(value, datetime): + return value if value.tzinfo else value.replace(tzinfo=timezone.utc) + if isinstance(value, str) and value.strip(): + try: + dt = datetime.fromisoformat(value.strip().replace("Z", "+00:00")) + except ValueError: + return None + return dt if dt.tzinfo else dt.replace(tzinfo=timezone.utc) + return None + + +def humanize_age(created: object, now: datetime) -> str: + """Compact relative age: "just now", "3d ago", "5mo ago". "" if unknown.""" + dt = _coerce(created) + if dt is None: + return "" + secs = (now - dt).total_seconds() + if secs < _MINUTE: + return "just now" + if secs < _HOUR: + return f"{int(secs // _MINUTE)}m ago" + if secs < _DAY: + return f"{int(secs // _HOUR)}h ago" + if secs < _MONTH: + return f"{int(secs // _DAY)}d ago" + if secs < _YEAR: + return f"{int(secs // _MONTH)}mo ago" + return f"{int(secs // _YEAR)}y ago" + + +def provenance_suffix(memory: dict, now: datetime) -> str: + """Age · provenance-grade · stale marker for one injected memory. + + Empty parts are omitted. A memory with no ``created_at``, an "unknown" + grade, and no stale flag yields "". + """ + parts: list[str] = [] + age = humanize_age(memory.get("created_at"), now) + if age: + parts.append(age) + grade = str(memory.get("source_attribution") or "").strip().lower() + if grade and grade != "unknown": + parts.append(f"src={grade}") + if memory.get("is_stale"): + parts.append("⚠stale") + return _SEP.join(parts) diff --git a/tests_py/hooks/test_hook_receipts_unit.py b/tests_py/hooks/test_hook_receipts_unit.py index afa38227..9b129f51 100644 --- a/tests_py/hooks/test_hook_receipts_unit.py +++ b/tests_py/hooks/test_hook_receipts_unit.py @@ -15,6 +15,8 @@ from __future__ import annotations +from datetime import datetime, timedelta, timezone + from mcp_server.hooks.auto_recall import _MAX_INJECTION_CHARS, _format_injection from mcp_server.hooks.session_start import _build_context @@ -61,6 +63,29 @@ def test_nothing_fits_returns_empty_and_no_items() -> None: assert included == [] +def test_freshness_suffix_rendered_when_fields_present() -> None: + # A memory carrying created_at / grade / stale gets an age·grade·stale + # suffix (fleet-watch #110) — the harness-comparison "no age signal" fix. + now = datetime(2026, 8, 24, 12, 0, 0, tzinfo=timezone.utc) + mem = _mem(1, content_len=20) + mem["created_at"] = now - timedelta(days=90) + mem["source_attribution"] = "verified" + mem["is_stale"] = True + text, included = _format_injection([mem], now=now) + assert "3mo ago" in text + assert "src=verified" in text + assert "⚠stale" in text + assert [m["id"] for m in included] == [1] + + +def test_bare_memory_has_no_suffix() -> None: + # Memories without freshness fields render exactly as before (no suffix), + # so existing call sites and budgets are unaffected. + now = datetime(2026, 8, 24, 12, 0, 0, tzinfo=timezone.utc) + text, _ = _format_injection([_mem(1, content_len=20)], now=now) + assert " · " not in text + + def test_banner_renders_marker_only_with_receipt() -> None: anchors = [{"id": 1, "content": "a fact", "domain": "", "is_global": False}] diff --git a/tests_py/shared/test_freshness.py b/tests_py/shared/test_freshness.py new file mode 100644 index 00000000..117deea6 --- /dev/null +++ b/tests_py/shared/test_freshness.py @@ -0,0 +1,65 @@ +"""Unit tests for shared.freshness (fleet-watch #110). + +Pure functions, fixed clock — no DB, no network. The behavior under test is +the one the harness-comparison rev.2 flagged as missing: a recalled memory must +carry its age, provenance grade, and stale flag so a months-old fact is +distinguishable from a fresh one in context. +""" + +from __future__ import annotations + +from datetime import datetime, timedelta, timezone + +from mcp_server.shared.freshness import humanize_age, provenance_suffix + +_NOW = datetime(2026, 8, 24, 12, 0, 0, tzinfo=timezone.utc) + + +def _ago(**kw) -> datetime: + return _NOW - timedelta(**kw) + + +def test_humanize_age_buckets() -> None: + assert humanize_age(_ago(seconds=30), _NOW) == "just now" + assert humanize_age(_ago(minutes=5), _NOW) == "5m ago" + assert humanize_age(_ago(hours=3), _NOW) == "3h ago" + assert humanize_age(_ago(days=5), _NOW) == "5d ago" + assert humanize_age(_ago(days=45), _NOW) == "1mo ago" + assert humanize_age(_ago(days=400), _NOW) == "1y ago" + + +def test_humanize_age_unknown_inputs() -> None: + assert humanize_age(None, _NOW) == "" + assert humanize_age("", _NOW) == "" + assert humanize_age("not-a-date", _NOW) == "" + + +def test_humanize_age_iso_string_and_naive() -> None: + # ISO string with Z, and a naive datetime, are both treated as UTC. + naive = datetime(2026, 8, 21, 12, 0, 0) # noqa: DTZ001 — coercion under test + assert humanize_age("2026-08-21T12:00:00Z", _NOW) == "3d ago" + assert humanize_age(naive, _NOW) == "3d ago" + + +def test_provenance_suffix_full() -> None: + mem = { + "created_at": _ago(days=90), + "source_attribution": "verified", + "is_stale": True, + } + suffix = provenance_suffix(mem, _NOW) + assert "3mo ago" in suffix + assert "src=verified" in suffix + assert "⚠stale" in suffix + + +def test_provenance_suffix_omits_unknown_and_absent() -> None: + # "unknown" grade and missing timestamp/stale -> nothing to show. + assert provenance_suffix({"source_attribution": "unknown"}, _NOW) == "" + assert provenance_suffix({}, _NOW) == "" + + +def test_provenance_suffix_partial() -> None: + # Only a grade present: no age, no stale marker. + mem = {"source_attribution": "verifiable"} + assert provenance_suffix(mem, _NOW) == "src=verifiable" From 487e033ead0cb76edf1e2b3f801c02f3aaf5bf21 Mon Sep 17 00:00:00 2001 From: Claude Date: Mon, 24 Aug 2026 14:30:56 +0000 Subject: [PATCH 2/3] refactor(recall): extract _render_memory_line to satisfy method-size gate MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit The freshness suffix pushed _format_injection over the craftsmanship method-size limit (CI flagged it as a NEW violation vs the base-ref baseline). Extract per-memory line rendering into _render_memory_line; both functions are now well under the limit. No behavior change — verified the freshness suffix still renders and bare memories are unchanged; scripts/check_craftsmanship.py reports OK locally. Co-Authored-By: Claude Fable 5 Claude-Session: https://claude.ai/code/session_017fnomaXS71hgxMw2zr7HGR --- mcp_server/hooks/auto_recall.py | 46 ++++++++++++++++----------------- 1 file changed, 23 insertions(+), 23 deletions(-) diff --git a/mcp_server/hooks/auto_recall.py b/mcp_server/hooks/auto_recall.py index 11613c6f..62dc6c69 100644 --- a/mcp_server/hooks/auto_recall.py +++ b/mcp_server/hooks/auto_recall.py @@ -325,6 +325,25 @@ def _process_event_sqlite(event: dict[str, Any], query: str) -> None: sys.exit(0) +def _render_memory_line(m: dict, now: datetime) -> str: + """Render one memory as a bullet with a freshness suffix. + + The suffix (age · provenance grade · stale marker; fleet-watch #110) makes + a months-old fact distinguishable from a fresh one — the failure the + harness-comparison rev.2 measured. Empty for memories lacking those fields, + so bare-memory call sites render exactly as before. + """ + content = m["content"].replace("\n", " ").strip() + if len(content) > _MAX_MEMORY_CHARS: + content = content[: _MAX_MEMORY_CHARS - 3] + "..." + agent = m.get("agent", "") + prefix = f"[{agent}] " if agent else "" + protected = " (decision)" if m.get("protected") else "" + suffix = provenance_suffix(m, now) + freshness = f" · {suffix}" if suffix else "" + return f"- {prefix}{content}{protected}{freshness}" + + def _format_injection( memories: list[dict], now: datetime | None = None ) -> tuple[str, list[dict]]: @@ -333,15 +352,9 @@ def _format_injection( Keeps total injection under _MAX_INJECTION_CHARS to avoid flooding the context window. Returns the block AND the memories that actually fit — the injection receipt must mirror what is printed, never what - was fetched (parity invariant, decision 4255039 correction 11): - entries dropped by the budget were never in context; entries printed - truncated keep their id and ARE in context. - - Each memory carries a freshness suffix (age · provenance grade · stale - marker; fleet-watch #110) so a months-old fact is distinguishable from a - fresh one in context — the failure the harness-comparison rev.2 measured. - The suffix counts toward the budget, so a memory is dropped on the full - rendered line and the receipt still mirrors exactly what is printed. + was fetched (parity invariant, decision 4255039 correction 11): a memory + is dropped on its full rendered line (freshness suffix included), so the + receipt mirrors exactly what is printed. """ now = now or datetime.now(timezone.utc) lines = ["**Cortex context:**"] @@ -349,22 +362,9 @@ def _format_injection( included: list[dict] = [] for m in memories: - content = m["content"].replace("\n", " ").strip() - # Truncate individual memories - if len(content) > _MAX_MEMORY_CHARS: - content = content[: _MAX_MEMORY_CHARS - 3] + "..." - - agent = m.get("agent", "") - prefix = f"[{agent}] " if agent else "" - protected = " (decision)" if m.get("protected") else "" - suffix = provenance_suffix(m, now) - freshness = f" · {suffix}" if suffix else "" - - line = f"- {prefix}{content}{protected}{freshness}" - + line = _render_memory_line(m, now) if total_chars + len(line) > _MAX_INJECTION_CHARS: break - lines.append(line) total_chars += len(line) included.append(m) From 688fc116a182ac4b805c4f888f19035b9d7d0b5d Mon Sep 17 00:00:00 2001 From: Claude Date: Mon, 24 Aug 2026 14:43:06 +0000 Subject: [PATCH 3/3] fix(session-start): stamp freshness on the SessionStart banner memories MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Extends the freshness stamping to the second injection surface: the SessionStart banner (_build_context) rendered anchors, team decisions, and hot memories as bare bullets, so the same "no age signal" staleness the harness-comparison rev.2 measured applied there too. - session_start.py: carry created_at / source_attribution / is_stale through the PG fetches (_fetch_anchors, _fetch_team_decisions, _fetch_hot_memories) AND the SQLite path (_partition_banner_rows), then append the freshness suffix per bullet in _build_context via a small _freshness() helper. - Reuses shared/freshness.py from the primary-recall-path change in this PR. - Memories lacking the fields render exactly as before (empty suffix). Test: tests_py/hooks/test_hook_receipts_unit.py — a stale anchor renders age + src=verified + ⚠stale. ruff + craftsmanship gate clean locally. Co-Authored-By: Claude Fable 5 Claude-Session: https://claude.ai/code/session_017fnomaXS71hgxMw2zr7HGR --- mcp_server/hooks/session_start.py | 35 +++++++++++++++++++---- tests_py/hooks/test_hook_receipts_unit.py | 22 ++++++++++++++ 2 files changed, 52 insertions(+), 5 deletions(-) diff --git a/mcp_server/hooks/session_start.py b/mcp_server/hooks/session_start.py index 1ee76806..5ef17a85 100644 --- a/mcp_server/hooks/session_start.py +++ b/mcp_server/hooks/session_start.py @@ -30,6 +30,7 @@ receipt_marker, session_id_from_transcript, ) +from mcp_server.shared.freshness import provenance_suffix from mcp_server.shared.platform import python_executable import sqlite3 import asyncio @@ -140,7 +141,8 @@ def _fetch_anchors(conn) -> list[dict]: # `memories.heat` is not a stored column; use effective_heat() # to match production recall semantics (lazy A3 decay). # Source: pg_schema.py EFFECTIVE_HEAT_FN. - "SELECT m.id, m.content, m.tags, m.domain, m.is_global " + "SELECT m.id, m.content, m.tags, m.domain, m.is_global, " + "m.created_at, m.source_attribution, m.is_stale " # JOIN current_memories (not FROM the view): effective_heat() # takes the `memories` composite type and a view row is not # coercible to it; the join keeps m table-typed while the @@ -181,6 +183,9 @@ def _fetch_anchors(conn) -> list[dict]: "content": r.get("content", ""), "domain": r.get("domain", ""), "is_global": bool(r.get("is_global", False)), + "created_at": r.get("created_at"), + "source_attribution": r.get("source_attribution", ""), + "is_stale": bool(r.get("is_stale")), } ) return anchors @@ -202,6 +207,7 @@ def _fetch_team_decisions(conn, exclude_ids: set) -> list[dict]: # matches production lazy A3 decay semantics. # Source: pg_schema.py EFFECTIVE_HEAT_FN. "SELECT m.id, m.content, m.domain, m.agent_context, " + "m.created_at, m.source_attribution, m.is_stale, " # JOIN current_memories: same pattern as _fetch_anchors — # supersession exclusion via the view, m stays table-typed # for effective_heat(). @@ -228,6 +234,9 @@ def _fetch_team_decisions(conn, exclude_ids: set) -> list[dict]: "domain": r.get("domain", ""), "agent": r.get("agent_context", ""), "heat": r.get("heat", 0.0), + "created_at": r.get("created_at"), + "source_attribution": r.get("source_attribution", ""), + "is_stale": bool(r.get("is_stale")), } ) return decisions[:3] # Keep injection compact @@ -243,7 +252,8 @@ def _fetch_hot_memories(conn, exclude_ids: set) -> list[dict]: """ try: rows = conn.execute( - "SELECT id, content, domain, heat_base AS heat, tags, is_global " + "SELECT id, content, domain, heat_base AS heat, tags, is_global, " + "created_at, source_attribution, is_stale " # current_memories: hot-pool content injected into the session # banner — supersession chain heads only. "FROM current_memories " @@ -273,6 +283,9 @@ def _fetch_hot_memories(conn, exclude_ids: set) -> list[dict]: "domain": r.get("domain", ""), "heat": r.get("heat", 0.0), "is_global": bool(r.get("is_global", False)), + "created_at": r.get("created_at"), + "source_attribution": r.get("source_attribution", ""), + "is_stale": bool(r.get("is_stale")), } ) return hot[:_HOT_LIMIT] @@ -584,6 +597,13 @@ def _emit_banner_receipt( ) +def _freshness(memory: dict, now: _dt) -> str: + """Rendered freshness suffix (age · grade · stale; fleet-watch #110), or "" + for memories that carry none of those signals.""" + suffix = provenance_suffix(memory, now) + return f" · {suffix}" if suffix else "" + + def _build_context( anchors: list[dict], hot: list[dict], @@ -614,6 +634,7 @@ def _build_context( if receipt_id is not None: header += f" {receipt_marker(receipt_id)}" lines = [header + "\n"] + now = _dt.now(_tz.utc) if checkpoint and checkpoint.get("current_task"): lines.extend(_format_checkpoint_section(checkpoint)) @@ -621,7 +642,7 @@ def _build_context( if anchors: lines.append("### Anchored Memories (critical)") for a in anchors: - lines.append(f"- {_short(a['content'])}") + lines.append(f"- {_short(a['content'])}{_freshness(a, now)}") lines.append("") # Team decisions from other agents (TMS directory layer, Wegner 1987) @@ -630,7 +651,7 @@ def _build_context( for d in team_decisions: agent = d.get("agent", "") prefix = f"[{agent}] " if agent else "" - lines.append(f"- {prefix}{_short(d['content'])}") + lines.append(f"- {prefix}{_short(d['content'])}{_freshness(d, now)}") lines.append("") if hot: @@ -638,7 +659,8 @@ def _build_context( for m in hot: heat_bar = "+" * min(5, int(m["heat"] * 5)) domain_hint = f" [{m['domain']}]" if m.get("domain") else "" - lines.append(f"- [{heat_bar}]{domain_hint} {_short(m['content'])}") + bullet = f"- [{heat_bar}]{domain_hint} {_short(m['content'])}" + lines.append(f"{bullet}{_freshness(m, now)}") lines.append("") # 2026-05-17: surface pending wiki authoring work to the in-session @@ -1084,6 +1106,9 @@ def _partition_banner_rows(rows: list[dict]) -> tuple[list[dict], list[dict]]: "domain": r.get("domain", "") or "", "heat": float(r.get("heat") or 0.0), "is_global": bool(r.get("is_global", False)), + "created_at": r.get("created_at"), + "source_attribution": r.get("source_attribution", ""), + "is_stale": bool(r.get("is_stale")), } is_anchor = bool(r.get("is_protected")) and any( t == "_anchor" or t.startswith("_anchor:") for t in tags diff --git a/tests_py/hooks/test_hook_receipts_unit.py b/tests_py/hooks/test_hook_receipts_unit.py index 9b129f51..deaf0c0b 100644 --- a/tests_py/hooks/test_hook_receipts_unit.py +++ b/tests_py/hooks/test_hook_receipts_unit.py @@ -86,6 +86,28 @@ def test_bare_memory_has_no_suffix() -> None: assert " · " not in text +def test_banner_stamps_freshness_on_stale_anchor() -> None: + # SessionStart banner: a stale anchor carries age + grade + stale marker + # (fleet-watch #110). created_at is 120 days back so the age bucket is + # stable regardless of the wall clock at test time. + old = datetime.now(timezone.utc) - timedelta(days=120) + anchors = [ + { + "id": 1, + "content": "PostgreSQL is the default store", + "domain": "", + "is_global": False, + "created_at": old, + "source_attribution": "verified", + "is_stale": True, + } + ] + text = _build_context(anchors, [], None) + assert "mo ago" in text or "y ago" in text + assert "src=verified" in text + assert "⚠stale" in text + + def test_banner_renders_marker_only_with_receipt() -> None: anchors = [{"id": 1, "content": "a fact", "domain": "", "is_global": False}]