fix(health): use explicit workspace component views for runtime reports

docs/en/operations.md

@@ -22,7 +22,7 @@ reme app_config
 
 - `find_reme` confirms the actual host, port, and PID;
 - `version` verifies that the CLI reaches the service;
-- `health_check` reports component health;
+- `health_check` reports shared and selected-workspace component health and labels the workspace;
 - `status` estimates shared and selected-workspace component memory, labels the workspace, and reports process-wide RSS;
 - `app_config` returns the effective configuration with secrets redacted.
 

docs/zh/operations.md

@@ -21,7 +21,7 @@ reme app_config
 
 - `find_reme`：服务是否存在，以及实际 host、port、PID；
 - `version`：CLI 能否成功访问服务；
-- `health_check`：组件健康状态；
+- `health_check`：共享组件与所选工作区组件的健康状态，并标明工作区；
 - `status`：共享组件和所选工作区状态组件的内存估算，标明工作区，并报告整个进程的 RSS；
 - `app_config`：隐藏密钥后的实际生效配置。
 

reme/components/application_context.py

@@ -87,6 +87,15 @@ class ApplicationContext:
         for ctype, group in value.items():
             self._components[ctype] = group
 
+    def components_for(self, workspace: WorkspaceContext) -> dict[str, dict[str, "BaseComponent"]]:
+        """Snapshot shared and bound-workspace groups without consulting the legacy view."""
+        if self.workspaces.get(workspace.alias) is not workspace:
+            raise ValueError(f"Workspace {workspace.alias!r} is not bound to this Application")
+        groups = {ctype: dict(group) for ctype, group in self.shared_components.items()}
+        for ctype, group in workspace.components.items():
+            groups.setdefault(ctype, {}).update(group)
+        return groups
+
     def get_component(self, ctype: str, name: str, alias: str | None = None) -> "BaseComponent | None":
         """Compatibility resolver for external callers that select a workspace by alias."""
         alias = alias or next(iter(self.workspaces))

reme/steps/common/health_check.py

@@ -170,13 +170,14 @@ def _is_healthy(ctype: ComponentEnum, status: dict) -> bool:
     return True
 
 
-def _collect_components(app_context) -> tuple[dict, bool]:
-    """Walk every registered component type and produce {type: {name: status}}, plus overall flag."""
+def _collect_components(app_context, workspace=None) -> tuple[dict, bool]:
+    """Collect shared and selected-workspace health, never default-workspace substitutes."""
     components: dict = {}
     healthy = True
+    selected_components = app_context.components_for(workspace or app_context.workspace)
     for ctype, handler in _HANDLERS.items():
         bucket = {}
-        for name, comp in app_context.components.get(ctype, {}).items():
+        for name, comp in selected_components.get(ctype, {}).items():
             status = handler(comp)
             bucket[name] = status
             if not _is_healthy(ctype, status):
@@ -193,15 +194,17 @@ class HealthCheckStep(BaseStep):
         assert self.context is not None
 
         if self.app_context is not None:
-            components, healthy = _collect_components(self.app_context)
+            components, healthy = _collect_components(self.app_context, self.context.workspace)
         else:
             components, healthy = {}, True
 
-        health = {"version": __version__, "healthy": healthy, "components": components}
+        workspace = self.context.workspace.alias if self.context.workspace is not None else None
+        health = {"version": __version__, "workspace": workspace, "healthy": healthy, "components": components}
         self.logger.info(f"[{self.name}] health collected: {health}")
 
         emoji = "✅" if healthy else "❌"
         label = "healthy" if healthy else "unhealthy"
-        self.context.response.answer = f"{emoji} ReMe v{__version__} - {label}"
+        scope = f" (workspace: {workspace})" if workspace is not None else ""
+        self.context.response.answer = f"{emoji} ReMe v{__version__} - {label}{scope}"
         self.context.response.metadata["health"] = health
         return self.context.response

reme/steps/common/status.py

@@ -96,13 +96,10 @@ def _collect_memory(app_context, workspace=None) -> dict:
     total = 0
     if app_context is not None:
         workspace = workspace or app_context.workspace
+        selected_components = app_context.components_for(workspace)
         for component_type in sorted(_TRACKED_COMPONENT_TYPES, key=lambda item: item.value):
             group = {}
-            selected_components = {
-                **app_context.shared_components.get(component_type, {}),
-                **workspace.components.get(component_type, {}),
-            }
-            for name, component in sorted(selected_components.items()):
+            for name, component in sorted(selected_components.get(component_type, {}).items()):
                 # _component_size keeps an independent seen set so each component
                 # remains understandable in isolation. Consequently, a non-component
                 # object shared by multiple components may be included more than once

tests/unit/test_health_check_workspace.py

