feat: expose lensing catch-up freshness in cockpit

README.md

@@ -82,7 +82,21 @@ Away for a while? [`/catch_up`](skills/catch_up/catch_up.md) collects every
 paper missed since memory last ingested one — lapsed suggestions from git
 history, the open queue, and an arXiv gap-fill over days the digest never ran
 (`scripts/catch_up.py`) — for one triage and one filing PR. The dashboard
-shows a banner once a week has passed without an ingest.
+shows a strong-lensing catch-up banner after seven days without recorded lensing
+paper activity. Recent activity in other topics does not reset that clock.
+
+The same `lensing_catch_up` record backs the banner and cockpit feed: canonical
+state, cutoff date, age, seven-day threshold/deadline, observation time, evidence
+links and explicit action descriptors. Healthy work adds no attention row;
+missing, invalid or future dates are unknown. `last_ingestion` identifies the
+verified bib-plus-sources commit, while `last_completed` is a queue DONE date
+(which may mean read-without-filing); `last_activity` is the existing catch-up
+cutoff, the later of those dates. Legacy all-scope fields remain available.
+
+The cockpit copies the existing `/catch_up lensing` prompt and links to its
+procedure. Its safety is `scientific_judgement`: candidate selection and filing
+retain the procedure's human review. Observing staleness executes nothing, and
+no scientific choice is fabricated before candidates are harvested.
 
 The wiki schema is defined in
 [`wiki/AGENTS.md`](wiki/AGENTS.md) and inherited by every

scripts/board.py

@@ -246,11 +246,16 @@ def collect(root: Path | None = None) -> dict:
         # When memory last took a paper in (any scope): git + files, never the
         # network. None when nothing is on record or git is unavailable.
         "last_ingested": None,
+        "lensing_ingest_evidence": None,
     }
     try:
         snapshot["last_ingested"] = catch_up.last_ingested("all", root)
     except Exception:  # noqa: BLE001 — the banner is advisory, never fatal
         snapshot["last_ingested"] = None
+    try:
+        snapshot["lensing_ingest_evidence"] = catch_up.ingest_evidence("lensing", root)
+    except Exception:  # advisory: failed observation is unknown, never fresh
+        snapshot["lensing_ingest_evidence"] = None
 
     stems: set[str] = set()
     all_slugs: list[str] = []
@@ -627,34 +632,96 @@ def _days_since_ingest(snapshot: dict) -> int | None:
     return max(0, (now - last).days)
 
 
+def _lensing_catch_up(snapshot: dict) -> dict:
+    """Owner's existing seven-day cutoff, with its evidence and manual action.
+
+    DONE can mean read-without-filing. Keep the actual bib+sources ingestion
+    separate; never call a queue completion a successful ingestion.
+    """
+    evidence = snapshot.get("lensing_ingest_evidence")
+    evidence = evidence if isinstance(evidence, dict) else {}
+    repo = _repo_url(snapshot)
+    runbook = f"{repo}/blob/main/skills/catch_up/catch_up.md" if repo else None
+    queue = f"{repo}/blob/main/reading-queue.md#strong-lensing" if repo else None
+    bib = evidence.get("bib") if isinstance(evidence.get("bib"), dict) else None
+    result = {
+        "id": "memory:lensing-catch-up", "scope": "lensing", "state": "unknown",
+        "last_activity": evidence.get("last_ingested"),
+        "last_completed": evidence.get("done"), "last_ingestion": bib,
+        "age_days": None, "threshold_days": CATCH_UP_DAYS, "due_at": None,
+        "checked_at": None, "evidence": [],
+        "reason": "No lensing paper activity is on record; freshness is unknown.",
+        "actions": [{"id": "catch-up", "label": "Copy lensing catch-up prompt",
+                     "kind": "prompt", "target": CATCH_UP_CMD,
+                     "safety": "scientific_judgement"}],
+        "recommended_action_id": "catch-up",
+    }
+    if queue and evidence.get("done"):
+        result["evidence"].append({"kind": "queue_completion", "date": evidence["done"], "url": queue})
+    if bib and repo and bib.get("commit"):
+        result["evidence"].append({"kind": "ingestion_commit", "date": bib.get("date"),
+                                   "url": f"{repo}/commit/{bib['commit']}"})
+    if runbook:
+        result["actions"].append({"id": "runbook", "label": "Catch-up procedure",
+                                  "kind": "link", "target": runbook, "safety": "read_only"})
+    try:
+        now = datetime.datetime.fromisoformat(str(snapshot.get("generated")))
+        if now.tzinfo is None:
+            raise ValueError("missing timezone")
+        today = now.astimezone(datetime.timezone.utc).date()
+        result["checked_at"] = _iso_z(snapshot["generated"])
+    except ValueError:
+        result["reason"] = "The snapshot time is missing or invalid; lensing freshness is unknown."
+        return result
+    dates = [evidence.get("last_ingested"), evidence.get("done"), bib and bib.get("date")]
+    if all(d is None for d in dates):
+        return result
+    try:
+        parsed = [datetime.date.fromisoformat(str(d)) for d in dates if d is not None]
+        cutoff = datetime.date.fromisoformat(str(evidence.get("last_ingested")))
+        if any(d > today for d in parsed):
+            raise ValueError("future evidence")
+        sources = [datetime.date.fromisoformat(str(d)) for d in dates[1:] if d is not None]
+        if not sources or cutoff != max(sources):
+            raise ValueError("inconsistent cutoff")
+    except ValueError:
+        result["reason"] = "Lensing activity dates are invalid, inconsistent or in the future; verify the evidence."
+        return result
+    days = (today - cutoff).days
+    result.update(age_days=days, state="stale" if days >= CATCH_UP_DAYS else "healthy",
+                  due_at=(cutoff + datetime.timedelta(days=CATCH_UP_DAYS)).isoformat() + "T00:00:00Z")
+    result["reason"] = (f"Last recorded lensing paper activity: {cutoff} ({days} days ago); "
+                        f"catch-up is due after {CATCH_UP_DAYS} days. "
+                        "Activity includes queue completion or verified ingestion; paper selection remains human-reviewed.")
+    return result
+
+
 def _catch_up_due(snapshot: dict) -> int | None:
