feat(manager): type declared evidence-source health in the turn window

loopx/chat_manager_context.py

@@ -42,6 +42,33 @@
 # on demand through the read tool, a prompt-only segment can only receive them.
 MANAGER_REMOTE_READ_INLINE = "inline_in_prompt"
 MANAGER_REMOTE_READ_ON_DEMAND = "on_demand_tool"
+# A declared source that contributed nothing to the window is a coverage fact,
+# not a quiet "no progress" reading. Every declared source therefore carries a
+# typed health row with the same four fields the read refusals already use
+# (source id, typed reason, coverage effect, next action) so the answer can cite
+# a row instead of narrating staleness as a disclaimer. The vocabulary is a
+# closed set: a source is either read in this window, declared but not read, or
+# impossible to reach because its configured alias is missing.
+MANAGER_SOURCE_FRESHNESS_VALUES = ("current", "stale", "unknown")
+MANAGER_SOURCE_READ_FRESHNESS = "current"
+MANAGER_SOURCE_UNREAD_FRESHNESS = "stale"
+MANAGER_SOURCE_UNKNOWN_FRESHNESS = "unknown"
+MANAGER_SOURCE_NOT_READ_REASON = "declared_source_not_read_this_turn"
+MANAGER_SOURCE_NOT_READ_NEXT_ACTION = (
+    "Read this source with the manager evidence read tool, or let the next "
+    "Turn's source rotation dial it."
+)
+MANAGER_SOURCE_UNCONFIGURED_NEXT_ACTION = (
+    "Register an existing SSH alias for this host and retry."
+)
+MANAGER_LOCAL_SOURCE_NOT_READ_REASON = "local_evidence_not_read_this_turn"
+MANAGER_LOCAL_SOURCE_NOT_READ_NEXT_ACTION = (
+    "Collect this window with details enabled, or narrow the evidence window."
+)
+MANAGER_SOURCE_COVERAGE_EFFECT = (
+    "no evidence was read from this source; the answer must not present it as "
+    "no progress"
+)
 
 
 def resolve_evidence_window_days(environ=None) -> tuple[int, str, str]:
