From 25e1b013f05c458f080d42e70481cb075c9c7b7d Mon Sep 17 00:00:00 2001 From: "liuhuadong.hans" Date: Mon, 5 Oct 2026 00:10:21 +0800 Subject: [PATCH 1/3] feat(projection): seal global todos and risks reads Signed-off-by: liuhuadong.hans --- loopx/global_risks.py | 146 ++++++++++++++++++++++++++++++++++++++++-- loopx/global_todos.py | 109 ++++++++++++++++++++++++++++++- 2 files changed, 245 insertions(+), 10 deletions(-) diff --git a/loopx/global_risks.py b/loopx/global_risks.py index b2b5c1b692..a920b8aac7 100644 --- a/loopx/global_risks.py +++ b/loopx/global_risks.py @@ -8,6 +8,12 @@ from pathlib import Path from typing import Any, Protocol, cast +from .control_plane.projection_envelope_facts import ( + render_projection_envelope_markdown, + seal_projection_envelope, + source_fact, +) + class _CollectStatus(Protocol): def __call__( self, @@ -157,6 +163,8 @@ def build_global_risks_error( *, time_range: str = "24h", error_code: str = "global_risks_unavailable", + status_payload: dict[str, Any] | None = None, + status_read_at: str | None = None, ) -> dict[str, Any]: return { "ok": False, @@ -168,6 +176,12 @@ def build_global_risks_error( "omissions": [ "Raw/private failure details and local paths were intentionally omitted." ], + "projection_envelope": _risks_envelope( + status_payload or {}, status_read_at=status_read_at, + host_poll_status="not_read", host_poll_read_at=None, + host_poll_goal_count=None, source_rows_omitted=0, + shown_count=0, available_count=0, agent_id=None, + ), "boundary": public_safe_boundary(), } @@ -390,14 +404,14 @@ def _collect_stale_host_poll_risks( registry_path: Path, scan_limit: int, warnings: list[dict[str, Any]], -) -> list[dict[str, Any]]: +) -> tuple[list[dict[str, Any]], str, int | None, int]: """Scan project-local host poll receipts for loops that died mid-wait.""" risks: list[dict[str, Any]] = [] try: registry_payload = load_registry(registry_path) except FileNotFoundError: - return risks + return risks, "missing", None, 0 except (OSError, ValueError) as exc: warnings.append( _warning( @@ -406,13 +420,15 @@ def _collect_stale_host_poll_risks( detail=str(exc)[:200], ) ) - return risks + return risks, "unreadable", None, 0 goals = registry_payload.get("goals") if isinstance(registry_payload, dict) else None if not isinstance(goals, list): - return risks + return risks, "unreadable", None, 0 + inspected_count = 0 for source_index, goal in enumerate(goals): if len(risks) >= scan_limit: break + inspected_count += 1 if not isinstance(goal, dict) or not goal.get("id"): continue state_path = resolve_goal_local_path( @@ -460,7 +476,7 @@ def _collect_stale_host_poll_risks( "requires_user_approval": True, } ) - return risks + return risks, "read", len(goals), len(goals) - inspected_count def _normalize_stale_warning( @@ -688,15 +704,108 @@ def _groups(risks: list[dict[str, Any]]) -> dict[str, list[dict[str, Any]]]: } +def _risks_envelope( + status_payload: dict[str, Any], + *, + status_read_at: str | None, + host_poll_status: str, + host_poll_read_at: str | None, + host_poll_goal_count: int | None, + source_rows_omitted: int, + host_poll_goals_omitted: int = 0, + shown_count: int, + available_count: int, + agent_id: str | None, +) -> dict[str, Any]: + global_registry = as_dict(status_payload.get("global_registry")) + status_envelope = as_dict(status_payload.get("projection_envelope")) + global_source = next( + (row for row in as_list(status_envelope.get("sources")) + if isinstance(row, dict) and row.get("source_id") == "global_registry"), + {}, + ) + available = global_registry.get("available") is True + count = global_registry.get("global_goal_count") + expected = count if available and isinstance(count, int) and not isinstance(count, bool) else None + excluded = int(global_registry.get("current_registry_excluded_goal_count") or 0) if available else 0 + omitted = [] + if excluded: + omitted.append({ + "reason": "outside_current_registry", "count": excluded, + "refs": [str(ref) for ref in as_list(global_registry.get("current_registry_excluded_goal_ids"))[:8]], + }) + if source_rows_omitted: + omitted.append({"reason": "source_rows_not_scanned", "count": source_rows_omitted}) + if host_poll_goals_omitted: + omitted.append({"reason": "host_poll_goals_not_scanned", "count": host_poll_goals_omitted}) + if status_payload and status_payload.get("ok") is not True: + omitted.append({"reason": "status_unavailable", "count": 1}) + contract = as_dict(status_payload.get("contract")) + queue = as_dict(status_payload.get("attention_queue")) + history = as_dict(status_payload.get("run_history")) + contract_read = isinstance(status_payload.get("contract"), dict) and isinstance(contract.get("error_diagnostics", []), list) + queue_read = isinstance(status_payload.get("attention_queue"), dict) and isinstance(queue.get("items", []), list) + sources = [ + source_fact( + "global_registry", + read_status=( + str(global_source.get("read_status") or "read") if available + else str(global_registry.get("read_status") or ("missing" if status_read_at else "not_read")) + ), + last_read_at=global_source.get("last_read_at") or (status_read_at if available else None), + item_count=expected, + ), + source_fact( + "status_contract", + read_status="read" if contract_read else "not_read", + last_read_at=status_read_at if contract_read else None, + item_count=len(as_list(contract.get("error_diagnostics"))) if contract_read else None, + ), + source_fact( + "attention_queue", + read_status="read" if queue_read else "not_read", + last_read_at=status_read_at if queue_read else None, + item_count=len(as_list(queue.get("items"))) if queue_read else None, + ), + source_fact( + "host_poll_receipts", read_status=host_poll_status, + last_read_at=host_poll_read_at, item_count=host_poll_goal_count, + ), + *([] if status_envelope else [source_fact("status", read_status="not_read")]), + ] + if agent_id: + sources.append(source_fact( + "agent_scope", + read_status="read" if isinstance(history.get("goals"), list) else "not_read", + last_read_at=status_read_at if isinstance(history.get("goals"), list) else None, + item_count=len(as_list(history.get("goals"))) if history else None, + )) + return seal_projection_envelope( + projection="global_risks", observed_at=now_utc_iso(), + sources=sources, + coverage={ + "scope": "global", "expected_count": expected, + "included_count": max(0, expected - excluded) if expected is not None else 0, + "omitted": omitted, + "shown_count": shown_count, "available_count": available_count, + }, + upstream=[status_envelope] if status_envelope else [], + ) + + def _malformed_projection_error( error: object, *, time_range: str, + status_payload: dict[str, Any] | None = None, + status_read_at: str | None = None, ) -> dict[str, Any]: return build_global_risks_error( error, time_range=time_range, error_code="malformed_status_projection", + status_payload=status_payload, + status_read_at=status_read_at, ) @@ -731,6 +840,7 @@ def build_global_risks( time_range=normalized_time_range, ) status_payload: dict[str, Any] = status_result + status_read_at = now_utc_iso() try: contract = _required_container(status_payload, "contract") global_registry = _required_container(status_payload, "global_registry") @@ -739,7 +849,10 @@ def build_global_risks( findings = _optional_source_list(global_registry, "findings") items = _optional_source_list(attention_queue, "items") except ValueError as exc: - return _malformed_projection_error(exc, time_range=normalized_time_range) + return _malformed_projection_error( + exc, time_range=normalized_time_range, + status_payload=status_payload, status_read_at=status_read_at, + ) warnings: list[dict[str, Any]] = [] risks: list[dict[str, Any]] = [] @@ -800,12 +913,14 @@ def build_global_risks( ) ) - host_poll_risks = _collect_stale_host_poll_risks( + host_poll_risks, host_poll_status, host_poll_goal_count, host_poll_goals_omitted = _collect_stale_host_poll_risks( registry_path=registry_path, scan_limit=scan_limit, warnings=warnings, ) + host_poll_read_at = now_utc_iso() if host_poll_status == "read" else None risks.extend(host_poll_risks) + source_rows_truncated = source_rows_truncated or host_poll_goals_omitted > 0 try: risks, history_truncated = _filter_for_agent( @@ -820,6 +935,8 @@ def build_global_risks( exc, time_range=normalized_time_range, error_code="agent_scope_unavailable", + status_payload=status_payload, + status_read_at=status_read_at, ) source_rows_truncated = source_rows_truncated or history_truncated @@ -859,6 +976,19 @@ def build_global_risks( "risks": retained, "source_warnings": warnings[:SOURCE_WARNING_LIMIT], "source_warnings_truncated": warning_count > SOURCE_WARNING_LIMIT, + "projection_envelope": _risks_envelope( + status_payload, status_read_at=status_read_at, + host_poll_status=host_poll_status, + host_poll_read_at=host_poll_read_at, + host_poll_goal_count=host_poll_goal_count, + source_rows_omitted=sum( + max(0, int(warning.get("available_count") or 0) - int(warning.get("inspected_count") or 0)) + for warning in warnings if warning.get("reason_code") == "source_rows_truncated" + ), + host_poll_goals_omitted=host_poll_goals_omitted, + shown_count=returned_risk_count, available_count=matched_risk_count, + agent_id=agent_id, + ), "omissions": [dict(_ROLLBACK_OMISSION)], "boundary": public_safe_boundary(), } @@ -911,6 +1041,7 @@ def render_global_risks_markdown(payload: dict[str, Any]) -> str: "- ok: `False`", f"- error_code: `{_redact_text(payload.get('error_code'), limit=120)}`", f"- error: {_redact_text(payload.get('error'))}", + *render_projection_envelope_markdown(payload.get("projection_envelope")), ] omissions = [ _redact_text(item) @@ -932,6 +1063,7 @@ def render_global_risks_markdown(payload: dict[str, Any]) -> str: f"- matched: `{summary.get('matched_risk_count')}`", f"- returned: `{summary.get('returned_risk_count')}`", f"- truncated: `{bool(summary.get('truncated'))}`", + *render_projection_envelope_markdown(payload.get("projection_envelope")), "", ( "No current accepted source proves a rollback candidate; " diff --git a/loopx/global_todos.py b/loopx/global_todos.py index 542087a458..c669144602 100644 --- a/loopx/global_todos.py +++ b/loopx/global_todos.py @@ -4,6 +4,11 @@ from pathlib import Path from typing import Any +from .control_plane.projection_envelope_facts import ( + render_projection_envelope_markdown, + seal_projection_envelope, + source_fact, +) from .control_plane.runtime.time import now_utc_iso from .control_plane.todos.decision_scope import todo_gate_relations from .control_plane.todos.user_gate import open_user_gate_todo_items @@ -58,7 +63,12 @@ def _request() -> dict[str, Any]: } -def build_global_todos_error(error: object) -> dict[str, Any]: +def build_global_todos_error( + error: object, + *, + status_payload: dict[str, Any] | None = None, + status_read_at: str | None = None, +) -> dict[str, Any]: return { "ok": False, "schema_version": SCHEMA_VERSION, @@ -68,6 +78,11 @@ def build_global_todos_error(error: object) -> dict[str, Any]: "omissions": [ "Raw/private failure details and local paths were intentionally omitted." ], + "projection_envelope": _todos_envelope( + status_payload or {}, status_read_at=status_read_at, + quota_read_at=None, quota_evaluated=0, quota_unavailable=0, + goal_scan_omitted=0, shown_count=0, available_count=0, + ), "boundary": public_safe_boundary(), } @@ -382,6 +397,78 @@ def _queue_goal_ids( return unique_goal_ids[:scan_limit], available_count +def _todos_envelope( + status_payload: dict[str, Any], + *, + status_read_at: str | None, + quota_read_at: str | None, + quota_evaluated: int, + quota_unavailable: int, + goal_scan_omitted: int, + shown_count: int, + available_count: int, +) -> dict[str, Any]: + global_registry = as_dict(status_payload.get("global_registry")) + status_envelope = as_dict(status_payload.get("projection_envelope")) + global_source = next( + (row for row in as_list(status_envelope.get("sources")) + if isinstance(row, dict) and row.get("source_id") == "global_registry"), + {}, + ) + available = global_registry.get("available") is True + count = global_registry.get("global_goal_count") + expected = count if available and isinstance(count, int) and not isinstance(count, bool) else None + excluded = int(global_registry.get("current_registry_excluded_goal_count") or 0) if available else 0 + omitted = [] + if excluded: + omitted.append({ + "reason": "outside_current_registry", "count": excluded, + "refs": [str(ref) for ref in as_list(global_registry.get("current_registry_excluded_goal_ids"))[:8]], + }) + if goal_scan_omitted: + omitted.append({"reason": "attention_goals_not_scanned", "count": goal_scan_omitted}) + if status_payload and status_payload.get("ok") is not True: + omitted.append({"reason": "status_unavailable", "count": 1}) + queue = status_payload.get("attention_queue") + queue_read = isinstance(queue, dict) and isinstance(queue.get("items"), list) + return seal_projection_envelope( + projection="global_todos", + observed_at=now_utc_iso(), + sources=[ + source_fact( + "global_registry", + read_status=( + str(global_source.get("read_status") or "read") if available + else str(global_registry.get("read_status") or ("missing" if status_read_at else "not_read")) + ), + last_read_at=global_source.get("last_read_at") or (status_read_at if available else None), + item_count=expected, + ), + source_fact( + "goal_quota", + read_status="read" if quota_read_at else "not_read", + last_read_at=quota_read_at, + item_count=quota_evaluated, + unreadable_count=quota_unavailable, + ), + source_fact( + "attention_queue", + read_status="read" if queue_read else "not_read", + last_read_at=status_read_at if queue_read else None, + item_count=len(queue["items"]) if queue_read else None, + ), + *([] if status_envelope else [source_fact("status", read_status="not_read")]), + ], + coverage={ + "scope": "global", "expected_count": expected, + "included_count": max(0, expected - excluded - goal_scan_omitted) if expected is not None else 0, + "omitted": omitted, + "shown_count": shown_count, "available_count": available_count, + }, + upstream=[status_envelope] if status_envelope else [], + ) + + def _groups(todos: list[dict[str, Any]]) -> dict[str, list[dict[str, Any]]]: return { "runnable": [todo for todo in todos if todo.get("readiness") == "runnable"], @@ -409,8 +496,12 @@ def build_global_todos( scan_roots=scan_roots, limit=scan_limit, ) + status_read_at = now_utc_iso() if status_payload.get("ok") is not True: - return build_global_todos_error("Global status source unavailable.") + return build_global_todos_error( + "Global status source unavailable.", + status_payload=status_payload, status_read_at=status_read_at, + ) goal_ids, available_goal_count = _queue_goal_ids( status_payload, @@ -418,6 +509,8 @@ def build_global_todos( ) warnings: list[dict[str, Any]] = [] classified: list[dict[str, Any]] = [] + quota_read_at = now_utc_iso() + quota_unavailable = 0 for goal_id in goal_ids: quota_payload = _quota_for_goal( status_payload, @@ -425,6 +518,7 @@ def build_global_todos( agent_id=agent_id, warnings=warnings, ) + quota_unavailable += int(quota_payload is None) if quota_payload is None or not _quota_matches_agent( quota_payload, agent_id=agent_id, @@ -466,12 +560,19 @@ def build_global_todos( "source_surfaces": SOURCE_SURFACES, "truncated": matched_count > returned_count, "goal_scan_limit": scan_limit, - "goal_scan_truncated": available_goal_count > scan_limit, + "goal_scan_truncated": available_goal_count > len(goal_ids), }, "groups": retained_groups, "todos": retained, "source_warnings": warnings[:SOURCE_WARNING_LIMIT], "source_warnings_truncated": warning_count > SOURCE_WARNING_LIMIT, + "projection_envelope": _todos_envelope( + status_payload, status_read_at=status_read_at, + quota_read_at=quota_read_at, quota_evaluated=len(goal_ids), + quota_unavailable=quota_unavailable, + goal_scan_omitted=max(0, available_goal_count - len(goal_ids)), + shown_count=returned_count, available_count=matched_count, + ), "omissions": [ ( "Raw logs, raw transcripts, connector payloads, credential values, " @@ -503,6 +604,7 @@ def render_global_todos_markdown(payload: dict[str, Any]) -> str: "", "- ok: `False`", f"- error: {_redact_text(payload.get('error'))}", + *render_projection_envelope_markdown(payload.get("projection_envelope")), ] omissions = [ _redact_text(item) @@ -524,6 +626,7 @@ def render_global_todos_markdown(payload: dict[str, Any]) -> str: f"- returned: `{summary.get('returned_todo_count')}`", f"- truncated: `{bool(summary.get('truncated'))}`", f"- goal_scan_truncated: `{bool(summary.get('goal_scan_truncated'))}`", + *render_projection_envelope_markdown(payload.get("projection_envelope")), "", "Review is an overlapping work-kind facet of the readiness groups.", ] From 7aa4d596a1adf83a896266529c09e57d677a9bd4 Mon Sep 17 00:00:00 2001 From: "liuhuadong.hans" Date: Mon, 5 Oct 2026 00:10:27 +0800 Subject: [PATCH 2/3] test(projection): cover global todo and risk envelopes Signed-off-by: liuhuadong.hans --- tests/test_projection_envelope.py | 142 +++++++++++++++++++++++++++++- 1 file changed, 140 insertions(+), 2 deletions(-) diff --git a/tests/test_projection_envelope.py b/tests/test_projection_envelope.py index 5f3754f6a0..5f930de3ce 100644 --- a/tests/test_projection_envelope.py +++ b/tests/test_projection_envelope.py @@ -5,6 +5,8 @@ from pathlib import Path import loopx.control_plane.projection_envelope_facts as projection_envelope +import loopx.global_risks as global_risks +import loopx.global_todos as global_todos import loopx.summary_all as summary_all from loopx.control_plane.runtime.status_projection_cache import ( load_status_projection_cache, @@ -268,6 +270,142 @@ def test_global_gates_counts_unavailable_quota_and_missing_status_envelope(tmp_p assert "🔴" in summary_all.render_global_gates_markdown(payload) +def test_global_todos_discloses_global_membership_and_display_truncation(tmp_path: Path, monkeypatch) -> None: + status_payload = _global_status_payload(tmp_path, excluded=3) + monkeypatch.setattr(global_todos, "collect_status", lambda **_: status_payload) + monkeypatch.setattr( + global_todos, + "build_quota_should_run", + lambda *_args, **_kwargs: { + "ok": True, + "normal_delivery_allowed": True, + "selected_todo": {"todo_id": "todo-a", "status": "open", "priority": "P1"}, + "agent_todo_summary": { + "deferred_resume_candidates": [ + {"todo_id": "todo-b", "status": "deferred", "priority": "P2"} + ] + }, + }, + ) + + payload = global_todos.build_global_todos( + registry_path=tmp_path / "registry.json", runtime_root_override=None, + scan_roots=[], agent_id=None, limit=1, + ) + + envelope = payload["projection_envelope"] + assert envelope["projection"] == "global_todos" + assert (envelope["coverage"]["expected_count"], envelope["coverage"]["included_count"]) == (4, 1) + assert envelope["coverage"]["omitted"][0]["reason"] == "outside_current_registry" + assert (envelope["coverage"]["shown_count"], envelope["coverage"]["available_count"]) == (1, 2) + assert envelope["coverage"]["truncated"] is True + assert "incomplete_coverage" in envelope["alert_reasons"] + assert envelope["upstream"][0]["projection"] == "status" + assert "status/registry" in _sources(envelope) + assert _sources(envelope)["goal_quota"]["item_count"] == 1 + assert "🔴 projection alerts" in global_todos.render_global_todos_markdown(payload) + assert str(tmp_path) not in json.dumps(envelope) + + +def test_global_todos_missing_status_envelope_and_failed_quota_alert(tmp_path: Path, monkeypatch) -> None: + status_payload = _global_status_payload(tmp_path, excluded=0) + status_payload.pop("projection_envelope") + monkeypatch.setattr(global_todos, "collect_status", lambda **_: status_payload) + monkeypatch.setattr( + global_todos, "build_quota_should_run", + lambda *_args, **_kwargs: (_ for _ in ()).throw(RuntimeError("quota unavailable")), + ) + + payload = global_todos.build_global_todos( + registry_path=tmp_path / "registry.json", runtime_root_override=None, + scan_roots=[], agent_id=None, limit=5, + ) + + envelope = payload["projection_envelope"] + assert _sources(envelope)["status"]["status"] == "not_read" + assert _sources(envelope)["goal_quota"]["unreadable_count"] == 1 + assert envelope["alert_reasons"] == ["missing_required_sources", "unreadable_sources"] + + +def test_global_todos_missing_attention_queue_alerts(tmp_path: Path, monkeypatch) -> None: + status_payload = _global_status_payload(tmp_path, excluded=0) + status_payload.pop("attention_queue") + monkeypatch.setattr(global_todos, "collect_status", lambda **_: status_payload) + + envelope = global_todos.build_global_todos( + registry_path=tmp_path / "registry.json", runtime_root_override=None, + scan_roots=[], agent_id=None, limit=5, + )["projection_envelope"] + + assert _sources(envelope)["attention_queue"]["status"] == "not_read" + assert "missing_required_sources" in envelope["alert_reasons"] + + +def test_global_risks_inherits_status_and_discloses_missing_global_membership(tmp_path: Path, monkeypatch) -> None: + status_payload = _global_status_payload(tmp_path, excluded=0) + status_payload.update({"contract": {"error_diagnostics": []}, "run_history": {"goals": []}}) + monkeypatch.setattr(global_risks, "collect_status", lambda **_: status_payload) + + payload = global_risks.build_global_risks( + registry_path=tmp_path / ".loopx" / "registry.json", runtime_root_override=None, + scan_roots=[], agent_id=None, time_range="24h", limit=5, + ) + + envelope = payload["projection_envelope"] + assert envelope["projection"] == "global_risks" + assert envelope["complete"] is True and envelope["alert"] is False + assert envelope["upstream"][0]["projection"] == "status" + assert "status/registry" in _sources(envelope) + assert _sources(envelope)["status_contract"]["item_count"] == 0 + + status_payload["global_registry"] = {"available": False, "findings": []} + missing = global_risks.build_global_risks( + registry_path=tmp_path / ".loopx" / "registry.json", runtime_root_override=None, + scan_roots=[], agent_id=None, time_range="24h", limit=5, + ) + envelope = missing["projection_envelope"] + assert envelope["coverage"]["expected_count"] is None + assert envelope["complete"] is False + assert "missing_required_sources" in envelope["alert_reasons"] + assert "incomplete_coverage" in envelope["alert_reasons"] + assert "🔴 projection alerts" in global_risks.render_global_risks_markdown(missing) + assert str(tmp_path) not in json.dumps(envelope) + + +def test_global_risks_scan_truncation_is_an_incomplete_coverage_alert(tmp_path: Path, monkeypatch) -> None: + status_payload = _global_status_payload(tmp_path, excluded=0) + status_payload["contract"] = { + "error_diagnostics": [ + {"code": "test_failure", "scope": "global"} for _ in range(41) + ] + } + monkeypatch.setattr(global_risks, "collect_status", lambda **_: status_payload) + + payload = global_risks.build_global_risks( + registry_path=tmp_path / ".loopx" / "registry.json", runtime_root_override=None, + scan_roots=[], agent_id=None, time_range="24h", limit=1, + ) + + envelope = payload["projection_envelope"] + assert payload["summary"]["source_rows_truncated"] is True + assert envelope["coverage"]["omitted"][-1]["reason"] == "source_rows_not_scanned" + assert envelope["coverage"]["truncated"] is True + assert "incomplete_coverage" in envelope["alert_reasons"] + + +def test_global_risks_empty_contract_container_is_a_read_source(tmp_path: Path, monkeypatch) -> None: + status_payload = _global_status_payload(tmp_path, excluded=0) + status_payload["contract"] = {} + monkeypatch.setattr(global_risks, "collect_status", lambda **_: status_payload) + + envelope = global_risks.build_global_risks( + registry_path=tmp_path / ".loopx" / "registry.json", runtime_root_override=None, + scan_roots=[], agent_id=None, time_range="24h", limit=5, + )["projection_envelope"] + + assert _sources(envelope)["status_contract"]["status"] == "fresh" + + def test_real_cli_global_membership_failure_and_recovery(tmp_path): """Global scope must never use local membership as its denominator.""" import subprocess @@ -298,7 +436,7 @@ def cli(command, registry_path=registry): global_path.unlink(missing_ok=True) else: global_path.write_text(content) - for command in ("global-summary", "global-gates"): + for command in ("global-summary", "global-gates", "global-todos", "global-risks"): envelope = cli(command)["projection_envelope"] assert envelope["coverage"]["expected_count"] is None assert envelope["complete"] is False and envelope["alert"] is True @@ -308,7 +446,7 @@ def cli(command, registry_path=registry): local = cli("status")["projection_envelope"] assert local["complete"] is True and local["alert"] is False global_path.write_text(json.dumps(global_registry)) - for command in ("global-summary", "global-gates"): + for command in ("global-summary", "global-gates", "global-todos", "global-risks"): local = cli(command)["projection_envelope"] assert (local["coverage"]["included_count"], local["coverage"]["expected_count"]) == (1, 2) assert local["complete"] is False From a0611332053b9c23c5a5fa628c0b55646f26cd5e Mon Sep 17 00:00:00 2001 From: "liuhuadong.hans" Date: Mon, 5 Oct 2026 00:10:33 +0800 Subject: [PATCH 3/3] docs(projection): list global todo and risk carriers Signed-off-by: liuhuadong.hans --- .../rfcs/typescript-control-plane-migration-v0.md | 14 +++++++------- .../contracts/projection-envelope-contract.md | 2 ++ 2 files changed, 9 insertions(+), 7 deletions(-) diff --git a/docs/architecture/rfcs/typescript-control-plane-migration-v0.md b/docs/architecture/rfcs/typescript-control-plane-migration-v0.md index 8fa226d176..88e2bf5194 100644 --- a/docs/architecture/rfcs/typescript-control-plane-migration-v0.md +++ b/docs/architecture/rfcs/typescript-control-plane-migration-v0.md @@ -1453,12 +1453,12 @@ facade-exit condition in §4), TS gathers the facts directly and the adapter is deleted. Rollout. `status` (including `--goal-id` and projection-cache hits), -`global-summary` and `global-gates` now carry the envelope. Every other -`collect_status` caller receives the status envelope in its payload but does -not yet emit its own. Next, in order: `global-todos` and `global-risks` (the -same composition, one call each), `quota should-run`, `review-packet`, and -Decision Context packets. A read model added to or migrated into TypeScript -emits the envelope in the same PR; §6 makes this a promotion gate. +`global-summary`, `global-gates`, `global-todos` and `global-risks` now carry +the envelope. Every other `collect_status` caller receives the status envelope +in its payload but does not yet emit its own. Next, in order: `quota +should-run`, `review-packet`, and Decision Context packets. A read model added +to or migrated into TypeScript emits the envelope in the same PR; §6 makes +this a promotion gate. Consumers treat a missing envelope as unknown freshness, and disclose an alerting one before stating any conclusion that depends on it. Field @@ -1530,7 +1530,7 @@ choice is now implemented rather than hypothetical. | Quota monitor-poll commit transaction | TypeScript owns monitor admission revalidation, target/event/result construction, effect replay/index CAS, provider intent, and repairable JSON/Markdown/index persistence | Python projects compact `should-run` facts, invokes the real Todo provider between at most two reductions, reloads legacy status, and holds the cross-writer index lock | | Runtime decoders ([#3443](https://github.com/loopx-project/loopx/pull/3443)) | Stable primitive decoding has one small shared module; domain decoders remain local | No larger schema framework is justified | | Transaction payoff ([#3464](https://github.com/loopx-project/loopx/pull/3464), [#3481](https://github.com/loopx-project/loopx/pull/3481), and Todo completion) | Turn settlement, quota delivery routing, and Todo completion each cross one coarse TS boundary; the Todo transaction owns identity, replay fencing, validation planning/result reduction, continuation/recovery, and completion metadata | Python still executes explicitly external providers and materializes legacy Markdown results; other domains still need their own bounded cutovers | -| Projection envelope | TypeScript owns decoding of `loopx_projection_envelope_v0` and every freshness, alert, completeness and replay decision | Python gathers read facts for `status`, `global-summary` and `global-gates` until those projections migrate | +| Projection envelope | TypeScript owns decoding of `loopx_projection_envelope_v0` and every freshness, alert, completeness and replay decision | Python gathers read facts for `status`, `global-summary`, `global-gates`, `global-todos` and `global-risks` until those projections migrate | | Promoted-authority Todo claim | TypeScript owns the provider-head read, lifecycle validation, complete-record update, hard-lease check, CAS, receipt, and readback-safe result for claims after authority promotion | Default local Markdown mode remains on the legacy writer; other Todo mutations and Markdown regeneration remain bounded follow-ups | The scheduler facade exit now includes its first bounded Stage 3 route. A diff --git a/docs/reference/contracts/projection-envelope-contract.md b/docs/reference/contracts/projection-envelope-contract.md index ce6fd7fa80..80799f6848 100644 --- a/docs/reference/contracts/projection-envelope-contract.md +++ b/docs/reference/contracts/projection-envelope-contract.md @@ -21,6 +21,8 @@ Python-owned projections do so through `loopx/control_plane/projection_envelope_ | `loopx status --use-projection-cache` hit | `status`, `served_from_cache: true` | as stored | | `loopx global-summary` | `global_summary` | `global` | | `loopx global-gates` | `global_gates` | `global` | +| `loopx global-todos` | `global_todos` | `global` | +| `loopx global-risks` | `global_risks` | `global` | Other read models adopt the envelope in the order listed in the RFC; until then they carry no freshness guarantee and consumer rule 1 below applies.