fix(recall): stamp injected memories with age, provenance grade, and stale flag

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,35 +325,46 @@ 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 _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]]:
     """Format memories as a compact context block for 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.
+    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:**"]
     total_chars = len(lines[0])
     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 ""
-
-        line = f"- {prefix}{content}{protected}"
-
+        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)

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,14 +634,15 @@ 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))
 
     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,15 +651,16 @@ 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:
         lines.append("### Hot Memories")
         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

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)

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,51 @@ 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_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}]