@@ -97,6 +124,65 @@ def _declared_sources(
     return declared or local
 
 
+def _source_health_rows(
+    sources: list[dict[str, Any]], *, read_status: str
+) -> list[dict[str, Any]]:
+    """Type what each declared source contributed to this window.
+
+    A source that was declared but read nothing is a coverage fact the answer
+    must name, never evidence that a Goal made no progress. Every row reuses the
+    declaration's own status vocabulary and adds freshness, the typed reason,
+    the coverage effect and the next action, so the packet carries the same four
+    fields the read refusals already use.
+    """
+
+    rows: list[dict[str, Any]] = []
+    for source in sources:
+        source_id = str(source.get("source_id") or "")
+        status = str(source.get("status") or "unknown")
+        read = status == "available" and read_status in {"read", "partial"}
+        if read:
+            freshness = MANAGER_SOURCE_READ_FRESHNESS
+            reason = None
+            coverage_effect = None
+            next_action = None
+        elif status == "available":
+            # Reachable but unread in this window: it cannot support a
+            # progress claim, and the answer has to say so with a row.
+            freshness = MANAGER_SOURCE_UNKNOWN_FRESHNESS
+            reason = MANAGER_LOCAL_SOURCE_NOT_READ_REASON
+            coverage_effect = MANAGER_SOURCE_COVERAGE_EFFECT
+            next_action = MANAGER_LOCAL_SOURCE_NOT_READ_NEXT_ACTION
+        elif status == "not_read":
+            freshness = MANAGER_SOURCE_UNREAD_FRESHNESS
+            reason = MANAGER_SOURCE_NOT_READ_REASON
+            coverage_effect = MANAGER_SOURCE_COVERAGE_EFFECT
+            next_action = MANAGER_SOURCE_NOT_READ_NEXT_ACTION
+        else:
+            # not_configured, and any future non-available declaration: the
+            # source cannot be read at all, which is drift the answer names.
+            freshness = MANAGER_SOURCE_UNKNOWN_FRESHNESS
+            reason = str(source.get("reason") or f"source_status_{status}")
+            coverage_effect = MANAGER_SOURCE_COVERAGE_EFFECT
+            next_action = (
+                MANAGER_SOURCE_UNCONFIGURED_NEXT_ACTION
+                if status == "not_configured"
+                else MANAGER_SOURCE_NOT_READ_NEXT_ACTION
+            )
+        rows.append(
+            {
+                "source_id": source_id,
+                "source_host": source.get("source_host"),
+                "status": status,
+                "freshness": freshness,
+                "reason": reason,
+                "coverage_effect": coverage_effect,
+                "next_action": next_action,
+            }
+        )
+    return rows
+
+
 def _bounded_receipts(history: dict[str, Any]) -> dict[str, Any]:
     """Keep each Goal's newest receipt whole and compact the older ones."""
     deliveries = history.get("deliveries")
@@ -223,6 +309,10 @@ def _evidence_window(
         "matched_by_day": matched_by_day,
         "invalid_delivery_records": invalid,
         "sources": sources,
+        # Every declared source reports its own health, so an unread or
+        # unreachable source is a typed row rather than a staleness caveat the
+        # answer has to narrate.
+        "source_health": _source_health_rows(sources, read_status=read_status),
         "declared_unread_sources": [
             str(source.get("source_id"))
             for source in sources

tests/test_chat_manager_context.py

@@ -275,6 +275,130 @@ def test_turn_context_reads_a_bounded_window_and_declares_sources(
     ]
 
 
+def test_source_health_rows_type_a_window_that_read_nothing():
+    declared = [{"source_id": "local", "source_host": "local", "status": "available"}]
+
+    unread = context._source_health_rows(declared, read_status="not_read")
+
+    assert unread == [
+        {
+            "source_id": "local",
+            "source_host": "local",
+            "status": "available",
+            "freshness": "unknown",
+            "reason": context.MANAGER_LOCAL_SOURCE_NOT_READ_REASON,
+            "coverage_effect": context.MANAGER_SOURCE_COVERAGE_EFFECT,
+            "next_action": context.MANAGER_LOCAL_SOURCE_NOT_READ_NEXT_ACTION,
+        }
+    ]
+    read = context._source_health_rows(declared, read_status="read")
+    assert read[0]["freshness"] == "current"
+    assert read[0]["reason"] is None
+    assert read[0]["coverage_effect"] is None
+    assert read[0]["next_action"] is None
+
+
+def test_remote_source_rows_are_typed_and_never_read_as_no_progress():
+    declared = [
+        {"source_id": "local", "source_host": "local", "status": "available"},
+        {
+            "source_id": "ssh:ark-devbox",
+            "source_host": "ark-devbox",
+            "status": "not_read",
+            "reason": None,
+            "scope": "remote_registry",
+        },
+        {
+            "source_id": "ssh:gone",
+            "source_host": "gone",
+            "status": "not_configured",
+            "reason": "ssh_alias_not_configured",
+        },
+    ]
+
+    health = {
+        row["source_id"]: row
+        for row in context._source_health_rows(declared, read_status="read")
+    }
+
+    assert health["local"]["freshness"] == "current"
+    assert health["ssh:ark-devbox"]["status"] == "not_read"
+    assert health["ssh:ark-devbox"]["freshness"] == "stale"
+    assert health["ssh:ark-devbox"]["reason"] == context.MANAGER_SOURCE_NOT_READ_REASON
+    assert health["ssh:gone"]["freshness"] == "unknown"
+    assert health["ssh:gone"]["reason"] == "ssh_alias_not_configured"
+    assert health["ssh:gone"]["next_action"] == context.MANAGER_SOURCE_UNCONFIGURED_NEXT_ACTION
+    for source_id, row in health.items():
+        assert row["freshness"] in context.MANAGER_SOURCE_FRESHNESS_VALUES
+        if row["freshness"] == "current":
+            continue
+        # The coverage effect is what stops a reader from reading a source that
+        # contributed nothing as "this Goal made no progress".
+        assert row["coverage_effect"] == context.MANAGER_SOURCE_COVERAGE_EFFECT
+        assert "must not present it as no progress" in row["coverage_effect"]
+        assert row["next_action"], source_id
+
+
+def test_turn_context_reports_health_for_every_declared_source(monkeypatch, tmp_path):
+    _write_delivery_index(
+        tmp_path,
+        "alpha",
+        [
+            {
+                "generated_at": datetime.now(timezone.utc).isoformat(),
+                "goal_id": "alpha",
+                "agent_id": "worker",
+                "todo_id": "todo_1",
+                "classification": "validated_progress",
+                "delivery_outcome": "outcome_progress",
+                "recommended_action": "follow up",
+            }
+        ],
+    )
+    monkeypatch.setattr(
+        context,
+        "build_goal_portfolio",
+        lambda **_: {
+            "goals": [{"goal_id": "alpha", "activation_state": "active"}],
+            "coverage": {"discovered": 1},
+        },
+    )
+    monkeypatch.setattr(
+        context,
+        "read_manager_goal_details",
+        lambda *args, **kwargs: {"status": "read", "todos": []},
+    )
+    monkeypatch.setattr(
+        context,
+        "_declared_sources",
+        lambda *args, **kwargs: [
+            {"source_id": "local", "source_host": "local", "status": "available"},
+            {
+                "source_id": "ssh:ark-devbox",
+                "source_host": "ark-devbox",
+                "status": "not_read",
+                "reason": None,
+            },
+        ],
+    )
+
+    result = context.manager_turn_context(
+        tmp_path / "registry.json", {"channel_id": "manager"}, tmp_path
+    )
+
+    window = result["evidence_window"]
+    assert window["read_status"] == "read"
+    assert [row["source_id"] for row in window["source_health"]] == [
+        source["source_id"] for source in window["sources"]
+    ]
+    health = {row["source_id"]: row for row in window["source_health"]}
+    assert health["local"]["freshness"] == "current"
+    assert health["ssh:ark-devbox"]["freshness"] == "stale"
+    assert health["ssh:ark-devbox"]["coverage_effect"] == (
+        context.MANAGER_SOURCE_COVERAGE_EFFECT
+    )
+
+
 def test_evidence_window_is_a_selected_bounded_decision():
     # The window is an explicit operator choice, bounded, and declared with its
     # source: it never follows a discovered environment fact silently.