diff --git a/loopx/chat_manager_context.py b/loopx/chat_manager_context.py index 804ce61d8f..77d702bcd2 100644 --- a/loopx/chat_manager_context.py +++ b/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 diff --git a/tests/test_chat_manager_context.py b/tests/test_chat_manager_context.py index f0f58b1f41..963a4b133c 100644 --- a/tests/test_chat_manager_context.py +++ b/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.