From c755aa0b421a2639ab9b1df16da553a1a9c4377e Mon Sep 17 00:00:00 2001 From: "lusendong.6789" Date: Wed, 4 Nov 2026 10:55:55 +0800 Subject: [PATCH] refactor(chat): move public status owner behind identity-compatible alias Signed-off-by: lusendong.6789 Co-authored-by: TRAE CLI --- .../rfcs/monorepo-distribution-split-v0.md | 29 ++ .../monorepo-distribution-split-v0.zh-CN.md | 22 + loopx/chat/status_api.py | 378 +++++++++++++++++ loopx/chat/todo_detail.py | 2 +- loopx/chat_server.py | 2 +- loopx/chat_status_api.py | 390 +----------------- .../test_source_session_registry_denial.py | 4 +- tests/test_chat_internal_module_layout.py | 7 + .../test_chat_status_public_compatibility.py | 127 ++++-- 9 files changed, 555 insertions(+), 406 deletions(-) create mode 100644 loopx/chat/status_api.py diff --git a/docs/architecture/rfcs/monorepo-distribution-split-v0.md b/docs/architecture/rfcs/monorepo-distribution-split-v0.md index 2aad7cff1b..9f8a59de33 100644 --- a/docs/architecture/rfcs/monorepo-distribution-split-v0.md +++ b/docs/architecture/rfcs/monorepo-distribution-split-v0.md @@ -786,6 +786,35 @@ Approval needs a reviewed decision, not implementation progress or silence. exact expiry remains a release-owner decision. M1 is open. No module-count, loading-performance, split-package or R1 managed-Turn benefit is claimed. +### 2026-11-04 — M1 public status owner and compatibility bridge + +- **Delta:** the complete `ChatStatusRequestMixin` and `goal_read_runtime_root` + implementation moves to `loopx.chat.status_api`. Chat HTTP and Todo detail + import that owner directly. The characterization prerequisite + [#8001](https://github.com/loopx-project/loopx/pull/8001) is already merged. +- **Public compatibility:** `loopx.chat_status_api` is a deprecated module-identity + alias, not copied exports or a second implementation. Historical patch points, + imports and pickle lookups reach the canonical module; reload retains live + globals for already-imported functions. The legacy import warns once per + process, while the canonical import does not warn. Retain the alias for at least + one minor release after the move; the release owner still decides D3's expiry. +- **Authority boundary:** retarget both exact owner-path entries in + `tests/architecture/test_source_session_registry_denial.py`, leaving every + denial assertion and the `_discovery_registry` exception unchanged. Real HTTP + coverage permits lifecycle-only registry discovery but refuses full status + before collector execution. Moving the presentation owner grants no new + runtime, source, provider or execution authority. +- **Evidence:** the public compatibility suite now includes canonical-first + import, module identity, warning/reload and both pickle paths. Existing HTTP, + File/SQLite source routing, source loss/recovery and delivery-review tests + protect the original behavior. Absolute-import-resolved AST comparison covers + the complete owner; the denial oracle differs only in its two owner paths. +- **Remaining:** this is one public-owner M1 slice, not completion of the wider + public Chat migration, D3, split distributions or R1 managed minimal. The alias + means no net top-level module or budget reduction; no loading-performance or + long-horizon outcome benefit is claimed. Source and isolated-wheel qualification + do not establish an upgrade of an existing installed host. + ## Appendix B: Decision log | Date | Decision | Owner / approval | Alternatives | Normative sections changed | diff --git a/docs/architecture/rfcs/monorepo-distribution-split-v0.zh-CN.md b/docs/architecture/rfcs/monorepo-distribution-split-v0.zh-CN.md index a4793dc90f..2e4d3dcf9d 100644 --- a/docs/architecture/rfcs/monorepo-distribution-split-v0.zh-CN.md +++ b/docs/architecture/rfcs/monorepo-distribution-split-v0.zh-CN.md @@ -570,6 +570,28 @@ R0 可以先于目录移动交付，但不代表内核合格。R3/R4 可选， 版本，D3 精确退役时间仍由发布负责人决定。M1 未关闭，不声明模块数量、加载性能、 拆包或 R1 managed Turn 收益。 +### 2026-11-04 — M1 公开 status owner 与兼容 bridge + +- **变化：** 完整 `ChatStatusRequestMixin` 与 `goal_read_runtime_root` 实现迁入 + `loopx.chat.status_api`，Chat HTTP 与 Todo detail 直接导入该 owner。刻画前置 + [#8001](https://github.com/loopx-project/loopx/pull/8001) 已合入。 +- **公开兼容：** `loopx.chat_status_api` 是弃用的模块身份 alias，而非复制导出或第二 + 实现。历史 patch 点、import 和 pickle 查找到达同一 canonical 模块；reload 后 + 已导入函数仍使用活跃 globals。旧 import 每进程仅警告一次，新 import 不警告。 + 移动后至少保留一个 minor 版本，D3 精确退役期限仍由发布负责人决定。 +- **权限边界：** 同步重定向 + `tests/architecture/test_source_session_registry_denial.py` 的两个精确 owner + 路径项，全部拒绝断言及 `_discovery_registry` 例外不变。真实 HTTP 覆盖允许仅生命周期 + registry 的发现，但在 collector 执行前拒绝完整 status。移动 presentation owner + 不授予新的 runtime、来源、provider 或执行权威。 +- **证据：** 公开兼容测试新增 canonical-first import、模块身份、warning／reload + 及两种 pickle 路径。既有 HTTP、File／SQLite 源路由、源丢失／恢复和 delivery review + 测试保护原行为。解析为绝对 import 后的 AST 对比覆盖完整 owner；拒绝 oracle + 仅有两个 owner 路径差异。 +- **剩余：** 这是 M1 的一个公开 owner 片，不关闭更广的公开 Chat 迁移、D3、发行物 + 拆分或 R1 managed minimal。保留 alias，因此顶层模块数量及预算无净减少；不声明 + 加载性能或长程成果收益。源码与隔离 wheel 验证不等于升级既有已安装宿主。 + ## 附录 B：决策日志 | 日期 | 决策 | 负责人 / 批准 | 备选 | 变更的规范章节 | diff --git a/loopx/chat/status_api.py b/loopx/chat/status_api.py new file mode 100644 index 0000000000..86e03e339b --- /dev/null +++ b/loopx/chat/status_api.py @@ -0,0 +1,378 @@ +"""Loopback status route owned by the Chat HTTP surface.""" + +from __future__ import annotations + +import errno +import json +import subprocess +from pathlib import Path +from typing import Any +from urllib.parse import parse_qs, urlparse + +from ..chat import redact_local_paths +from .goal_subagent_api import goal_subagent_configuration_enabled +from .workspace_directory import workspace_goal_directory +from ..codex_app_thread_activity import ( + codex_delivery_expectations, + codex_thread_observers, +) +from ..control_plane.agents.host_thread_activity import ( + attach_host_delivery_windows, + attach_host_thread_activity, +) +from ..control_plane.effect_runtime import ( + EffectRuntimePermanentIOError, + EffectRuntimeRemoteError, + EffectRuntimeStartupError, + effect_runtime_request_scope, +) +from ..control_plane.coordination.local_authority import LocalCoordinationAuthorityUnavailable +from ..control_plane.runtime.public_safety import ( + public_safe_compact_text, + validate_public_safe_value, +) +from ..control_plane.todos.contract import ( + TODO_STATUS_OPEN, + TODO_TASK_CLASS_MONITOR, + normalize_todo_status, + todo_done_for_status, +) +from ..control_plane.work_items.task_graph import build_goal_task_map +from ..control_plane.projection_envelope_facts import seal_projection_envelope, source_fact +from ..control_plane.runtime.time import now_utc_iso +from ..feedback import validate_goal_id +from ..history import load_registry +from ..control_plane.projects.registry_codec import load_project_registry +from ..status import collect_status +from ..status_server import ( + DEFAULT_EXTENSION_PRESENTATION_SURFACES_PATH, + DEFAULT_EXTENSION_PROJECTION_PATH, + ExtensionPresentationRequestMixin, + parse_goal_activation_filter, +) +from ..control_plane.todos.list_readback import list_goal_todos + + +def goal_read_runtime_root(server: Any) -> str | None: + """Keep registry source routing; Chat storage is only an unbound fallback.""" + registry = _discovery_registry(server.registry_path) + if registry.get("common_runtime_root"): + # Passing the coordinator's resolved root as an explicit override + # suppresses the collector's per-Goal source_registry routing. A + # global directory may contain Goals owned by different runtimes. + return None + controller = getattr(server, "runtime_controller", None) + root = getattr(controller, "coordination_runtime_root", None) + return str(root) if root is not None else server.runtime_root_override + + +def _goal_task_map(server: Any, goal_id: str) -> dict[str, Any] | None: + runtime_root = goal_read_runtime_root(server) + source_read_at = now_utc_iso() + try: + listed = list_goal_todos( + registry_path=server.registry_path, + goal_id=goal_id, + runtime_root_arg=str(runtime_root) if runtime_root else None, + ) + except (OSError, ValueError): + # An unreadable Todo source leaves the rest of the review usable. + return None + todos = [row for row in listed.get("todos") or [] if isinstance(row, dict)] + return build_goal_task_map( + goal_id=goal_id, + todos=todos, + source_truncated=int(listed.get("todo_count") or 0) > len(todos), + public_safe_compact_text=public_safe_compact_text, + normalize_todo_status=normalize_todo_status, + todo_done_for_status=todo_done_for_status, + todo_status_open=TODO_STATUS_OPEN, + monitor_task_class=TODO_TASK_CLASS_MONITOR, + source_read_at=source_read_at, + ) + + +def _discovery_registry(registry_path: Path) -> dict[str, Any]: + """Read the registry for the workspace directory; a missing file is an empty workspace. + + A fresh install has no registry until the first Goal is created. Discovery + reports that as zero registered Goals rather than a projection failure, so + the home page can show its first-Goal state. Unreadable or incompatible + registries still raise. + """ + + try: + return load_project_registry(registry_path) + except FileNotFoundError: + return {} + + +def _discovery_contract(registry_path: Path) -> dict[str, Any]: + """The source facts a discovery read can state without opening runtime state. + + A fresh install has no registry until the first Goal is created. The full + status reports that as a warning (allow_missing_registry); the directory + states the same fact in the same words, so the first-Goal home can show why + the workspace is empty. The caller redacts paths like every status response. + """ + + warnings: list[str] = [] + if not registry_path.exists(): + warnings.append(f"registry file does not exist: {registry_path}") + return { + "ok": True, + "summary": {"errors": 0, "warnings": len(warnings), "checks": 0}, + "errors": [], + "warnings": warnings, + "checks": [], + } + + +def _status_access_denied(error: BaseException) -> bool: + """Recognize permission causes without reclassifying unrelated I/O failures.""" + current: BaseException | None = error + seen: set[int] = set() + for _ in range(8): + if current is None or id(current) in seen: + break + seen.add(id(current)) + if isinstance(current, EffectRuntimeRemoteError): + return ( + isinstance(current, EffectRuntimePermanentIOError) + and current.diagnostic_code == "io_permission_denied" + ) + if isinstance(current, OSError): + winerror = getattr(current, "winerror", None) + if winerror is not None: + # Windows also maps sharing/lock violations to PermissionError. + return type(winerror) is int and winerror in {5, 65} + if isinstance(current, PermissionError) or current.errno in { + errno.EACCES, errno.EPERM, + }: + return True + if current.__cause__ is not None: + current = current.__cause__ + elif ( + isinstance(current, EffectRuntimeStartupError) + and current.diagnostic_code in { + "runtime_launch_failed", "runtime_request_failed", + } + and not current.__suppress_context__ + ): + current = current.__context__ + else: + break + return False + + +class ChatStatusRequestMixin(ExtensionPresentationRequestMixin): + """Serve the full dashboard status contract through the Chat origin.""" + + server: Any + + def _send_error(self, message: str, **kwargs: Any) -> None: + raise NotImplementedError + + def _send_json(self, payload: dict[str, Any], *, status: int = 200) -> None: + raise NotImplementedError + + def _delivery_review(self) -> None: + with effect_runtime_request_scope(): + self._status(delivery_review=True) + + def _status(self, *, delivery_review: bool = False) -> None: + query = parse_qs(urlparse(self.path).query, keep_blank_values=True) + query_error_code = "invalid_goal_activation" + try: + activation_state_filter = parse_goal_activation_filter(query) + query_error_code = "invalid_workspace_query" + views = query.get("view") + if views is not None and (delivery_review or views != ["workspace-directory"]): + raise ValueError("unsupported workspace status view") + if delivery_review and not self._require_loopback_origin(): + return + requested_goals = query.get("goal_id") + goal_id = self.server.selected_goal_id + if requested_goals is not None: + if len(requested_goals) != 1 or not requested_goals[0].strip(): + raise ValueError("one nonempty goal_id is required") + requested_goal = requested_goals[0] + validate_goal_id(requested_goal) + if goal_id and requested_goal != goal_id: + raise ValueError("goal_id is outside this workspace") + goal_id = requested_goal + if delivery_review and not goal_id: + raise ValueError("delivery review requires one goal_id") + except ValueError as exc: + self._send_error( + str(exc), + status=400, + error_code=query_error_code, + ) + return + try: + status_runtime_root = goal_read_runtime_root(self.server) + registry = {} + if views is not None or requested_goals is not None or delivery_review: + # Discovery reads only metadata. Full status/runtime reads keep + # their existing lifecycle-only profile refusal. + registry = (_discovery_registry(self.server.registry_path) if views == ["workspace-directory"] + else load_registry(self.server.registry_path)) + directory = workspace_goal_directory(registry, selected_goal_id=goal_id) + if goal_id and not directory["goals"]: + self._send_error("Goal is no longer registered.", status=404) + return + if views == ["workspace-directory"]: + directory["contract"] = _discovery_contract(self.server.registry_path) + # Discovery stays metadata-only: redact with the roots this + # server knows, without requiring the full runtime wiring. + runtime_root = getattr(self.server, "runtime_root", None) + directory = json.loads( + redact_local_paths( + json.dumps(directory, ensure_ascii=False), + protected_paths=[ + self.server.registry_path, + *(getattr(self.server, "scan_roots", None) or []), + *([runtime_root] if runtime_root else []), + ], + ) + ) + self._send_json(directory) + return + linked = None + if goal_id and registry.get("goal_source_links"): + from ..control_plane.status.goal_source import SourceScopeError + from ..presentation.goal_source_routing import goal_source_catalog + catalog = goal_source_catalog(self.server) + if delivery_review and catalog.resolve(goal_id, registry): + self._send_error("Execution review belongs to the original Goal runtime.", status=403) + return + try: + linked = catalog.status(goal_id, registry) + except SourceScopeError: + self._send_error("The original Goal is no longer available in this source.", + status=409) + return + except (OSError, subprocess.SubprocessError): + self._send_error("The original Goal source is unavailable; its link is preserved.", + status=503, error_code="goal_source_unavailable") + return + projection = linked if linked is not None else collect_status( + registry_path=self.server.registry_path, + # The Chat service already starts on an absent registry as an + # empty workspace (serve_chat); its status reports the same + # fact as a warning, so a fresh install is not a failed source. + allow_missing_registry=True, + runtime_root_override=status_runtime_root, + limit=self.server.limit, + goal_id=goal_id, + include_zcode_goal_eligibility=not delivery_review, + include_task_graph=delivery_review, + activation_state_filter=activation_state_filter, + include_goal_subagent_configuration=( + goal_subagent_configuration_enabled(self.server) + ), + ) + if requested_goals is not None or delivery_review: + latest = workspace_goal_directory( + load_registry(self.server.registry_path) + ) + if latest["registry_revision"] != directory["registry_revision"]: + self._send_error( + "Goal directory changed; refresh the workspace.", status=409 + ) + return + projection["workspace_registry_revision"] = directory[ + "registry_revision" + ] + if not delivery_review: + projection["local_dashboard_api"] = { + **(projection.get("local_dashboard_api") or {}), + "source": "chat", + "status_url": "/status.json", + "presentation_surfaces_url": DEFAULT_EXTENSION_PRESENTATION_SURFACES_PATH, + "presentation_detail_url": DEFAULT_EXTENSION_PROJECTION_PATH, + } + if linked is None: + attach_host_thread_activity( + projection, observers=codex_thread_observers( + cache_root=self.server.runtime_root / "observation-cache" / "codex-activity", + ) + ) + attach_host_delivery_windows( + projection, + expectations=codex_delivery_expectations, + ) + if delivery_review: + if projection.get("ok") is not True: + self._send_error("Delivery review sources are unavailable.", status=503) + return + item = next((row for row in projection.get("attention_queue", {}).get("items", []) + if row.get("goal_id") == goal_id), {}) + goal = next((row for row in projection.get("run_history", {}).get("goals", []) + if row.get("id") == goal_id), {}) + goal_map = _goal_task_map(self.server, goal_id) + observed_at = now_utc_iso() + components = [item.get("task_graph_projection"), goal_map, goal.get("acceptance_observation")] + upstream = [value for value in [projection.get("projection_envelope"), + goal_map.get("projection_envelope") if goal_map else None] if isinstance(value, dict)] + sources = [] + if not isinstance(projection.get("projection_envelope"), dict): + sources.append(source_fact("status", read_status="not_read")) + if goal_map is None: + sources.append(source_fact("goal_todos", read_status="unreadable")) + missing = sum(value is None for value in components) + envelope = seal_projection_envelope( + projection="delivery_review", observed_at=observed_at, + sources=sources, upstream=upstream, + coverage={"scope": "selected_goal_delivery", "expected_count": 3, + "included_count": 3 - missing, + "omitted": [{"reason": "unavailable_components", "count": missing}] if missing else []}, + ) + projection = { + "ok": True, + "goal_id": goal_id, + "observed_at": observed_at, + "projection_envelope": envelope, + "graph": item.get("task_graph_projection"), + "goal_map": goal_map, + "acceptance": goal.get("acceptance_observation"), + } + protected_paths = [ + self.server.registry_path, + *self.server.scan_roots, + self.server.runtime_root, + *([status_runtime_root] if status_runtime_root else []), + ] + projection = json.loads( + redact_local_paths( + json.dumps(projection, ensure_ascii=False), + protected_paths=protected_paths, + ) + ) + if delivery_review: + # Validate the outbound view, after removing owner-local paths. + validate_public_safe_value(projection) + except Exception as exc: # noqa: BLE001 - do not expose local projection failures. + # The storage owner already exposes inspection and reviewed import. + # Preserve its typed refusal, not private paths or migration payloads. + if isinstance(exc, LocalCoordinationAuthorityUnavailable) and exc.code == "canonical_authority_required": + self._send_error( + "This Goal needs a storage upgrade. Open its storage settings to review recovery.", + status=503, + error_code=exc.code, + ) + return + self._send_error( + "LoopX status could not be projected for the workspace.", + status=500, + error_code=( + "workspace_status_access_denied" + if requested_goals is not None + and views is None + and _status_access_denied(exc) + else None + ), + ) + return + self._send_json(projection) diff --git a/loopx/chat/todo_detail.py b/loopx/chat/todo_detail.py index a84706f34d..08def79434 100644 --- a/loopx/chat/todo_detail.py +++ b/loopx/chat/todo_detail.py @@ -3,7 +3,7 @@ import hashlib from urllib.parse import parse_qs, urlparse -from ..chat_status_api import goal_read_runtime_root +from .status_api import goal_read_runtime_root from ..status_server import is_loopback_host from ..control_plane.todos.list_readback import list_goal_todos diff --git a/loopx/chat_server.py b/loopx/chat_server.py index 1166fe1a33..12de5187fe 100644 --- a/loopx/chat_server.py +++ b/loopx/chat_server.py @@ -42,7 +42,7 @@ add_goal_subagent_routes, ) from .capabilities.native_chat.model_settings_api import ChatModelSettingsRequestMixin -from .chat_status_api import ChatStatusRequestMixin +from .chat.status_api import ChatStatusRequestMixin from .chat_runtime import ( STEERING_NOT_DELIVERED_CODES, ChatRuntimeController, diff --git a/loopx/chat_status_api.py b/loopx/chat_status_api.py index 0e1bbb96bc..43265db796 100644 --- a/loopx/chat_status_api.py +++ b/loopx/chat_status_api.py @@ -1,378 +1,22 @@ -"""Loopback status route owned by the Chat HTTP surface.""" +"""Deprecated public alias for :mod:`loopx.chat.status_api`. -from __future__ import annotations +Keep historical imports and their module-global patch points for at least one +minor release after the M1 move. The release owner decides the exact expiry. +""" -import errno -import json -import subprocess -from pathlib import Path -from typing import Any -from urllib.parse import parse_qs, urlparse +import sys +import warnings -from .chat import redact_local_paths -from .chat.goal_subagent_api import goal_subagent_configuration_enabled -from .chat.workspace_directory import workspace_goal_directory -from .codex_app_thread_activity import ( - codex_delivery_expectations, - codex_thread_observers, -) -from .control_plane.agents.host_thread_activity import ( - attach_host_delivery_windows, - attach_host_thread_activity, -) -from .control_plane.effect_runtime import ( - EffectRuntimePermanentIOError, - EffectRuntimeRemoteError, - EffectRuntimeStartupError, - effect_runtime_request_scope, -) -from .control_plane.coordination.local_authority import LocalCoordinationAuthorityUnavailable -from .control_plane.runtime.public_safety import ( - public_safe_compact_text, - validate_public_safe_value, -) -from .control_plane.todos.contract import ( - TODO_STATUS_OPEN, - TODO_TASK_CLASS_MONITOR, - normalize_todo_status, - todo_done_for_status, -) -from .control_plane.work_items.task_graph import build_goal_task_map -from .control_plane.projection_envelope_facts import seal_projection_envelope, source_fact -from .control_plane.runtime.time import now_utc_iso -from .feedback import validate_goal_id -from .history import load_registry -from .control_plane.projects.registry_codec import load_project_registry -from .status import collect_status -from .status_server import ( - DEFAULT_EXTENSION_PRESENTATION_SURFACES_PATH, - DEFAULT_EXTENSION_PROJECTION_PATH, - ExtensionPresentationRequestMixin, - parse_goal_activation_filter, -) -from .control_plane.todos.list_readback import list_goal_todos +from .chat import status_api as _implementation - -def goal_read_runtime_root(server: Any) -> str | None: - """Keep registry source routing; Chat storage is only an unbound fallback.""" - registry = _discovery_registry(server.registry_path) - if registry.get("common_runtime_root"): - # Passing the coordinator's resolved root as an explicit override - # suppresses the collector's per-Goal source_registry routing. A - # global directory may contain Goals owned by different runtimes. - return None - controller = getattr(server, "runtime_controller", None) - root = getattr(controller, "coordination_runtime_root", None) - return str(root) if root is not None else server.runtime_root_override - - -def _goal_task_map(server: Any, goal_id: str) -> dict[str, Any] | None: - runtime_root = goal_read_runtime_root(server) - source_read_at = now_utc_iso() - try: - listed = list_goal_todos( - registry_path=server.registry_path, - goal_id=goal_id, - runtime_root_arg=str(runtime_root) if runtime_root else None, - ) - except (OSError, ValueError): - # An unreadable Todo source leaves the rest of the review usable. - return None - todos = [row for row in listed.get("todos") or [] if isinstance(row, dict)] - return build_goal_task_map( - goal_id=goal_id, - todos=todos, - source_truncated=int(listed.get("todo_count") or 0) > len(todos), - public_safe_compact_text=public_safe_compact_text, - normalize_todo_status=normalize_todo_status, - todo_done_for_status=todo_done_for_status, - todo_status_open=TODO_STATUS_OPEN, - monitor_task_class=TODO_TASK_CLASS_MONITOR, - source_read_at=source_read_at, +# Copying exports would leave functions bound to different globals. Both import +# paths must resolve to the same module, including after reload or pickle lookup. +if not getattr(_implementation, "_legacy_status_import_warned", False): + warnings.warn( + "loopx.chat_status_api is deprecated; use loopx.chat.status_api. " + "The legacy import remains available for at least one minor release.", + DeprecationWarning, + stacklevel=2, ) - - -def _discovery_registry(registry_path: Path) -> dict[str, Any]: - """Read the registry for the workspace directory; a missing file is an empty workspace. - - A fresh install has no registry until the first Goal is created. Discovery - reports that as zero registered Goals rather than a projection failure, so - the home page can show its first-Goal state. Unreadable or incompatible - registries still raise. - """ - - try: - return load_project_registry(registry_path) - except FileNotFoundError: - return {} - - -def _discovery_contract(registry_path: Path) -> dict[str, Any]: - """The source facts a discovery read can state without opening runtime state. - - A fresh install has no registry until the first Goal is created. The full - status reports that as a warning (allow_missing_registry); the directory - states the same fact in the same words, so the first-Goal home can show why - the workspace is empty. The caller redacts paths like every status response. - """ - - warnings: list[str] = [] - if not registry_path.exists(): - warnings.append(f"registry file does not exist: {registry_path}") - return { - "ok": True, - "summary": {"errors": 0, "warnings": len(warnings), "checks": 0}, - "errors": [], - "warnings": warnings, - "checks": [], - } - - -def _status_access_denied(error: BaseException) -> bool: - """Recognize permission causes without reclassifying unrelated I/O failures.""" - current: BaseException | None = error - seen: set[int] = set() - for _ in range(8): - if current is None or id(current) in seen: - break - seen.add(id(current)) - if isinstance(current, EffectRuntimeRemoteError): - return ( - isinstance(current, EffectRuntimePermanentIOError) - and current.diagnostic_code == "io_permission_denied" - ) - if isinstance(current, OSError): - winerror = getattr(current, "winerror", None) - if winerror is not None: - # Windows also maps sharing/lock violations to PermissionError. - return type(winerror) is int and winerror in {5, 65} - if isinstance(current, PermissionError) or current.errno in { - errno.EACCES, errno.EPERM, - }: - return True - if current.__cause__ is not None: - current = current.__cause__ - elif ( - isinstance(current, EffectRuntimeStartupError) - and current.diagnostic_code in { - "runtime_launch_failed", "runtime_request_failed", - } - and not current.__suppress_context__ - ): - current = current.__context__ - else: - break - return False - - -class ChatStatusRequestMixin(ExtensionPresentationRequestMixin): - """Serve the full dashboard status contract through the Chat origin.""" - - server: Any - - def _send_error(self, message: str, **kwargs: Any) -> None: - raise NotImplementedError - - def _send_json(self, payload: dict[str, Any], *, status: int = 200) -> None: - raise NotImplementedError - - def _delivery_review(self) -> None: - with effect_runtime_request_scope(): - self._status(delivery_review=True) - - def _status(self, *, delivery_review: bool = False) -> None: - query = parse_qs(urlparse(self.path).query, keep_blank_values=True) - query_error_code = "invalid_goal_activation" - try: - activation_state_filter = parse_goal_activation_filter(query) - query_error_code = "invalid_workspace_query" - views = query.get("view") - if views is not None and (delivery_review or views != ["workspace-directory"]): - raise ValueError("unsupported workspace status view") - if delivery_review and not self._require_loopback_origin(): - return - requested_goals = query.get("goal_id") - goal_id = self.server.selected_goal_id - if requested_goals is not None: - if len(requested_goals) != 1 or not requested_goals[0].strip(): - raise ValueError("one nonempty goal_id is required") - requested_goal = requested_goals[0] - validate_goal_id(requested_goal) - if goal_id and requested_goal != goal_id: - raise ValueError("goal_id is outside this workspace") - goal_id = requested_goal - if delivery_review and not goal_id: - raise ValueError("delivery review requires one goal_id") - except ValueError as exc: - self._send_error( - str(exc), - status=400, - error_code=query_error_code, - ) - return - try: - status_runtime_root = goal_read_runtime_root(self.server) - registry = {} - if views is not None or requested_goals is not None or delivery_review: - # Discovery reads only metadata. Full status/runtime reads keep - # their existing lifecycle-only profile refusal. - registry = (_discovery_registry(self.server.registry_path) if views == ["workspace-directory"] - else load_registry(self.server.registry_path)) - directory = workspace_goal_directory(registry, selected_goal_id=goal_id) - if goal_id and not directory["goals"]: - self._send_error("Goal is no longer registered.", status=404) - return - if views == ["workspace-directory"]: - directory["contract"] = _discovery_contract(self.server.registry_path) - # Discovery stays metadata-only: redact with the roots this - # server knows, without requiring the full runtime wiring. - runtime_root = getattr(self.server, "runtime_root", None) - directory = json.loads( - redact_local_paths( - json.dumps(directory, ensure_ascii=False), - protected_paths=[ - self.server.registry_path, - *(getattr(self.server, "scan_roots", None) or []), - *([runtime_root] if runtime_root else []), - ], - ) - ) - self._send_json(directory) - return - linked = None - if goal_id and registry.get("goal_source_links"): - from .control_plane.status.goal_source import SourceScopeError - from .presentation.goal_source_routing import goal_source_catalog - catalog = goal_source_catalog(self.server) - if delivery_review and catalog.resolve(goal_id, registry): - self._send_error("Execution review belongs to the original Goal runtime.", status=403) - return - try: - linked = catalog.status(goal_id, registry) - except SourceScopeError: - self._send_error("The original Goal is no longer available in this source.", - status=409) - return - except (OSError, subprocess.SubprocessError): - self._send_error("The original Goal source is unavailable; its link is preserved.", - status=503, error_code="goal_source_unavailable") - return - projection = linked if linked is not None else collect_status( - registry_path=self.server.registry_path, - # The Chat service already starts on an absent registry as an - # empty workspace (serve_chat); its status reports the same - # fact as a warning, so a fresh install is not a failed source. - allow_missing_registry=True, - runtime_root_override=status_runtime_root, - limit=self.server.limit, - goal_id=goal_id, - include_zcode_goal_eligibility=not delivery_review, - include_task_graph=delivery_review, - activation_state_filter=activation_state_filter, - include_goal_subagent_configuration=( - goal_subagent_configuration_enabled(self.server) - ), - ) - if requested_goals is not None or delivery_review: - latest = workspace_goal_directory( - load_registry(self.server.registry_path) - ) - if latest["registry_revision"] != directory["registry_revision"]: - self._send_error( - "Goal directory changed; refresh the workspace.", status=409 - ) - return - projection["workspace_registry_revision"] = directory[ - "registry_revision" - ] - if not delivery_review: - projection["local_dashboard_api"] = { - **(projection.get("local_dashboard_api") or {}), - "source": "chat", - "status_url": "/status.json", - "presentation_surfaces_url": DEFAULT_EXTENSION_PRESENTATION_SURFACES_PATH, - "presentation_detail_url": DEFAULT_EXTENSION_PROJECTION_PATH, - } - if linked is None: - attach_host_thread_activity( - projection, observers=codex_thread_observers( - cache_root=self.server.runtime_root / "observation-cache" / "codex-activity", - ) - ) - attach_host_delivery_windows( - projection, - expectations=codex_delivery_expectations, - ) - if delivery_review: - if projection.get("ok") is not True: - self._send_error("Delivery review sources are unavailable.", status=503) - return - item = next((row for row in projection.get("attention_queue", {}).get("items", []) - if row.get("goal_id") == goal_id), {}) - goal = next((row for row in projection.get("run_history", {}).get("goals", []) - if row.get("id") == goal_id), {}) - goal_map = _goal_task_map(self.server, goal_id) - observed_at = now_utc_iso() - components = [item.get("task_graph_projection"), goal_map, goal.get("acceptance_observation")] - upstream = [value for value in [projection.get("projection_envelope"), - goal_map.get("projection_envelope") if goal_map else None] if isinstance(value, dict)] - sources = [] - if not isinstance(projection.get("projection_envelope"), dict): - sources.append(source_fact("status", read_status="not_read")) - if goal_map is None: - sources.append(source_fact("goal_todos", read_status="unreadable")) - missing = sum(value is None for value in components) - envelope = seal_projection_envelope( - projection="delivery_review", observed_at=observed_at, - sources=sources, upstream=upstream, - coverage={"scope": "selected_goal_delivery", "expected_count": 3, - "included_count": 3 - missing, - "omitted": [{"reason": "unavailable_components", "count": missing}] if missing else []}, - ) - projection = { - "ok": True, - "goal_id": goal_id, - "observed_at": observed_at, - "projection_envelope": envelope, - "graph": item.get("task_graph_projection"), - "goal_map": goal_map, - "acceptance": goal.get("acceptance_observation"), - } - protected_paths = [ - self.server.registry_path, - *self.server.scan_roots, - self.server.runtime_root, - *([status_runtime_root] if status_runtime_root else []), - ] - projection = json.loads( - redact_local_paths( - json.dumps(projection, ensure_ascii=False), - protected_paths=protected_paths, - ) - ) - if delivery_review: - # Validate the outbound view, after removing owner-local paths. - validate_public_safe_value(projection) - except Exception as exc: # noqa: BLE001 - do not expose local projection failures. - # The storage owner already exposes inspection and reviewed import. - # Preserve its typed refusal, not private paths or migration payloads. - if isinstance(exc, LocalCoordinationAuthorityUnavailable) and exc.code == "canonical_authority_required": - self._send_error( - "This Goal needs a storage upgrade. Open its storage settings to review recovery.", - status=503, - error_code=exc.code, - ) - return - self._send_error( - "LoopX status could not be projected for the workspace.", - status=500, - error_code=( - "workspace_status_access_denied" - if requested_goals is not None - and views is None - and _status_access_denied(exc) - else None - ), - ) - return - self._send_json(projection) + _implementation._legacy_status_import_warned = True +sys.modules[__name__] = _implementation diff --git a/tests/architecture/test_source_session_registry_denial.py b/tests/architecture/test_source_session_registry_denial.py index a6e0306716..95bda00492 100644 --- a/tests/architecture/test_source_session_registry_denial.py +++ b/tests/architecture/test_source_session_registry_denial.py @@ -13,7 +13,7 @@ "loopx/control_plane/coordination/new_goal_storage_transport.py", "loopx/chat_server.py", "loopx/presentation/goal_direction_api.py", - "loopx/chat_status_api.py", + "loopx/chat/status_api.py", "loopx/control_plane/goals/goal_amendment_runtime.py", "loopx/control_plane/goals/source_session_registration.py", "loopx/presentation/goal_amendment_api.py", @@ -137,7 +137,7 @@ def test_direct_project_registry_loaders_have_source_session_denial() -> None: "loopx/presentation/request_control_api.py": {"_request_control_source"}, "loopx/chat_server.py": {"serve_chat"}, "loopx/presentation/goal_direction_api.py": {"_direction_route"}, - "loopx/chat_status_api.py": {"_discovery_registry"}, + "loopx/chat/status_api.py": {"_discovery_registry"}, "loopx/control_plane/goals/direction_runtime.py": {"_source"}, "loopx/control_plane/coordination/new_goal_storage_transport.py": {"initialize_goal_storage_target"}, "loopx/presentation/goal_amendment_api.py": {"_amendment_inspect"}, diff --git a/tests/test_chat_internal_module_layout.py b/tests/test_chat_internal_module_layout.py index ef0478b0a9..1e4c6a0f86 100644 --- a/tests/test_chat_internal_module_layout.py +++ b/tests/test_chat_internal_module_layout.py @@ -13,6 +13,7 @@ from loopx.chat.monitor_actions import ChatMonitorActionMixin from loopx.chat.session_open import open_chat_session from loopx.chat.ssh_source_api import SshSourceRequestMixin +from loopx.chat.status_api import ChatStatusRequestMixin, goal_read_runtime_root from loopx.chat.todo_actions import ChatTodoActionMixin from loopx.chat.todo_detail import TodoDetailRequestMixin from loopx.chat.usage_statistics_api import UsageStatisticsRequestMixin @@ -56,3 +57,9 @@ def test_chat_internal_helpers_have_no_top_level_compatibility_shims() -> None: "loopx.chat_workspace_directory", ): assert importlib.util.find_spec(old_module) is None + + +def test_public_status_owner_is_nested_without_removing_the_legacy_import() -> None: + assert ChatStatusRequestMixin.__module__ == "loopx.chat.status_api" + assert goal_read_runtime_root.__module__ == "loopx.chat.status_api" + assert importlib.util.find_spec("loopx.chat_status_api") is not None diff --git a/tests/test_chat_status_public_compatibility.py b/tests/test_chat_status_public_compatibility.py index bf3737cafe..0a1e25b6ea 100644 --- a/tests/test_chat_status_public_compatibility.py +++ b/tests/test_chat_status_public_compatibility.py @@ -16,7 +16,8 @@ @pytest.mark.parametrize("first_import", [ - "loopx.chat_status_api", "loopx.chat_server", "loopx.chat.todo_detail", + "loopx.chat_status_api", "loopx.chat.status_api", "loopx.chat_server", + "loopx.chat.todo_detail", ]) def test_historical_status_imports_survive_import_order_reload_and_pickle(first_import): # A fresh process avoids an earlier test silently choosing the import order. @@ -28,14 +29,21 @@ def test_historical_status_imports_survive_import_order_reload_and_pickle(first_ importlib.import_module(sys.argv[1]) public = importlib.import_module('loopx.chat_status_api') +canonical = importlib.import_module('loopx.chat.status_api') from loopx.chat_status_api import ChatStatusRequestMixin, goal_read_runtime_root from loopx.chat_server import ChatRequestHandler from loopx.chat.todo_detail import goal_read_runtime_root as todo_runtime_root assert issubclass(ChatRequestHandler, ChatStatusRequestMixin) +assert public is canonical +assert sys.modules['loopx.chat_status_api'] is canonical +assert ChatStatusRequestMixin.__module__ == 'loopx.chat.status_api' +assert goal_read_runtime_root.__module__ == 'loopx.chat.status_api' assert todo_runtime_root is goal_read_runtime_root assert pickle.loads(b'cloopx.chat_status_api\\nChatStatusRequestMixin\\n.') is ChatStatusRequestMixin assert pickle.loads(b'cloopx.chat_status_api\\ngoal_read_runtime_root\\n.') is goal_read_runtime_root +assert pickle.loads(pickle.dumps(ChatStatusRequestMixin)) is ChatStatusRequestMixin +assert pickle.loads(pickle.dumps(goal_read_runtime_root)) is goal_read_runtime_root # Reload replaces definitions, but already-imported functions keep live globals. assert importlib.reload(public) is public @@ -52,6 +60,34 @@ def test_historical_status_imports_survive_import_order_reload_and_pickle(first_ assert result.returncode == 0, result.stderr +def test_historical_status_deprecation_is_once_per_process_and_not_on_canonical_import(): + code = """ +import importlib +import sys +import warnings + +with warnings.catch_warnings(record=True) as seen: + warnings.simplefilter('always', DeprecationWarning) + canonical = importlib.import_module('loopx.chat.status_api') + assert not seen + public = importlib.import_module('loopx.chat_status_api') + assert public is canonical + assert importlib.import_module('loopx.chat_status_api') is canonical + assert importlib.reload(public) is canonical + del sys.modules['loopx.chat_status_api'] + assert importlib.import_module('loopx.chat_status_api') is canonical + assert len(seen) == 1 + assert seen[0].category is DeprecationWarning + assert 'loopx.chat.status_api' in str(seen[0].message) + assert 'at least one minor release' in str(seen[0].message) +""" + result = subprocess.run( + [sys.executable, "-c", code], + capture_output=True, text=True, timeout=60, check=False, + ) + assert result.returncode == 0, result.stderr + + @pytest.mark.parametrize("source_bound,controller_root,expected", [ (True, "coordinator-decoy", None), (False, "controller-runtime", "controller-runtime"), @@ -95,7 +131,39 @@ def unreadable(_path): public.goal_read_runtime_root(server) -def test_historical_status_patch_points_reach_the_real_http_handler(monkeypatch, tmp_path): +@pytest.fixture +def status_server(tmp_path): + server = ChatHTTPServer(("127.0.0.1", 0), ChatRequestHandler) + server.registry_path = tmp_path / "registry.json" + server.runtime_root = tmp_path / "runtime" + server.runtime_root_override = None + server.selected_goal_id = None + server.scan_roots = [] + server.limit = 20 + server.verbose = False + server.goal_subagent_configuration_enabled = False + thread = threading.Thread(target=server.serve_forever, daemon=True) + thread.start() + + def get(query): + connection = http.client.HTTPConnection(*server.server_address, timeout=5) + try: + connection.request("GET", "/status.json" + query) + response = connection.getresponse() + return response.status, json.loads(response.read()) + finally: + connection.close() + + try: + yield server, get + finally: + server.shutdown() + thread.join(timeout=5) + server.server_close() + + +def test_historical_status_patch_points_reach_the_real_http_handler(monkeypatch, status_server): + server, get = status_server registry = {"goals": [{"id": "alpha"}]} registry_reads, collected = [], [] @@ -119,30 +187,31 @@ def attach_delivery(projection, **_kwargs): monkeypatch.setattr("loopx.chat_status_api.attach_host_thread_activity", attach_activity) monkeypatch.setattr("loopx.chat_status_api.attach_host_delivery_windows", attach_delivery) monkeypatch.setattr("loopx.chat_status_api.codex_thread_observers", lambda **_: []) - server = ChatHTTPServer(("127.0.0.1", 0), ChatRequestHandler) - server.registry_path = tmp_path / "registry.json" - server.runtime_root = tmp_path / "runtime" - server.runtime_root_override = None - server.selected_goal_id = None - server.scan_roots = [] - server.limit = 20 - server.verbose = False - server.goal_subagent_configuration_enabled = False - thread = threading.Thread(target=server.serve_forever, daemon=True) - thread.start() - connection = http.client.HTTPConnection(*server.server_address, timeout=5) - try: - connection.request("GET", "/status.json?goal_id=alpha") - response = connection.getresponse() - payload = json.loads(response.read()) - assert response.status == 200, payload - assert payload["compatibility_probe"] == "collector" - assert payload["compatibility_activity"] is True - assert payload["compatibility_delivery"] is True - assert len(collected) == 1 and collected[0]["goal_id"] == "alpha" - assert registry_reads == [server.registry_path, server.registry_path] - finally: - connection.close() - server.shutdown() - thread.join(timeout=5) - server.server_close() + code, payload = get("?goal_id=alpha") + assert code == 200, payload + assert payload["compatibility_probe"] == "collector" + assert payload["compatibility_activity"] is True + assert payload["compatibility_delivery"] is True + assert len(collected) == 1 and collected[0]["goal_id"] == "alpha" + assert registry_reads == [server.registry_path, server.registry_path] + + +def test_lifecycle_registry_discovery_does_not_grant_status_execution(monkeypatch, status_server): + server, get = status_server + server.registry_path.write_text(json.dumps({ + "schema_version": 1, "profile_id": "source_session_v1", + "goals": [{"id": "alpha"}], + })) + collected = [] + + def forbidden(**kwargs): + collected.append(kwargs) + pytest.fail("lifecycle metadata must not reach the status collector") + + monkeypatch.setattr("loopx.chat.status_api.collect_status", forbidden) + code, directory = get("?view=workspace-directory") + assert code == 200, directory + assert [row["id"] for row in directory["goals"]] == ["alpha"] + code, refused = get("?goal_id=alpha") + assert code >= 400 and not refused["ok"] + assert collected == []