-    """The day count when the catch-up banner should show, else None."""
-    days = _days_since_ingest(snapshot)
-    return days if days is not None and days >= CATCH_UP_DAYS else None
+    model = _lensing_catch_up(snapshot)
+    return model["age_days"] if model["state"] == "stale" else None
 
 
 def _catch_up_md(snapshot: dict) -> list[str]:
-    days = _catch_up_due(snapshot)
-    if days is None:
+    model = _lensing_catch_up(snapshot)
+    if model["state"] == "healthy":
         return []
-    return [f"> _{CATCH_UP_LYRIC}_", ">",
-            f"> **{days} days since papers were last ingested into memory** "
-            f"(last: {snapshot.get('last_ingested')}) — run `{CATCH_UP_CMD}`",
-            ""]
+    title = "Strong-lensing catch-up due" if model["state"] == "stale" else "Strong-lensing freshness unknown"
+    return [f"> **{title}**", f"> {model['reason']}",
+            f"> Next: `{CATCH_UP_CMD}` — human-reviewed paper selection.", ""]
 
 
 def _catch_up_html(snapshot: dict) -> str:
-    days = _catch_up_due(snapshot)
-    if days is None:
+    model = _lensing_catch_up(snapshot)
+    if model["state"] == "healthy":
         return ""
-    last = _html.escape(str(snapshot.get("last_ingested")))
-    return (f"<section class='catchup'><p class='lyric'><em>"
-            f"{_html.escape(CATCH_UP_LYRIC)}</em></p>"
-            f"<p><strong>{days} days since papers were last ingested into "
-            f"memory</strong> <span class='muted'>(last: {last})</span> "
-            f"<code>{_html.escape(CATCH_UP_CMD)}</code> "
-            f"{_copy_btn(CATCH_UP_CMD, 'copy: catch up on strong lensing')}"
-            f"</p></section>")
+    title = "Strong-lensing catch-up due" if model["state"] == "stale" else "Strong-lensing freshness unknown"
+    lyric = (f"<p class='lyric'><em>{_html.escape(CATCH_UP_LYRIC)}</em></p>"
+             if model["state"] == "stale" else "")
+    return (f"<section class='catchup'>{lyric}<p><strong>{title}</strong></p>"
+            f"<p>{_html.escape(model['reason'])}</p>"
+            f"<p><code>{_html.escape(CATCH_UP_CMD)}</code> "
+            f"{_copy_btn(CATCH_UP_CMD, 'copy: catch up on strong lensing')} "
+            "<span class='muted'>Human-reviewed paper selection</span></p></section>")
 
 
 def _totals(snapshot: dict) -> dict:
@@ -1593,6 +1660,7 @@ def to_state(snapshot: dict) -> dict:
         "items": [],
         # Days since memory last ingested a paper (None: nothing on record).
         "days_since_ingest": _days_since_ingest(snapshot),