@@ -0,0 +1,103 @@
+"""Health and memory reports must describe the request's actual workspace."""
+
+import pytest
+
+from reme.application import Application
+from reme.components.application_context import ApplicationContext
+from reme.components.base_component import BaseComponent
+from reme.enumeration import ComponentEnum
+from reme.schema import FileNode
+from reme.steps.common.health_check import HealthCheckStep
+from reme.steps.common.status import _collect_memory
+
+
+@pytest.mark.asyncio
+async def test_health_job_reports_beta_failure_without_affecting_alpha(tmp_path):
+    """Exercise real graphs and the public job before and after only beta stops."""
+    app = Application(
+        workspace_dir={"alpha": str(tmp_path / "alpha"), "beta": str(tmp_path / "beta")},
+        enable_logo=False,
+        log_to_console=False,
+        log_to_file=False,
+        service={"backend": "cli"},
+        components={"file_graph": {"default": {"backend": "local"}}},
+        jobs={"health_check": {"backend": "base", "steps": [{"backend": "health_check_step"}]}},
+    )
+    async with app:
+        alpha = app.context.workspaces["alpha"].get_component(ComponentEnum.FILE_GRAPH, "default")
+        beta = app.context.workspaces["beta"].get_component(ComponentEnum.FILE_GRAPH, "default")
+        await beta.upsert_nodes([FileNode(path="beta.md", st_mtime=1)])
+        alpha_before = (await app.run_job("health_check", workspace="alpha")).metadata["health"]
+        response = await app.run_job("health_check", workspace="beta")
+        before = response.metadata["health"]
+        assert before["components"]["file_graph"]["default"]["n_nodes"] == 1
+        assert before["workspace"] == "beta" and before["healthy"] is True
+        assert "beta" in response.answer
+        assert before["components"]["file_graph"]["default"]["is_started"] is True
+        await beta.close()
+        response = await app.run_job("health_check", workspace="beta")
+        assert response.success is True  # The diagnostic job succeeded; its health result is negative.
+        after = response.metadata["health"]
+        assert after["workspace"] == "beta" and after["healthy"] is False
+        assert after["components"]["file_graph"]["default"]["is_started"] is False
+        assert after["components"]["file_graph"]["default"]["n_nodes"] == 1
+        assert alpha.is_started is True and beta.is_started is False
+        assert (await app.run_job("health_check", workspace="alpha")).metadata["health"] == alpha_before
+        assert (await app.run_job("health_check")).metadata["health"] == alpha_before
+
+
+@pytest.mark.asyncio
+@pytest.mark.parametrize("named", [False, True])
+async def test_health_single_workspace_labels_default_and_preserves_payload(tmp_path, named):
+    """String and one-entry mapping reports identify their default workspace."""
+    context = ApplicationContext(workspace_dir={"named": str(tmp_path)} if named else str(tmp_path))
+    alias = context.workspace.alias
+    step = HealthCheckStep(app_context=context)
+    response = await step()
+    assert response.metadata["health"]["workspace"] == alias
+    assert response.metadata["health"]["healthy"] is True
+    assert set(response.metadata["health"]["components"]) == {
+        "embedding_store",
+        "file_graph",
+        "file_store",
+        "keyword_index",
+    }
+
+
+@pytest.mark.asyncio
+async def test_reports_include_shared_and_selected_components_without_mutating_registries(tmp_path):
+    """Shared and scoped names of the same type both remain visible; alpha never leaks."""
+    context = ApplicationContext(workspace_dir={"alpha": str(tmp_path / "alpha"), "beta": str(tmp_path / "beta")})
+    shared, alpha, beta = (BaseComponent() for _ in range(3))
+    context.shared_components[ComponentEnum.FILE_GRAPH] = {"shared": shared}
+    context.workspaces["alpha"].components[ComponentEnum.FILE_GRAPH] = {"alpha": alpha}
+    context.workspaces["beta"].components[ComponentEnum.FILE_GRAPH] = {"beta": beta}
+    await shared.start()
+    await alpha.start()
+    await beta.start()
+    try:
+        from reme.components.runtime_context import RuntimeContext
+
+        step = HealthCheckStep(app_context=context)
+        response = await step(RuntimeContext(workspace=context.workspaces["beta"]))
+        assert set(response.metadata["health"]["components"]["file_graph"]) == {"shared", "beta"}
+        memory = _collect_memory(context, context.workspaces["beta"])
+        assert set(memory["components"]["file_graph"]) == {"shared", "beta"}
+        await shared.close()
+        response = await step(RuntimeContext(workspace=context.workspaces["beta"]))
+        assert response.metadata["health"]["healthy"] is False
+        assert set(context.shared_components[ComponentEnum.FILE_GRAPH]) == {"shared"}
+        assert set(context.workspaces["alpha"].components[ComponentEnum.FILE_GRAPH]) == {"alpha"}
+        assert set(context.workspaces["beta"].components[ComponentEnum.FILE_GRAPH]) == {"beta"}
+    finally:
+        await shared.close()
+        await alpha.close()
+        await beta.close()
+
+
+def test_component_report_view_rejects_foreign_workspace(tmp_path):
+    """A workspace object from another application cannot select this application's resources."""
+    context = ApplicationContext(workspace_dir=str(tmp_path / "current"))
+    foreign = ApplicationContext(workspace_dir=str(tmp_path / "foreign"))
+    with pytest.raises(ValueError, match="not bound to this Application"):
+        context.components_for(foreign.workspace)