Expose actionable arXiv digest freshness in the cockpit

README.md

@@ -98,6 +98,16 @@ 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 feed also exposes independent `digests.lensing` and `digests.interests`
+records, shared with the board freshness indicators. Each records its canonical
+state, last recorded date, checked time, elapsed weekdays, two-weekday threshold,
+reason and evidence/action links. Empty queues with a recent stamp stay healthy;
+missing, invalid or future stamps are unknown. The stamp proves a digest was
+recorded, not that every workflow step succeeded. Stale/unknown cockpit rows
+link to the owning Mind workflow and offer a manual investigation prompt
+(`requires_approval`); detection never reruns a workflow. Digest delivery and
+human lensing catch-up remain separate clocks.
+
 The wiki schema is defined in
 [`wiki/AGENTS.md`](wiki/AGENTS.md) and inherited by every
 sub-wiki. How agents should read this repo: [AGENTS.md](AGENTS.md). The

scripts/board.py

@@ -810,37 +810,81 @@ def _stub_prompt(snapshot: dict, wiki: str) -> str:
             f"in wiki/AGENTS.md, set status: drafted, and run make validate.")
 
 
-def _inbox_freshness(snapshot: dict, key: str = "inbox_last_digest") -> dict:
-    """A suggestion tier's freshness facts, shared by both renderers.
+# The existing Mind producers write independent date stamps in these files.
+DIGESTS = {
+    "inbox_last_digest": ("lensing", "arXiv inbox", "arxiv-inbox.md", "arxiv_papers.yml"),
+    "interests_last_digest": ("interests", "arXiv interests", "arxiv-interests.md", "arxiv_interests.yml"),
+}
+
 
-    ``key`` picks the tier — the strong-lensing inbox or the interests list.
-    Each has its own nightly digest and so its own way to be silently broken.
+def _inbox_freshness(snapshot: dict, key: str = "inbox_last_digest") -> dict:
+    """One owner policy for both board renderers and the cockpit feed.
 