+        "lensing_catch_up": _lensing_catch_up(snapshot),
     }
     if not t["pages"]:
         return state
@@ -1652,6 +1720,19 @@ def to_state(snapshot: dict) -> dict:
             or None,
             "prompt": None})
 
+    catchup = state["lensing_catch_up"]
+    if catchup["state"] != "healthy":
+        stale = catchup["state"] == "stale"
+        text = "Strong-lensing catch-up due" if stale else "Strong-lensing freshness unknown"
+        reasons.append(text)
+        yellow.append({
+            "id": catchup["id"], "severity": "yellow" if stale else "info",
+            "state": catchup["state"], "text": text, "reason": catchup["reason"],
+            "url": catchup["evidence"][0]["url"] if catchup["evidence"] else page,
+            "prompt": CATCH_UP_CMD, "actions": catchup["actions"],
+            "recommended_action_id": catchup["recommended_action_id"],
+        })
+
     badge = badge_endpoint(snapshot)["message"]
     state.update(
         status="yellow" if reasons else "green",

tests/test_board.py

@@ -11,6 +11,7 @@
 
 import datetime
 import json
+import pytest
 import re
 import sys
 from pathlib import Path
@@ -101,6 +102,7 @@ def _snap_with_remote(tmp_path):
     snap = board.collect(_tree(tmp_path))
     snap["owner"], snap["repo"] = "PyAutoLabs", "PyAutoMemory"
     snap["generated"] = FIXTURE_NOW
+    snap["lensing_ingest_evidence"] = {"last_ingested": FIXTURE_NOW[:10], "done": FIXTURE_NOW[:10], "bib": None}
     return snap
 
 
@@ -830,6 +832,7 @@ def _state_snap(tmp_path):
     snap = board.collect(_tree(tmp_path))
     snap["owner"], snap["repo"] = "SomeOrg", "MemRepo"
     snap["generated"] = FIXTURE_NOW
+    snap["lensing_ingest_evidence"] = {"last_ingested": FIXTURE_NOW[:10], "done": FIXTURE_NOW[:10], "bib": None}
     return snap
 
 
@@ -919,11 +922,12 @@ def test_render_state_is_valid_json(tmp_path):
 
 
 # --- the catch-up banner -------------------------------------------------------------
-# `last_ingested` is computed by `catch_up.last_ingested("all")`; each test
+# Legacy all-scope activity and scoped lensing evidence stay distinct; each test
 # below pins both it and the snapshot's own clock, so no wall clock is read.
 def _banner_snap(tmp_path, last, now="2026-09-30T09:00:00+00:00"):
     snap = _state_snap(tmp_path)
     snap["last_ingested"], snap["generated"] = last, now
+    snap["lensing_ingest_evidence"] = {"last_ingested": last, "done": last, "bib": None}
     return snap
 
 
@@ -948,26 +952,105 @@ def test_banner_at_the_threshold_carries_the_day_count(tmp_path):
     snap = _banner_snap(tmp_path, "2026-09-23")  # exactly CATCH_UP_DAYS
     assert board.CATCH_UP_DAYS == 7
     md = board.render(snap, "md")
-    assert "_…so… I've been lost for a while_" in md
-    assert "**7 days since papers were last ingested into memory**" in md
+    assert "**Strong-lensing catch-up due**" in md
+    assert "2026-09-23 (7 days ago)" in md
+    assert "after 7 days" in md
     assert "`/catch_up lensing`" in md
     # Top of the markdown: before the contents line.
-    assert md.index("lost for a while") < md.index("_Contents")
+    assert md.index("Strong-lensing catch-up due") < md.index("_Contents")
 
 
 def test_banner_html_sits_above_the_hero_with_a_copy_button(tmp_path):
     snap = _banner_snap(tmp_path, "2026-09-11")
     page = board.render(snap, "html")
     assert "<em>…so… I&#x27;ve been lost for a while</em>" in page
-    assert "19 days since papers were last ingested into memory" in page
+    assert "2026-09-11 (19 days ago)" in page
     assert 'data-cmd="/catch_up lensing"' in page
     assert page.index("class='catchup'") < page.index('class="hero"')
     assert board.to_state(snap)["days_since_ingest"] == 19
 
 
-def test_no_banner_when_nothing_is_on_record(tmp_path):
+def test_unknown_banner_when_nothing_is_on_record(tmp_path):
     snap = _banner_snap(tmp_path, None)
     assert board._days_since_ingest(snap) is None
     assert "lost for a while" not in board.render(snap, "md")
-    assert "class='catchup'" not in board.render(snap, "html")
+    assert "Strong-lensing freshness unknown" in board.render(snap, "html")
     assert board.to_state(snap)["days_since_ingest"] is None
+
+
+@pytest.mark.parametrize("age,expected", [(6, "healthy"), (7, "stale"), (8, "stale")])
+def test_lensing_catch_up_threshold_has_one_model_for_every_surface(tmp_path, age, expected):
+    cutoff = (datetime.date(2026, 9, 30) - datetime.timedelta(days=age)).isoformat()
+    snap = _banner_snap(tmp_path, cutoff)
+    model = board._lensing_catch_up(snap)
+    state = board.to_state(snap)
+    assert state["lensing_catch_up"] == model
+    assert model["state"] == expected and model["threshold_days"] == 7
+    assert model["age_days"] == age
+    rows = [i for i in state["items"] if i.get("id") == model["id"]]
+    if expected == "healthy":
+        assert not rows
+    else:
+        assert len(rows) == 1 and rows[0]["reason"] == model["reason"]
+        assert model["reason"] in board.render(snap, "md")
+        assert model["reason"] in board.render(snap, "html")
+        assert rows[0]["actions"][0]["target"] == "/catch_up lensing"
+        assert rows[0]["actions"][0]["safety"] == "scientific_judgement"
+        assert "requires_human_decision" not in rows[0]  # No invented choice.
+
+
+def test_non_lensing_activity_cannot_mask_lensing_staleness(tmp_path):
+    root = _tree(tmp_path)
+    q = root / "reading-queue.md"
+    q.write_text(q.read_text() + "\n## Strong Lensing\nDONE 2026-09-01 — Lens\n\n## SMBHs\nDONE 2026-09-30 — BH\n")
+    snap = board.collect(root)
+    snap["generated"] = "2026-09-30T09:00:00Z"
+    assert snap["last_ingested"] == "2026-09-30"  # Legacy all-scope field.
+    assert board._lensing_catch_up(snap)["last_activity"] == "2026-09-01"
+    assert board._lensing_catch_up(snap)["state"] == "stale"
+
+
+def test_read_completion_is_not_misreported_as_verified_ingestion(tmp_path):
+    snap = _banner_snap(tmp_path, "2026-09-23")
+    model = board._lensing_catch_up(snap)
+    assert model["last_ingestion"] is None
+    assert model["last_completed"] == "2026-09-23"
+    snap["lensing_ingest_evidence"]["bib"] = {"date": "2026-09-21", "commit": "abc123"}
+    model = board._lensing_catch_up(snap)
+    assert model["last_ingestion"] == {"date": "2026-09-21", "commit": "abc123"}
+    assert model["last_activity"] == "2026-09-23"
+    assert any(e["url"].endswith("/commit/abc123") for e in model["evidence"])
+    assert any("reading-queue.md#strong-lensing" in e["url"] for e in model["evidence"])
+
+
+@pytest.mark.parametrize("last", [None, "invalid", "2026-10-02"])
+def test_missing_invalid_and_future_activity_is_unknown(tmp_path, last):
+    snap = _banner_snap(tmp_path, last)
+    model = board._lensing_catch_up(snap)
+    assert model["state"] == "unknown" and model["age_days"] is None
+    row = next(i for i in board.to_state(snap)["items"] if i.get("id") == model["id"])
+    assert row["state"] == "unknown"
+    assert "freshness unknown" in row["text"]
+
+
+def test_invalid_snapshot_clock_is_unknown(tmp_path):
+    model = board._lensing_catch_up(_banner_snap(tmp_path, "2026-09-23", "invalid"))
+    assert model["state"] == "unknown" and model["checked_at"] is None
+
+
+def test_unknown_collector_evidence_does_not_fall_back_to_other_topics(tmp_path, monkeypatch):
+    def failed(*args, **kwargs):
+        raise OSError("unavailable")
+    monkeypatch.setattr(board.catch_up, "ingest_evidence", failed)
+    snap = board.collect(_tree(tmp_path))
+    snap["last_ingested"] = "2026-09-30"
+    snap["generated"] = "2026-09-30T09:00:00Z"
+    assert board._lensing_catch_up(snap)["state"] == "unknown"
+
+
+def test_cutoff_requires_matching_scoped_evidence(tmp_path):
+    snap = _banner_snap(tmp_path, "2026-09-29")
+    snap["lensing_ingest_evidence"]["done"] = "2026-09-01"
+    assert board._lensing_catch_up(snap)["state"] == "unknown"
+    snap["lensing_ingest_evidence"]["done"] = None
+    assert board._lensing_catch_up(snap)["state"] == "unknown"