diff --git a/docs/webui-local-dev.md b/docs/webui-local-dev.md index d9cc022..3e01401 100644 --- a/docs/webui-local-dev.md +++ b/docs/webui-local-dev.md @@ -77,7 +77,9 @@ status, onboarding checklist state, and the fail-closed error payloads (#635). | `/api/actions/{id}/preview` | Mutation ledger preview (GET, read-only) | | `/leases` | Lease and collision visibility (#433) | | `/api/leases` | JSON lease/collision export | -| `/sessions` | Phase 1 shell stub — session inventory (backed by #636) | +| `/sessions` | Runtime and session view (#641) — health + inventory sessions/namespaces/worktrees | +| `/api/sessions` | JSON export for the runtime/session view | +| `/api/v1/sessions` | Versioned alias of `/api/sessions` | | `/inventory` | Phase 1 shell stub — unified inventory (backed by #636) | | `/timeline` | Phase 1 shell stub — workflow event timeline | | `/policy` | Phase 1 shell stub — capability/role policy placeholder | @@ -284,11 +286,46 @@ The header carries two read-only status badges — an **environment** badge a **mode: read-only** badge — plus a **Docs** link to this document. No privileged action controls are present in the Phase 1 shell. -Not-yet-implemented surfaces (`/sessions`, `/inventory`, `/timeline`, -`/policy`, `/insights`) resolve to graceful read-only stub pages instead of -404s; their backing views land in later child issues of #631 (the inventory -surfaces are backed by #636). Mutating methods on stub routes still fail closed -with `read-only-mvp`. +Not-yet-implemented surfaces (`/inventory`, `/timeline`, `/policy`, +`/insights`) resolve to graceful read-only stub pages instead of 404s; their +backing views land in later child issues of #631 (the inventory surfaces are +backed by #636). Mutating methods on stub routes still fail closed with +`read-only-mvp`. + +### Runtime and sessions (#641) + +`/sessions` is a live Phase 1 read-only view that composes: + +* runtime health from `#430` (profile, role, identity, master parity, stale warning) +* control-plane sessions / leases and filesystem locks / worktrees / namespaces from `#636` +* durable contamination markers when detectable (`#630` runtime recovery, `#671` stable-branch push) + +It surfaces stale indicators (dead PID, expired lease) and never silences an +active contamination marker. Recovery links point only at sanctioned +reconnect/operator restart docs (`docs/mcp-namespace-eof-recovery.md`, +`docs/mcp-namespace-health.md`, `docs/mcp-restart-path-inventory.md`, this +document). The page does **not** restart, kill, or take over sessions; manual +`pkill` of MCP daemons is contamination, not recovery. + +Honesty rules specific to this view: + +* **Ownership columns never assert absence they cannot prove.** When the + `leases` or `locks` section is degraded or unavailable, the Leases and + Worktree-binding cells render `unknown (inventory )` with an + *authority unproven* badge instead of `none` / `unbound`, and a caveat names + the unreadable sections. A worktree binding is correlated through lease work + numbers, so it is unproven when *either* section fails to read. + `/api/sessions` carries the same facts as `ownership_authority_complete`, + `ownership_section_status`, and per-row `lease_authority` / + `worktree_authority`, so a JSON consumer can tell "holds none" from "could + not be read". +* **Contamination text is redacted at the display boundary.** Marker payloads + (`command_summary`, `reason_class`, `session_id`, `role`) are + operator-supplied free text that does not arrive through inventory scrubbing, + so they pass through `webui.inventory.scrub_text`, which collapses `$HOME` and + redacts credential-shaped tokens and URL userinfo *anywhere* in the string. + The write-time redactor is a narrow denylist and is not relied on. The field + itself is kept — it is the `#630` evidence naming which daemon was killed. ## System-health dashboard (#639) diff --git a/tests/test_webui_sessions_view.py b/tests/test_webui_sessions_view.py new file mode 100644 index 0000000..4799b77 --- /dev/null +++ b/tests/test_webui_sessions_view.py @@ -0,0 +1,739 @@ +"""Tests for the Runtime and session view (Phase 1, #641). + +Covers clean and stale session rendering, contamination marker surfacing, +worktree binding display, sanctioned recovery links (no pkill), nav/live +status, and the JSON API export. + +Also pins the two invariants a reviewer found violated at head a81db754: +degraded ownership sections must render as *unknown* rather than as an +affirmative "none"/"unbound", and contamination payload text must be redacted +at the display boundary rather than trusted from the write-time denylist. +""" + +from __future__ import annotations + +import os +import sys +import unittest +from pathlib import Path +from unittest import mock + +sys.path.insert(0, str(Path(__file__).resolve().parent.parent)) + +from tests.webui_testclient import TestClient + +from webui.app import create_app +from webui.inventory import ( + AUTHORITY_CONTROL_PLANE_DB, + AUTHORITY_FILESYSTEM, + InventorySection, + InventorySnapshot, + STATUS_DEGRADED, + STATUS_OK, + STATUS_UNAVAILABLE, +) +from webui.nav import NAV_GROUPS, STUB_PAGES, iter_nav_items +from webui.runtime_health import FileHash, RuntimeSnapshot +from webui.session_loader import ( + ContaminationMarker, + SessionRow, + SessionViewSnapshot, + _build_session_rows, + _inspect_contamination, + load_session_view_snapshot, + snapshot_to_dict, +) +from webui.session_views import render_sessions_page + + +def _runtime( + *, + stale: str | None = None, + profile: str = "prgs-author", + role: str = "author", +) -> RuntimeSnapshot: + return RuntimeSnapshot( + project_id="gitea-tools", + repo_root="/tmp/repo", + remote="prgs", + host="gitea.prgs.cc", + profile_name=profile, + role_kind=role, + config_model="v2-contexts", + profile_mode="dynamic-profile", + profile_source="config file profile", + authenticated_username="jcwalker3", + identity_error=None, + repo_sha="a" * 40, + remote_master_sha="a" * 40, + commits_behind_master=0, + stale_runtime_warning=stale, + shell_health={"shell_use_allowed": True, "consecutive_spawn_failures": 0}, + workflow_hashes=( + FileHash(label="SKILL.md", path="skills/llm-project-workflow/SKILL.md", sha256="abc"), + ), + schema_hashes=(), + restart_guidance="docs/mcp-namespace-eof-recovery.md", + fetch_error=None, + ) + + +def _inventory( + *, + sessions: tuple[dict, ...] = (), + leases: tuple[dict, ...] = (), + locks: tuple[dict, ...] = (), + worktrees: tuple[dict, ...] = (), + namespaces: tuple[dict, ...] = (), + statuses: dict[str, str] | None = None, +) -> InventorySnapshot: + """Build a snapshot; ``statuses`` degrades named sections (default all ok).""" + status_of = statuses or {} + + def _status(name: str) -> str: + return status_of.get(name, STATUS_OK) + + sections = ( + InventorySection( + name="sessions", + authority=AUTHORITY_CONTROL_PLANE_DB, + status=_status("sessions"), + items=sessions, + ), + InventorySection( + name="leases", + authority=AUTHORITY_CONTROL_PLANE_DB, + status=_status("leases"), + items=leases, + ), + InventorySection( + name="locks", + authority=AUTHORITY_FILESYSTEM, + status=_status("locks"), + items=locks, + ), + InventorySection( + name="worktrees", + authority=AUTHORITY_FILESYSTEM, + status=_status("worktrees"), + items=worktrees, + ), + InventorySection( + name="namespaces", + authority=AUTHORITY_FILESYSTEM, + status=_status("namespaces"), + items=namespaces + or ( + { + "profile_name": "prgs-author", + "role": "author", + "mcp_namespace": "gitea-author", + "capability_summary": { + "can_author": True, + "can_review": False, + "can_merge": False, + }, + "active": True, + }, + ), + reason="only the profile serving this web process is observable", + ), + ) + index = {section.name: section for section in sections} + return InventorySnapshot( + generated_at="2026-07-25T00:00:00+00:00", + sections=sections, + collisions=(), + correlations=(), + scan_ms=1.0, + _section_index=index, + ) + + +def _clean_session() -> dict: + return { + "session_id": "prgs-author-111-clean", + "role": "author", + "profile": "prgs-author", + "namespace": "gitea-author", + "pid": 1111, + "pid_alive": True, + "status": "active", + "started_at": "2026-07-25T00:00:00Z", + "last_heartbeat_at": "2026-07-25T01:00:00Z", + } + + +def _stale_session() -> dict: + return { + "session_id": "prgs-author-222-stale", + "role": "author", + "profile": "prgs-author", + "namespace": "gitea-author", + "pid": 2222, + "pid_alive": False, + "status": "active", + "started_at": "2026-07-24T00:00:00Z", + "last_heartbeat_at": "2026-07-24T01:00:00Z", + } + + +class TestBuildSessionRows(unittest.TestCase): + def test_clean_session_has_no_stale_or_contamination_flags(self): + inventory = _inventory( + sessions=(_clean_session(),), + leases=( + { + "lease_id": "lease-clean", + "session_id": "prgs-author-111-clean", + "status": "active", + "expired": False, + "work_kind": "issue", + "work_number": 641, + }, + ), + locks=( + { + "issue_number": 641, + "branch_name": "feat/issue-641-runtime-session-view", + "worktree_path": "~/Development/Gitea-Tools/branches/feat-issue-641", + "live": True, + }, + ), + ) + rows = _build_session_rows(inventory, contamination=()) + self.assertEqual(len(rows), 1) + row = rows[0] + self.assertEqual(row.session_id, "prgs-author-111-clean") + self.assertEqual(row.role, "author") + self.assertEqual(row.namespace, "gitea-author") + self.assertEqual(row.pid_alive, True) + self.assertEqual(row.lease_ids, ("lease-clean",)) + self.assertEqual(row.work_refs, ("issue#641",)) + self.assertTrue(row.worktree_paths) + self.assertEqual(row.stale_flags, ()) + self.assertEqual(row.contamination_flags, ()) + + def test_stale_session_flags_dead_pid(self): + inventory = _inventory(sessions=(_stale_session(),)) + rows = _build_session_rows(inventory, contamination=()) + self.assertEqual(rows[0].stale_flags, ("pid-dead",)) + + def test_contamination_marker_binds_to_session(self): + inventory = _inventory(sessions=(_clean_session(),)) + marker = ContaminationMarker( + kind="runtime_recovery_contamination", + on_disk=True, + has_payload=True, + summary="manual daemon kill", + reason_class="manual_daemon_kill", + session_id="prgs-author-111-clean", + role="author", + command_summary="pkill -f mcp_server.py", + cleared=False, + ) + rows = _build_session_rows(inventory, contamination=(marker,)) + self.assertIn("runtime_recovery_contamination", rows[0].contamination_flags) + + def test_process_wide_contamination_surfaces_on_all_sessions(self): + inventory = _inventory(sessions=(_clean_session(), _stale_session())) + marker = ContaminationMarker( + kind="stable_branch_contamination", + on_disk=True, + has_payload=True, + summary="direct master push attempt", + reason_class="stable_branch_push", + session_id=None, + cleared=False, + ) + rows = _build_session_rows(inventory, contamination=(marker,)) + self.assertEqual(len(rows), 2) + for row in rows: + self.assertTrue( + any("stable_branch_contamination" in f for f in row.contamination_flags) + ) + + +class TestRenderSessionsPage(unittest.TestCase): + def _snapshot( + self, + *, + sessions: tuple[dict, ...], + contamination: tuple[ContaminationMarker, ...] = (), + stale_runtime: str | None = None, + ) -> SessionViewSnapshot: + inventory = _inventory( + sessions=sessions, + leases=( + { + "lease_id": "lease-1", + "session_id": sessions[0]["session_id"] if sessions else "", + "status": "active", + "expired": False, + "work_kind": "issue", + "work_number": 641, + }, + ) + if sessions + else (), + locks=( + { + "issue_number": 641, + "worktree_path": "branches/feat-issue-641", + }, + ) + if sessions + else (), + worktrees=( + { + "rel_path": "branches/feat-issue-641", + "branch": "feat/issue-641-runtime-session-view", + "classification": "active_issue_work", + "registered_worktree": True, + "dirty": False, + }, + ), + ) + rows = _build_session_rows(inventory, contamination) + return SessionViewSnapshot( + runtime=_runtime(stale=stale_runtime), + inventory=inventory, + sessions=rows, + contamination_markers=contamination, + ) + + def test_clean_session_render(self): + html = render_sessions_page(self._snapshot(sessions=(_clean_session(),))) + self.assertIn("Runtime and sessions", html) + self.assertIn("prgs-author-111-clean", html) + self.assertIn("gitea-author", html) + self.assertIn("branches/feat-issue-641", html) + self.assertIn("Sanctioned recovery", html) + self.assertIn("docs/mcp-namespace-eof-recovery.md", html) + # Recovery section must name reconnect and forbid manual kill. + recovery_idx = html.lower().find("sanctioned recovery") + self.assertGreaterEqual(recovery_idx, 0) + recovery = html[recovery_idx:].lower() + self.assertIn("reconnect", recovery) + self.assertIn("contamination", recovery) + self.assertIn("not recovery", recovery) + self.assertNotIn("run pkill", recovery) + self.assertNotIn("killall", recovery) + + def test_stale_session_render(self): + html = render_sessions_page(self._snapshot(sessions=(_stale_session(),))) + self.assertIn("prgs-author-222-stale", html) + self.assertIn("pid-dead", html) + self.assertIn("badge-stale", html) + + def test_contamination_render_is_not_silent(self): + marker = ContaminationMarker( + kind="runtime_recovery_contamination", + on_disk=True, + has_payload=True, + summary="manual kill", + reason_class="manual_daemon_kill", + session_id="prgs-author-111-clean", + command_summary="pkill -f mcp_server.py", + cleared=False, + ) + html = render_sessions_page( + self._snapshot(sessions=(_clean_session(),), contamination=(marker,)) + ) + self.assertIn("Contamination markers", html) + self.assertIn("runtime_recovery_contamination", html) + self.assertIn("ACTIVE", html) + self.assertIn("badge-blocked", html) + + def test_stale_runtime_banner(self): + html = render_sessions_page( + self._snapshot( + sessions=(_clean_session(),), + stale_runtime="server behind master by 3 commits", + ) + ) + self.assertIn("Stale runtime", html) + self.assertIn("server behind master", html) + + +class TestSessionLoaderComposition(unittest.TestCase): + def test_load_with_injected_sources(self): + inventory = _inventory(sessions=(_clean_session(), _stale_session())) + snap = load_session_view_snapshot( + load_runtime=lambda: _runtime(), + load_inventory=lambda: inventory, + inspect_contamination=lambda **_k: { + "on_disk": False, + "has_payload": False, + "summary": "absent", + }, + load_contamination_payload=lambda **_k: None, + ) + self.assertEqual(len(snap.sessions), 2) + self.assertEqual(snap.stale_session_count, 1) + self.assertEqual(snap.contaminated_session_count, 0) + data = snapshot_to_dict(snap) + self.assertEqual(data["view"], "runtime-sessions") + self.assertEqual(data["issue"], 641) + self.assertTrue(data["read_only"]) + self.assertEqual(data["session_counts"]["total"], 2) + self.assertEqual(data["session_counts"]["stale"], 1) + self.assertIn("recovery_docs", data) + + +class TestSessionsRoutes(unittest.TestCase): + def setUp(self): + self.client = TestClient(create_app()) + inventory = _inventory( + sessions=(_clean_session(), _stale_session()), + leases=( + { + "lease_id": "lease-x", + "session_id": "prgs-author-111-clean", + "status": "active", + "expired": False, + "work_kind": "issue", + "work_number": 641, + }, + ), + locks=( + { + "issue_number": 641, + "worktree_path": "branches/feat-issue-641", + }, + ), + worktrees=( + { + "rel_path": "branches/feat-issue-641", + "branch": "feat/issue-641-runtime-session-view", + "classification": "active_issue_work", + "registered_worktree": True, + "dirty": False, + }, + ), + ) + rows = _build_session_rows(inventory, contamination=()) + self.snapshot = SessionViewSnapshot( + runtime=_runtime(stale="stale for test"), + inventory=inventory, + sessions=rows, + contamination_markers=(), + ) + self._patch = mock.patch( + "webui.app.load_session_view_snapshot", + return_value=self.snapshot, + ) + self._patch.start() + + def tearDown(self): + self._patch.stop() + + def test_sessions_page_live(self): + response = self.client.get("/sessions") + self.assertEqual(response.status_code, 200) + self.assertIn("Runtime and sessions", response.text) + self.assertIn("prgs-author-111-clean", response.text) + self.assertIn("prgs-author-222-stale", response.text) + self.assertIn("pid-dead", response.text) + self.assertIn("Sanctioned recovery", response.text) + self.assertNotIn("Phase 1 shell placeholder", response.text) + self.assertNotIn("child issue of #425", response.text.lower()) + + def test_api_sessions_json(self): + for path in ("/api/sessions", "/api/v1/sessions"): + response = self.client.get(path) + self.assertEqual(response.status_code, 200, path) + data = response.json() + self.assertEqual(data["view"], "runtime-sessions") + self.assertEqual(data["session_counts"]["total"], 2) + self.assertEqual(data["session_counts"]["stale"], 1) + self.assertTrue(data["read_only"]) + self.assertEqual(data["mutations"], []) + + def test_nav_marks_sessions_live(self): + sessions_items = [ + item for item in iter_nav_items() if item.href == "/sessions" + ] + self.assertEqual(len(sessions_items), 1) + self.assertEqual(sessions_items[0].status, "live") + self.assertNotIn("/sessions", STUB_PAGES) + # Home page should not mark Sessions as stub. + home = self.client.get("/") + self.assertEqual(home.status_code, 200) + self.assertIn('href="/sessions"', home.text) + # Stub marker only appears next to remaining stub destinations. + self.assertNotIn( + 'href="/sessions">Sessions (stub)', + home.text, + ) + + +class TestDegradedOwnershipAuthority(unittest.TestCase): + """B1: a section that could not be read must never render as absence.""" + + def _snapshot(self, inventory: InventorySnapshot) -> SessionViewSnapshot: + return SessionViewSnapshot( + runtime=_runtime(), + inventory=inventory, + sessions=_build_session_rows(inventory, contamination=()), + contamination_markers=(), + ) + + def test_unavailable_leases_mark_row_authority_unproven(self): + inventory = _inventory( + sessions=(_clean_session(),), + statuses={"leases": STATUS_UNAVAILABLE}, + ) + row = _build_session_rows(inventory, contamination=())[0] + self.assertEqual(row.lease_ids, ()) + self.assertEqual(row.lease_authority, STATUS_UNAVAILABLE) + # Worktree binding is correlated through lease work numbers, so it + # inherits the unreadable lease section. + self.assertEqual(row.worktree_authority, STATUS_UNAVAILABLE) + self.assertFalse(row.ownership_authority_complete) + + def test_readable_locks_are_not_reported_unbound_when_leases_degrade(self): + # The narrow variant: locks hold a real worktree_path and read cleanly, + # but the lease section that supplies the correlating work number does + # not. The row must say unknown, not "unbound". + inventory = _inventory( + sessions=(_clean_session(),), + locks=( + { + "issue_number": 641, + "worktree_path": "branches/feat-issue-641", + }, + ), + statuses={"leases": STATUS_DEGRADED}, + ) + row = _build_session_rows(inventory, contamination=())[0] + self.assertEqual(row.worktree_paths, ()) + self.assertEqual(row.worktree_authority, STATUS_DEGRADED) + self.assertFalse(row.ownership_authority_complete) + + def test_degraded_render_says_unknown_not_none_or_unbound(self): + inventory = _inventory( + sessions=(_clean_session(),), + statuses={"leases": STATUS_UNAVAILABLE, "locks": STATUS_UNAVAILABLE}, + ) + html = render_sessions_page(self._snapshot(inventory)) + self.assertIn("unknown (inventory unavailable)", html) + self.assertIn("authority unproven", html) + self.assertIn("Ownership authority incomplete", html) + # The affirmative-absence strings must be gone from the row entirely. + self.assertNotIn(">none<", html) + self.assertNotIn(">unbound<", html) + + def test_clean_inventory_still_renders_affirmative_absence(self): + # Guards against over-correcting B1 into "everything is unknown". + inventory = _inventory(sessions=(_clean_session(),)) + html = render_sessions_page(self._snapshot(inventory)) + self.assertIn(">none<", html) + self.assertIn(">unbound<", html) + # The column legend mentions "unknown (inventory …)" as static copy, so + # assert on the per-row marker and the concrete statuses instead. + self.assertNotIn("authority unproven", html) + self.assertNotIn("unknown (inventory unavailable)", html) + self.assertNotIn("unknown (inventory degraded)", html) + self.assertNotIn("Ownership authority incomplete", html) + + def test_json_export_carries_snapshot_and_per_row_authority(self): + inventory = _inventory( + sessions=(_clean_session(),), + statuses={"locks": STATUS_UNAVAILABLE}, + ) + data = snapshot_to_dict(self._snapshot(inventory)) + self.assertFalse(data["ownership_authority_complete"]) + self.assertEqual( + data["ownership_section_status"]["locks"], STATUS_UNAVAILABLE + ) + self.assertEqual(data["ownership_section_status"]["leases"], STATUS_OK) + self.assertIn("unknown, not unowned", data["ownership_note"]) + + row = data["sessions"][0] + self.assertTrue(row["lease_authority_complete"]) + self.assertFalse(row["worktree_authority_complete"]) + self.assertEqual(row["worktree_authority"], STATUS_UNAVAILABLE) + self.assertFalse(row["ownership_authority_complete"]) + self.assertIn("unknown, not unowned", row["ownership_note"]) + + def test_json_export_is_affirmative_when_every_source_reads(self): + inventory = _inventory(sessions=(_clean_session(),)) + data = snapshot_to_dict(self._snapshot(inventory)) + self.assertTrue(data["ownership_authority_complete"]) + self.assertTrue(data["sessions"][0]["ownership_authority_complete"]) + + def test_missing_session_list_is_not_reported_as_no_sessions(self): + inventory = _inventory(statuses={"sessions": STATUS_UNAVAILABLE}) + html = render_sessions_page(self._snapshot(inventory)) + self.assertIn("could not be read", html) + self.assertIn("not evidence that no sessions exist", html) + + def test_expired_lease_flags_row_as_stale(self): + inventory = _inventory( + sessions=(_clean_session(),), + leases=( + { + "lease_id": "lease-expired-1", + "session_id": "prgs-author-111-clean", + "status": "active", + "expired": True, + "work_kind": "issue", + "work_number": 641, + }, + ), + ) + row = _build_session_rows(inventory, contamination=())[0] + self.assertIn("lease-expired", row.stale_flags) + self.assertIn("active-lease-past-expiry", row.stale_flags) + + +class TestContaminationRedaction(unittest.TestCase): + """B2: marker payload text is redacted at the display boundary.""" + + def _marker(self, payload: dict) -> ContaminationMarker: + return _inspect_contamination( + "runtime_recovery_contamination", + remote="prgs", + inspect=lambda **_k: { + "on_disk": True, + "has_payload": True, + "summary": "", + }, + load=lambda **_k: payload, + ) + + def test_home_paths_are_collapsed(self): + home = os.path.expanduser("~") + marker = self._marker( + {"command_summary": f"pkill -f {home}/Development/Gitea-Tools/x.py"} + ) + self.assertNotIn(home, marker.command_summary) + self.assertIn("~/Development/Gitea-Tools/x.py", marker.command_summary) + + def test_secrets_missed_by_the_write_time_denylist_are_redacted(self): + # Each of these was verified in review to survive + # stable_branch_push_guard.redact_command untouched. + cases = ( + ("curl -H 'X-Api-Key: SUPERSECRET123' https://example.invalid", "SUPERSECRET123"), + ("cmd --password hunter2 origin master", "hunter2"), + ("PRIVATE_KEY=abc123 python deploy.py", "abc123"), + ("fetch https://user:pw@example.invalid/x.git", "user:pw"), + ) + for raw, secret in cases: + with self.subTest(raw=raw): + marker = self._marker({"command_summary": raw}) + self.assertNotIn(secret, marker.command_summary) + self.assertIn("[redacted]", marker.command_summary) + + def test_command_summary_is_redacted_not_removed(self): + # It is legitimate #630 evidence: the operator must still see which + # daemon was killed. + marker = self._marker( + { + "command_summary": "pkill -f gitea_mcp_server.py", + "reason_class": "manual_daemon_kill", + "session_id": "prgs-author-111-clean", + "role": "author", + } + ) + self.assertIn("pkill -f gitea_mcp_server.py", marker.command_summary) + self.assertEqual(marker.reason_class, "manual_daemon_kill") + self.assertEqual(marker.session_id, "prgs-author-111-clean") + self.assertEqual(marker.role, "author") + + def test_rendered_page_exposes_no_home_path_from_a_marker(self): + home = os.path.expanduser("~") + marker = self._marker( + { + "command_summary": f"pkill -f {home}/Development/Gitea-Tools/x.py", + "reason_class": "manual_daemon_kill", + } + ) + inventory = _inventory(sessions=(_clean_session(),)) + html = render_sessions_page( + SessionViewSnapshot( + runtime=_runtime(), + inventory=inventory, + sessions=_build_session_rows(inventory, (marker,)), + contamination_markers=(marker,), + ) + ) + self.assertIn("Contamination markers", html) + self.assertNotIn(home, html) + + +class TestSessionsPageEscaping(unittest.TestCase): + """Hostile values from every rendered source stay inert (N2).""" + + HOSTILE = '' + + def test_hostile_session_and_marker_values_are_escaped(self): + session = dict(_clean_session()) + session["session_id"] = f"sid-{self.HOSTILE}" + session["role"] = self.HOSTILE + session["profile"] = self.HOSTILE + session["namespace"] = self.HOSTILE + session["status"] = self.HOSTILE + inventory = _inventory( + sessions=(session,), + leases=( + { + "lease_id": self.HOSTILE, + "session_id": session["session_id"], + "status": "active", + "expired": False, + "work_kind": self.HOSTILE, + "work_number": 641, + }, + ), + locks=( + { + "issue_number": 641, + "worktree_path": self.HOSTILE, + }, + ), + ) + marker = ContaminationMarker( + kind="runtime_recovery_contamination", + on_disk=True, + has_payload=True, + summary=self.HOSTILE, + reason_class=self.HOSTILE, + session_id=session["session_id"], + role=self.HOSTILE, + command_summary=self.HOSTILE, + cleared=False, + ) + html = render_sessions_page( + SessionViewSnapshot( + runtime=_runtime(), + inventory=inventory, + sessions=_build_session_rows(inventory, (marker,)), + contamination_markers=(marker,), + ) + ) + self.assertNotIn("