-    ``stale`` is what the *markdown* fallback can say. The HTML page must not
-    trust it for the warning: see ``_EXTRA_JS``.
+    A filing stamp is evidence of a recorded digest, not a GitHub run ID or
+    proof that every step of that workflow succeeded. Empty queues do not
+    change freshness: a quiet-day stamp is just as valid as a populated one.
     """
+    scope, label, filename, workflow = DIGESTS[key]
     stamp = snapshot.get(key)
+    repo = _repo_url(snapshot)
+    evidence_url = f"{repo}/blob/main/{filename}" if repo else None
+    owner = snapshot.get("owner")
+    workflow_url = (f"https://github.com/{owner}/PyAutoMind/actions/workflows/{workflow}"
+                    if owner else None)
+    prompt = (f"/bug investigate {label} digest freshness; inspect the recorded "
+              f"date in {filename} and the {workflow} workflow before choosing a remedy")
+    if workflow_url:
+        prompt += f" — {workflow_url}"
+    actions = []
+    if workflow_url:
+        actions.append({"id": "inspect-workflow", "label": "Inspect workflow", "kind": "link",
+                        "target": workflow_url, "safety": "read_only"})
+    if evidence_url:
+        actions.append({"id": "view-stamp", "label": "View digest stamp", "kind": "link",
+                        "target": evidence_url, "safety": "read_only"})
+    actions.append({"id": "investigate", "label": "Copy investigation prompt", "kind": "prompt",
+                    "target": prompt, "safety": "requires_approval"})
+    result = {"id": f"memory:digest:{scope}", "scope": scope, "label": label,
+              "stamp": stamp, "last_recorded": None, "checked_at": None,
+              "weekdays": None, "threshold_weekdays": inbox_actions.INBOX_STALE_WEEKDAYS,
+              "state": "unknown", "stale": False, "evidence_url": evidence_url,
+              "workflow_url": workflow_url, "actions": actions,
+              "recommended_action_id": "inspect-workflow" if workflow_url else "investigate",
+              "reason": "no run recorded yet — digest freshness unknown"}
     try:
-        today = datetime.datetime.fromisoformat(
-            str(snapshot.get("generated"))).date()
+        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 (TypeError, ValueError):
-        today = datetime.datetime.now(datetime.timezone.utc).date()
-    return {
-        "stamp": stamp,
-        "weekdays": inbox_actions.weekdays_since(stamp, today) if stamp else 0,
-        "stale": inbox_actions.is_stale(stamp, today),
-    }
+        result["reason"] = "snapshot time is missing or invalid — digest freshness unknown"
+        return result
+    if stamp is None:
+        return result
+    try:
+        parsed = datetime.date.fromisoformat(stamp)
+        if parsed.isoformat() != stamp or parsed > today:
+            raise ValueError("future or noncanonical date")
+    except (TypeError, ValueError):
+        result["reason"] = "digest date is invalid or in the future — freshness unknown; inspect the stamp"
+        return result
+    weekdays = inbox_actions.weekdays_since(stamp, today)
+    stale = inbox_actions.is_stale(stamp, today)
+    result.update(last_recorded=stamp, weekdays=weekdays, stale=stale,
+                  state="stale" if stale else "healthy")
+    result["reason"] = (f"no digest since {stamp} ({weekdays} weekdays) — the nightly filing may be broken; "
+                        f"threshold: {inbox_actions.INBOX_STALE_WEEKDAYS} weekdays" if stale else
+                        f"digest recorded {stamp}; {weekdays} weekdays elapsed, below the "
+                        f"{inbox_actions.INBOX_STALE_WEEKDAYS}-weekday threshold")
+    return result
 
 
 def _inbox_empty_note(fresh: dict) -> str:
-    """Why a tier is empty — quiet, suspect, or never run. Plain text."""
-    if fresh["stale"]:
-        return (f"no digest since {fresh['stamp']} "
-                f"({fresh['weekdays']} weekdays) — the nightly filing may be "
-                f"broken")
-    if fresh["stamp"]:
-        return f"the last digest ran {fresh['stamp']} and found nothing"
-    return "the nightly arXiv digest fills this — no run recorded yet"
+    """Why a tier is empty without inferring the number of papers fetched."""
+    if fresh["state"] != "healthy":
+        return fresh["reason"]
+    return f"the last digest ran {fresh['stamp']}; nothing is waiting"
 
 
 # --- renderers ------------------------------------------------------------------
@@ -880,7 +924,9 @@ def _render_md(snapshot: dict) -> str:
     inbox = snapshot.get("inbox") or []
     fresh = _inbox_freshness(snapshot)
     if inbox:
-        digest = (f"last digest {fresh['stamp']} · " if fresh["stamp"] else "")
+        digest = (f"last digest {fresh['last_recorded']} · " if fresh["last_recorded"] else "")
+        if fresh["state"] != "healthy":
+            digest += fresh["reason"] + " · "
         warn = ("⚠ " if fresh["stale"] else "")
         lines += [f"_{len(inbox)} waiting · {warn}{digest}un-acted suggestions "
                   f"lapse after {inbox_actions.INBOX_WINDOW_DAYS} days._", ""]
@@ -906,6 +952,7 @@ def _render_md(snapshot: dict) -> str:
     interests = snapshot.get("interests") or []
     ifresh = _inbox_freshness(snapshot, "interests_last_digest")
     if interests:
+        lines += [f"_{ifresh['reason']}_", ""]
         date = snapshot.get("interests_date") or "?"
         left = max(0, (snapshot.get("interests_batches") or 1) - 1)
         backlog = (f" · {left} more day{'s' if left != 1 else ''} behind it"
@@ -1343,7 +1390,9 @@ def _fresh_span(fresh: dict, now_text: str, stale_text: str,
                     cls: str = "meta") -> str:
         """`fresh` is explicit because the page renders two tiers, each with
         its own digest, its own stamp and its own way of going silent."""
-        if not fresh["stamp"]:
+        if fresh["state"] == "unknown":
+            if fresh["reason"] not in now_text:
+                now_text += " · " + fresh["reason"]
             return f"<span class='{cls}'>{_html.escape(now_text)}</span>"
         if fresh["stale"]:
             # Already stale at render time: show the warning now, so a reader
@@ -1661,24 +1710,28 @@ def to_state(snapshot: dict) -> dict:
         # 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),
+        "digests": {spec[0]: _inbox_freshness(snapshot, key) for key, spec in DIGESTS.items()},
     }
     if not t["pages"]:
         return state
 
     yellow: list[dict] = []
     info: list[dict] = []
     reasons: list[str] = []
-    for key, label in (("inbox_last_digest", "arXiv inbox"),
-                       ("interests_last_digest", "arXiv interests")):
-        fresh = _inbox_freshness(snapshot, key)
-        if fresh["stale"]:
-            reasons.append(f"{label} digest stale")
+    for fresh in state["digests"].values():
+        if fresh["state"] != "healthy":
+            label = fresh["label"]
+            text = f"{label} digest stale" if fresh["stale"] else f"{label} digest freshness unknown"
+            reasons.append(text)
+            prompt = next(a["target"] for a in fresh["actions"] if a["id"] == "investigate")
             yellow.append({
-                "severity": "yellow",
-                "text": _clip(f"{label}: no digest since {fresh['stamp']} "
-                              f"({fresh['weekdays']} weekdays) — the nightly "
-                              f"filing may be broken"),
-                "url": pages_url(snapshot) or None, "prompt": None})
+                "id": fresh["id"], "state": fresh["state"],
+                "severity": "yellow" if fresh["stale"] else "info",
+                "text": _clip(text), "reason": fresh["reason"],
+                "url": fresh["evidence_url"], "prompt": prompt,
+                "actions": fresh["actions"],
+                "recommended_action_id": fresh["recommended_action_id"],
+            })
     filings = snapshot.get("filings") or []
     if filings:
         reasons.append("filings awaiting merge")

tests/test_board.py

@@ -448,7 +448,7 @@ def test_the_stamp_is_collected(tmp_path):
 def test_a_quiet_day_says_the_digest_ran(tmp_path):
     """The state the old wording could not distinguish from a broken run."""
     md = board._render_md(_inbox(tmp_path, "2026-08-25"))
-    assert "the last digest ran 2026-08-25 and found nothing" in md
+    assert "the last digest ran 2026-08-25; nothing is waiting" in md
     assert "⚠" not in md
 
 
@@ -716,10 +716,10 @@ def test_the_two_tiers_carry_their_own_stamps(tmp_path):
     """One digest can break while the other keeps running; the page must not
     report the healthy one's date for the broken one."""
     snap = _snap_with_remote(tmp_path)
-    snap["inbox_last_digest"] = "2026-08-27"
+    snap["inbox_last_digest"] = "2026-08-25"
     snap["interests_last_digest"] = "2026-08-20"
     html = board._render_html(snap)
-    assert "data-last-digest='2026-08-27'" in html
+    assert "data-last-digest='2026-08-25'" in html
     assert "data-last-digest='2026-08-20'" in html
 
 
@@ -832,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["inbox_last_digest"] = snap["interests_last_digest"] = FIXTURE_NOW[:10]
     snap["lensing_ingest_evidence"] = {"last_ingested": FIXTURE_NOW[:10], "done": FIXTURE_NOW[:10], "bib": None}
     return snap
 
@@ -895,7 +896,7 @@ def test_a_stale_inbox_turns_the_state_yellow(tmp_path):
     assert state["status"] == "yellow"
     assert state["headline"].startswith("arXiv inbox digest stale")
     assert state["items"][0]["severity"] == "yellow"
-    assert "no digest since 2026-08-19" in state["items"][0]["text"]
+    assert "no digest since 2026-08-19" in state["items"][0]["reason"]
 
 
 def test_pending_filings_turn_the_state_yellow(tmp_path):
@@ -1054,3 +1055,89 @@ def test_cutoff_requires_matching_scoped_evidence(tmp_path):
     assert board._lensing_catch_up(snap)["state"] == "unknown"
     snap["lensing_ingest_evidence"]["done"] = None
     assert board._lensing_catch_up(snap)["state"] == "unknown"
+
+
+@pytest.mark.parametrize('today,age,state', [
+    ('2026-08-21', 0, 'healthy'), ('2026-08-23', 0, 'healthy'),
+    ('2026-08-24', 1, 'healthy'), ('2026-08-25', 2, 'stale'),
+])
+def test_digest_weekday_boundary(tmp_path, today, age, state):
+    snap = _state_snap(tmp_path)
+    snap.update(generated=today + 'T09:00:00Z', inbox_last_digest='2026-08-21')
+    fresh = board._inbox_freshness(snap)
+    assert (fresh['weekdays'], fresh['state']) == (age, state)
+    assert fresh['threshold_weekdays'] == 2
+    assert fresh['last_recorded'] == '2026-08-21'
+
+
+@pytest.mark.parametrize('stamp', [None, '', 'bad', '2026-02-30',
+                                  '2026-08-27', '20260826', 123])
+def test_untrustworthy_digest_stamp_is_unknown(tmp_path, stamp):
+    snap = _state_snap(tmp_path)
+    snap['inbox_last_digest'] = stamp
+    fresh = board.to_state(snap)['digests']['lensing']
+    assert fresh['state'] == 'unknown'
+    assert fresh['last_recorded'] is None and fresh['weekdays'] is None
+    row = next(r for r in board.to_state(snap)['items']
+               if r.get('id') == 'memory:digest:lensing')
+    assert row['state'] == 'unknown' and row['reason'] == fresh['reason']
+    assert "data-last-digest='bad'" not in board._render_html(snap)
+    assert fresh['reason'] in board._render_md(snap)
+
+
+@pytest.mark.parametrize('clock', [None, 'bad', '2026-08-26T09:00:00'])
+def test_digest_requires_a_valid_observation_time(tmp_path, clock):
+    snap = _state_snap(tmp_path)
+    snap['generated'] = clock
+    fresh = board._inbox_freshness(snap)
+    assert fresh['state'] == 'unknown' and fresh['checked_at'] is None
+
+
+def test_digest_scopes_share_feed_evidence_and_explicit_actions(tmp_path):
+    snap = _state_snap(tmp_path)
+    snap['inbox_last_digest'] = '2026-08-21'
+    state = board.to_state(snap)
+    assert state['digests']['interests']['state'] == 'healthy'
+    assert state['digests']['lensing']['state'] == 'stale'
+    rows = [r for r in state['items'] if r.get('id', '').startswith('memory:digest:')]
+    assert len(rows) == 1
+    row = rows[0]
+    fresh = state['digests']['lensing']
+    assert fresh == board._inbox_freshness(snap)
+    assert row['actions'] == fresh['actions']
+    assert row['recommended_action_id'] == 'inspect-workflow'
+    assert fresh['workflow_url'].endswith('/PyAutoMind/actions/workflows/arxiv_papers.yml')
+    assert fresh['evidence_url'].endswith('/MemRepo/blob/main/arxiv-inbox.md')
+    assert state['digests']['interests']['workflow_url'].endswith('/arxiv_interests.yml')
+    assert state['digests']['interests']['evidence_url'].endswith('/arxiv-interests.md')
+    assert all(a['safety'] == ('read_only' if a['kind'] == 'link' else 'requires_approval')
+               for a in row['actions'])
+    assert row['prompt'].startswith('/bug investigate')
+
+
+def test_quiet_digest_and_backlog_have_same_freshness(tmp_path):
+    snap = _state_snap(tmp_path)
+    populated = board.to_state(snap)['digests']
+    snap['inbox'], snap['interests'] = [], []
+    assert board.to_state(snap)['digests'] == populated
+    assert all(f['state'] == 'healthy' for f in populated.values())
+    assert not any(r.get('id', '').startswith('memory:digest:')
+                   for r in board.to_state(snap)['items'])
+
+
+def test_digest_without_remote_offers_manual_investigation(tmp_path):
+    snap = _state_snap(tmp_path)
+    snap['owner'], snap['repo'] = None, None
+    fresh = board._inbox_freshness(snap)
+    assert fresh['workflow_url'] is None and fresh['evidence_url'] is None
+    assert [a['id'] for a in fresh['actions']] == ['investigate']
+    assert fresh['recommended_action_id'] == 'investigate'
+
+
+def test_invalid_digest_is_escaped_and_never_given_to_browser_clock(tmp_path):
+    snap = _state_snap(tmp_path)
+    snap['inbox_last_digest'] = '<script>oops</script>'
+    html = board._render_html(snap)
+    assert '<script>oops</script>' not in html
+    assert 'digest date is invalid or in the future' in html
+    assert "data-last-digest='<" not in html