Scaled-Tech-Consulting/Gitea-Tools
2
0
Fork 0
Code Issues 91 Pull Requests 3 Actions Packages Projects Releases Wiki Activity

fix(webui): authority-aware ownership + redacted contamination text (#641)

Browse Source 
Addresses the two blockers from the PR #898 review at a81db754.

B1 - degraded ownership inventory was rendered as affirmative absence.
_build_session_rows read the sessions/leases/locks sections without
consulting their status, so a session row emitted lease_ids=() and
worktree_paths=() whether the session genuinely held nothing or the
lease store simply could not be read. The renderer printed both as
"none" and "unbound", contradicting the ownership_authority_complete
invariant documented on InventorySnapshot.

SessionRow now carries lease_authority and worktree_authority. A
worktree binding is correlated through lease work numbers, so it is
unproven when either the leases or the locks section fails to read --
this covers the narrow variant where locks hold real worktree paths but
a degraded leases section leaves work_numbers empty. The renderer emits
"unknown (inventory <status>)" with an authority-unproven badge instead
of none/unbound, the card names the unreadable sections, and an empty
session list from an unreadable sessions section no longer reads as
"no sessions recorded". snapshot_to_dict exports
ownership_authority_complete, ownership_section_status, and per-row
lease_authority / worktree_authority so /api/sessions consumers can
distinguish the two cases.

B2 - contamination payload strings bypassed redaction.
_inspect_contamination copied command_summary, session_id, role and
reason_class out of the marker payload with only str(), while every
inventory-sourced field on the same page arrives through
webui.inventory.scrub(). The write-time redactor
stable_branch_push_guard.redact_command is a narrow denylist that leaves
absolute $HOME paths, -H 'X-Api-Key: <value>', --password <value>, and
PRIVATE_KEY=<value> intact, and this is the first web surface to render
command_summary at all.

Adds webui.inventory.scrub_text(), which collapses $HOME and redacts
credential-shaped tokens and URL userinfo anywhere inside a string rather
than only at its start, and routes the marker payload through it. scrub()
and every existing caller are untouched. The command_summary field is
kept: it is the #630 evidence naming which daemon was killed. The module
docstring claiming absolute paths were already collapsed is corrected.

Also: removes the locks_by_session_hint dead loop and its discard (N1),
adds the missing trailing newline to webui/runtime_views.py (N4), drops
an unused dataclasses.field import, and documents both honesty rules in
docs/webui-local-dev.md.

Tests: tests/test_webui_sessions_view.py grows from 12 to 26 cases,
covering degraded and unavailable ownership sections in both the HTML and
JSON paths, the locks-readable/leases-degraded variant, a guard against
over-correcting clean inventory into "unknown", the previously untested
expired-lease flag, HTML escaping of hostile values in clean and degraded
renders, and each secret class the write-time denylist misses. The
STATUS_UNAVAILABLE import that was present but unused is now exercised.

Validation, from the issue worktree with venv/bin/python (Python 3.14.5,
pytest 9.1.1):

  pytest tests/test_webui_sessions_view.py tests/test_webui_*.py -q
    -> 512 passed, 376 subtests (was 498 / 372; +14 new tests)
  pytest tests/test_issue_854_semantic_container_exclusion.py -q
    -> 13 passed, 8 subtests
  13-file runtime/health/inventory/restart set
    -> 1 failed, 223 passed; the single failure is
       test_runtime_clarity.py::TestRuntimeClarity::
       test_activate_profile_succeeds_when_enabled, the identical test and
       assertion the reviewer recorded on master at 7af40fb5, so it is
       baseline-equivalent and not introduced here.

Co-Authored-By: Claude Opus 4.8 (1M context) <[email protected]>
This commit is contained in:
sysadmin
2026-07-25 01:25:28 -04:00
co-authored by Claude Opus 4.8
parent a81db75402
commit 1ca2b50406
6 changed files with 556 additions and 41 deletions
Download Patch File Download Diff File Expand all files Collapse all files
docs/webui-local-dev.md
+20
View File
@@ -307,6 +307,26 @@ reconnect/operator restart docs (`docs/mcp-namespace-eof-recovery.md`,
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 <status>)` 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)
`/system-health` renders the same snapshot the `/api/v1/system/health` API
tests/test_webui_sessions_view.py
+287 -5
View File
@@ -3,10 +3,16 @@
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
@@ -22,6 +28,7 @@ from webui.inventory import (
AUTHORITY_FILESYSTEM,
InventorySection,
InventorySnapshot,
STATUS_DEGRADED,
STATUS_OK,
STATUS_UNAVAILABLE,
)
@@ -32,6 +39,7 @@ from webui.session_loader import (
SessionRow,
SessionViewSnapshot,
_build_session_rows,
_inspect_contamination,
load_session_view_snapshot,
snapshot_to_dict,
)
@@ -77,36 +85,43 @@ def _inventory(
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_OK,
status=_status("sessions"),
items=sessions,
),
InventorySection(
name="leases",
authority=AUTHORITY_CONTROL_PLANE_DB,
status=STATUS_OK,
status=_status("leases"),
items=leases,
),
InventorySection(
name="locks",
authority=AUTHORITY_FILESYSTEM,
status=STATUS_OK,
status=_status("locks"),
items=locks,
),
InventorySection(
name="worktrees",
authority=AUTHORITY_FILESYSTEM,
status=STATUS_OK,
status=_status("worktrees"),
items=worktrees,
),
InventorySection(
name="namespaces",
authority=AUTHORITY_FILESYSTEM,
status=STATUS_OK,
status=_status("namespaces"),
items=namespaces
or (
{
@@ -453,5 +468,272 @@ class TestSessionsRoutes(unittest.TestCase):
)
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:[email protected]/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 = '<script>alert("xss")</script>'
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("<script>", html)
self.assertNotIn('alert("xss")', html)
self.assertIn("&lt;script&gt;", html)
def test_hostile_values_in_a_degraded_render_are_escaped(self):
inventory = _inventory(
sessions=(dict(_clean_session(), session_id=f"sid-{self.HOSTILE}"),),
statuses={"leases": STATUS_UNAVAILABLE, "locks": STATUS_DEGRADED},
)
html = render_sessions_page(
SessionViewSnapshot(
runtime=_runtime(),
inventory=inventory,
sessions=_build_session_rows(inventory, contamination=()),
contamination_markers=(),
)
)
self.assertNotIn("<script>", html)
self.assertIn("&lt;script&gt;", html)
self.assertIn("unknown (inventory", html)
if __name__ == "__main__":
unittest.main()
webui/inventory.py
+65
View File
@@ -76,6 +76,35 @@ _CREDENTIAL_KEY_RE = re.compile(
)
_REDACTED = "[redacted]"
#: Credential-shaped *name* as it appears inside a free-form command line. This
#: is deliberately broader than :data:`_CREDENTIAL_KEY_RE` — it also matches a
#: bare ``key`` component, so ``PRIVATE_KEY=`` is caught. Over-redacting a
#: displayed string is safe; under-redacting one is not.
_TEXT_CREDENTIAL_NAME = (
r"[A-Za-z0-9_.\-]*"
r"(?:token|secret|password|passwd|key|authorization|bearer|credential)"
r"[A-Za-z0-9_.\-]*"
)
#: A value following such a name: single-quoted, double-quoted, or bare. The
#: bare form stops at a quote so an enclosing quote survives the redaction.
_TEXT_CREDENTIAL_VALUE = r"'[^']*'|\"[^\"]*\"|[^\s'\"]+"
_TEXT_CREDENTIAL_FLAG_RE = re.compile(
rf"(?P<key>(?<![\w\-])--?{_TEXT_CREDENTIAL_NAME})"
rf"(?P<sep>[=\s]+)"
rf"(?P<value>{_TEXT_CREDENTIAL_VALUE})",
re.IGNORECASE,
)
_TEXT_CREDENTIAL_ASSIGN_RE = re.compile(
rf"(?P<key>(?<![\w\-]){_TEXT_CREDENTIAL_NAME})"
rf"(?P<sep>\s*[:=]\s*)"
rf"(?P<value>{_TEXT_CREDENTIAL_VALUE})",
re.IGNORECASE,
)
_TEXT_URL_USERINFO_RE = re.compile(
r"(?P<scheme>\b[A-Za-z][A-Za-z0-9+.\-]*://)[^\s/@]+@"
)
@dataclass(frozen=True)
class InventorySection:
@@ -225,6 +254,42 @@ def scrub(value: Any, *, key: str | None = None) -> Any:
return repr(value)
def collapse_home(text: str) -> str:
"""Collapse every ``$HOME`` occurrence *inside* a string, not just a prefix."""
home = os.path.expanduser("~")
if not home or home == "/":
return text
return text.replace(home, "~")
def scrub_text(value: Any) -> Any:
"""Redact a free-form text blob such as a recorded command line.
:func:`scrub` keys off structured field *names* and whole-value prefixes,
which is right for inventory records but blind to a secret embedded in the
middle of a sentence. This collapses ``$HOME`` and redacts credential-shaped
tokens and URL userinfo *anywhere* in the string, so operator-supplied text
rendered verbatim — contamination ``command_summary`` (#630) above all — is
held to the same standard as every other field on the page.
Returns ``None`` unchanged so callers can keep "absent" distinct from "".
"""
if value is None:
return None
text = value if isinstance(value, str) else str(value)
text = collapse_home(text)
text = _TEXT_URL_USERINFO_RE.sub(
lambda m: f"{m.group('scheme')}{_REDACTED}@", text
)
text = _TEXT_CREDENTIAL_FLAG_RE.sub(
lambda m: f"{m.group('key')}{m.group('sep')}{_REDACTED}", text
)
text = _TEXT_CREDENTIAL_ASSIGN_RE.sub(
lambda m: f"{m.group('key')}{m.group('sep')}{_REDACTED}", text
)
return text
# ── control-plane database (read-only) ───────────────────────────────────────
webui/session_loader.py
+105 -16
View File
@@ -12,18 +12,30 @@ Sources:
and collision signals from the control-plane DB + filesystem.
* :mod:`mcp_session_state` — durable contamination markers (inspect only).
Secrets are never read. Absolute paths are collapsed by inventory redaction.
Secrets are never read. Inventory-sourced values arrive already redacted by
:func:`webui.inventory.scrub`; free-form marker text this module loads itself is
put through :func:`webui.inventory.scrub_text`, which collapses ``$HOME`` and
redacts credential-shaped tokens *inside* a string rather than only at its start.
Ownership columns are authority-aware. A lease or lock section that could not be
read renders as ``unknown``, never as ``none`` or ``unbound``: a lease the reader
could not load is not an absent lease (see
:attr:`webui.inventory.InventorySnapshot.ownership_authority_complete`).
"""
from __future__ import annotations
from dataclasses import dataclass, field
from dataclasses import dataclass
from typing import Any, Callable
import mcp_session_state
from webui.inventory import (
OWNERSHIP_SECTIONS,
STATUS_OK,
InventorySection,
InventorySnapshot,
load_inventory_snapshot,
scrub_text,
snapshot_to_dict as inventory_snapshot_to_dict,
)
from webui.runtime_health import (
@@ -61,6 +73,29 @@ _CONTAMINATION_KINDS: tuple[str, ...] = (
mcp_session_state.KIND_STABLE_BRANCH_CONTAMINATION,
)
#: Status recorded on a row when the backing inventory section is absent
#: entirely — distinct from a section that reported itself degraded.
AUTHORITY_MISSING = "missing"
def _section_status(section: InventorySection | None) -> str:
"""Status of an ownership section, treating an absent section as missing."""
if section is None:
return AUTHORITY_MISSING
return section.status
def _combined_authority(*statuses: str) -> str:
"""Worst status of the sections a derived column depends on.
A column proved from two sections is only trustworthy when *both* read
cleanly, so the first non-``ok`` status wins.
"""
for status in statuses:
if status != STATUS_OK:
return status
return STATUS_OK
@dataclass(frozen=True)
class ContaminationMarker:
@@ -109,6 +144,17 @@ class SessionRow:
worktree_paths: tuple[str, ...] = ()
stale_flags: tuple[str, ...] = ()
contamination_flags: tuple[str, ...] = ()
#: Status of the section backing ``lease_ids``/``work_refs``. While this is
#: not ``ok`` those tuples mean "could not be read", never "none held".
lease_authority: str = STATUS_OK
#: Worst status across the sections backing ``worktree_paths`` (locks are
#: correlated through lease work numbers, so both must read cleanly).
worktree_authority: str = STATUS_OK
@property
def ownership_authority_complete(self) -> bool:
"""True only when this row's ownership columns are provable."""
return self.lease_authority == STATUS_OK and self.worktree_authority == STATUS_OK
def to_dict(self) -> dict[str, Any]:
return {
@@ -128,6 +174,18 @@ class SessionRow:
"contamination_flags": list(self.contamination_flags),
"is_stale": bool(self.stale_flags),
"is_contaminated": bool(self.contamination_flags),
"lease_authority": self.lease_authority,
"worktree_authority": self.worktree_authority,
"lease_authority_complete": self.lease_authority == STATUS_OK,
"worktree_authority_complete": self.worktree_authority == STATUS_OK,
"ownership_authority_complete": self.ownership_authority_complete,
"ownership_note": (
"Lease and worktree columns are proved from sections that read "
"cleanly."
if self.ownership_authority_complete
else "An ownership source could not be read; empty lease_ids or "
"worktree_paths on this row mean unknown, not unowned."
),
}
@@ -154,6 +212,19 @@ class SessionViewSnapshot:
def active_contamination(self) -> tuple[ContaminationMarker, ...]:
return tuple(m for m in self.contamination_markers if m.to_dict()["active"])
@property
def ownership_authority_complete(self) -> bool:
"""False while any ownership section is degraded, absent, or unavailable."""
return self.inventory.ownership_authority_complete
@property
def ownership_section_status(self) -> dict[str, str]:
"""Per-section status for the three ownership-bearing sections."""
return {
name: _section_status(self.inventory.section(name))
for name in OWNERSHIP_SECTIONS
}
def _inspect_contamination(
kind: str,
@@ -180,7 +251,7 @@ def _inspect_contamination(
role = None
command_summary = None
cleared = False
summary = str(envelope.get("summary") or "")
summary = scrub_text(str(envelope.get("summary") or ""))
if envelope.get("on_disk") and envelope.get("has_payload"):
try:
@@ -194,20 +265,25 @@ def _inspect_contamination(
command_summary = payload.get("command_summary") or payload.get("detail")
cleared = bool(payload.get("cleared_by_reconciler"))
if not summary:
summary = (
summary = scrub_text(
f"{kind}: {reason_class or 'present'}"
+ (" (cleared)" if cleared else "")
)
# Marker payloads are operator-supplied free text and are the one thing on
# this page that does not arrive through inventory scrubbing. The write-time
# redactor is a narrow denylist, so redact again at the display boundary:
# it leaves $HOME paths, `-H 'X-Api-Key: …'`, `--password …`, and
# `PRIVATE_KEY=…` intact. The field itself stays — it is #630 evidence.
return ContaminationMarker(
kind=kind,
on_disk=bool(envelope.get("on_disk")),
has_payload=bool(envelope.get("has_payload")),
summary=summary or f"{kind}: not present",
reason_class=str(reason_class) if reason_class else None,
session_id=str(session_id) if session_id else None,
role=str(role) if role else None,
command_summary=str(command_summary) if command_summary else None,
reason_class=scrub_text(str(reason_class)) if reason_class else None,
session_id=scrub_text(str(session_id)) if session_id else None,
role=scrub_text(str(role)) if role else None,
command_summary=scrub_text(str(command_summary)) if command_summary else None,
cleared=cleared,
)
@@ -232,18 +308,21 @@ def _build_session_rows(
leases_section = inventory.section("leases")
locks_section = inventory.section("locks")
# Ownership columns may only assert absence when their source read cleanly.
lease_authority = _section_status(leases_section)
# Locks carry claimant profile/username, not control-plane session ids, so a
# worktree binding is correlated through lease work numbers: it depends on
# the locks *and* the leases section.
worktree_authority = _combined_authority(
lease_authority, _section_status(locks_section)
)
leases_by_session: dict[str, list[dict[str, Any]]] = {}
for lease in (leases_section.items if leases_section else ()):
sid = str(lease.get("session_id") or "")
if sid:
leases_by_session.setdefault(sid, []).append(lease)
locks_by_session_hint: dict[str, list[dict[str, Any]]] = {}
for lock in (locks_section.items if locks_section else ()):
# Locks carry claimant profile/username, not control-plane session ids.
# Correlate later by matching lease work_number ↔ lock issue_number.
pass
active_markers = [m for m in contamination if m.to_dict()["active"]]
marker_session_ids = {
m.session_id for m in active_markers if m.session_id
@@ -324,11 +403,11 @@ def _build_session_rows(
worktree_paths=tuple(worktree_paths),
stale_flags=tuple(dict.fromkeys(stale_flags)),
contamination_flags=tuple(dict.fromkeys(contamination_flags)),
lease_authority=lease_authority,
worktree_authority=worktree_authority,
)
)
# Silence unused variable for the locks_by_session_hint placeholder path.
_ = locks_by_session_hint
return tuple(rows)
@@ -415,6 +494,16 @@ def snapshot_to_dict(snapshot: SessionViewSnapshot) -> dict[str, Any]:
"stale": snapshot.stale_session_count,
"contaminated": snapshot.contaminated_session_count,
},
"ownership_authority_complete": snapshot.ownership_authority_complete,
"ownership_section_status": snapshot.ownership_section_status,
"ownership_note": (
"Every ownership source read cleanly; a session with no lease and no "
"worktree path genuinely holds neither."
if snapshot.ownership_authority_complete
else "An ownership source is degraded or unavailable. Empty lease_ids "
"and worktree_paths mean unknown, not unowned; consult each row's "
"lease_authority and worktree_authority."
),
"contamination_markers": [
marker.to_dict() for marker in snapshot.contamination_markers
],
webui/session_views.py
+78 -19
View File
@@ -11,6 +11,7 @@ from __future__ import annotations
from html import escape
from typing import Sequence
from webui.inventory import STATUS_OK
from webui.layout import render_page
from webui.session_loader import (
ContaminationMarker,
@@ -29,6 +30,19 @@ def _flags(flags: Sequence[str], *, css: str) -> str:
return " ".join(_badge(flag, css) for flag in flags)
def _unproven_cell(status: str) -> str:
"""Render an ownership column whose backing inventory section failed to read.
Never "none" and never "unbound": an unreadable source proves nothing about
ownership, and claiming otherwise is the exact failure the
``ownership_authority_complete`` invariant exists to prevent.
"""
return (
f'<span class="muted">unknown (inventory {escape(status)})</span><br>'
f'{_badge("authority unproven", "badge-health-degraded")}'
)
def _runtime_banner(snapshot: SessionViewSnapshot) -> str:
runtime = snapshot.runtime
stale = runtime.stale_runtime_warning
@@ -73,15 +87,37 @@ def _summary_bar(snapshot: SessionViewSnapshot) -> str:
contaminated = snapshot.contaminated_session_count
active_markers = len(snapshot.active_contamination)
inv_status = snapshot.inventory.status
authority_complete = snapshot.ownership_authority_complete
authority_text = "complete" if authority_complete else "incomplete"
authority_css = "badge-health-ok" if authority_complete else "badge-health-degraded"
return f"""<div class="health-card" style="display:flex; flex-wrap:wrap; gap:1rem; align-items:center;">
<div><strong>Sessions:</strong> <span class="badge badge-health-ok">{total}</span></div>
<div><strong>Stale:</strong> <span class="badge badge-stale">{stale}</span></div>
<div><strong>Contaminated:</strong> <span class="badge badge-blocked">{contaminated}</span></div>
<div><strong>Active markers:</strong> <span class="badge badge-health-unproven">{active_markers}</span></div>
<div><strong>Inventory:</strong> <span class="badge badge-health-skipped">{escape(inv_status)}</span></div>
<div><strong>Ownership authority:</strong> <span class="badge {authority_css}">{authority_text}</span></div>
</div>"""
def _ownership_caveat(snapshot: SessionViewSnapshot) -> str:
"""Name the unreadable ownership sections, or render nothing when all read."""
if snapshot.ownership_authority_complete:
return ""
degraded = ", ".join(
f"{name}: {status}"
for name, status in snapshot.ownership_section_status.items()
if status != STATUS_OK
)
return (
'<div class="health-card health-stale">'
"<strong>Ownership authority incomplete:</strong> "
f"{escape(degraded)}. Columns marked <em>unknown</em> could not be read. "
"No session below may be treated as holding no lease or no worktree "
"binding — absence of evidence is not evidence of absence.</div>"
)
def _render_session_row(row: SessionRow) -> str:
pid = "—" if row.pid is None else str(row.pid)
pid_alive = "—" if row.pid_alive is None else ("alive" if row.pid_alive else "dead")
@@ -90,21 +126,33 @@ def _render_session_row(row: SessionRow) -> str:
if row.pid_alive is True
else ("badge-blocked" if row.pid_alive is False else "badge-health-skipped")
)
leases = (
", ".join(f"<code>{escape(lid)}</code>" for lid in row.lease_ids)
if row.lease_ids
else '<span class="muted">none</span>'
)
work = (
", ".join(escape(ref) for ref in row.work_refs)
if row.work_refs
else '<span class="muted">—</span>'
)
worktrees = (
"<br>".join(f"<code>{escape(path)}</code>" for path in row.worktree_paths)
if row.worktree_paths
else '<span class="muted">unbound</span>'
)
# An empty tuple only means "holds none" when its source read cleanly.
if row.lease_authority != STATUS_OK:
lease_cell = _unproven_cell(row.lease_authority)
else:
leases = (
", ".join(f"<code>{escape(lid)}</code>" for lid in row.lease_ids)
if row.lease_ids
else '<span class="muted">none</span>'
)
work = (
", ".join(escape(ref) for ref in row.work_refs)
if row.work_refs
else '<span class="muted">—</span>'
)
lease_cell = (
f'{leases}<div class="muted" style="font-size:0.82rem; '
f'margin-top:0.2rem;">{work}</div>'
)
if row.worktree_authority != STATUS_OK:
worktrees = _unproven_cell(row.worktree_authority)
else:
worktrees = (
"<br>".join(f"<code>{escape(path)}</code>" for path in row.worktree_paths)
if row.worktree_paths
else '<span class="muted">unbound</span>'
)
return f"""<tr>
<td><code>{escape(row.session_id)}</code></td>
<td>
@@ -116,15 +164,24 @@ def _render_session_row(row: SessionRow) -> str:
{_badge(pid_alive, pid_css)}
</td>
<td>{escape(str(row.status or "—"))}<div class="muted" style="font-size:0.82rem;">{escape(str(row.last_heartbeat_at or ""))}</div></td>
<td>{leases}<div class="muted" style="font-size:0.82rem; margin-top:0.2rem;">{work}</div></td>
<td>{lease_cell}</td>
<td>{worktrees}</td>
<td>{_flags(row.stale_flags, css="badge-stale")}</td>
<td>{_flags(row.contamination_flags, css="badge-blocked")}</td>
</tr>"""
def _sessions_table(rows: Sequence[SessionRow]) -> str:
def _sessions_table(snapshot: SessionViewSnapshot) -> str:
rows: Sequence[SessionRow] = snapshot.sessions
if not rows:
sessions_status = snapshot.ownership_section_status.get("sessions", STATUS_OK)
if sessions_status != STATUS_OK:
return (
'<p class="muted">Session inventory is '
f"<strong>{escape(sessions_status)}</strong> — the session list "
"could not be read. This is not evidence that no sessions "
"exist.</p>"
)
return (
'<p class="muted">No control-plane sessions recorded. Inventory may '
"be unavailable, or no MCP workers have registered yet.</p>"
@@ -329,8 +386,10 @@ def render_sessions_page(snapshot: SessionViewSnapshot) -> str:
<div class="prompt-card">
<h3>Sessions</h3>
<p class="muted">Control-plane sessions correlated with leases and worktree bindings.
Stale and contamination flags are fail-soft: absence of a marker is not proof of cleanliness when inventory is degraded.</p>
{_sessions_table(snapshot.sessions)}
Stale and contamination flags are fail-soft: absence of a marker is not proof of cleanliness when inventory is degraded.
Lease and worktree columns read <em>unknown (inventory …)</em> when their source could not be loaded.</p>
{_ownership_caveat(snapshot)}
{_sessions_table(snapshot)}
</div>
{_namespaces_section(snapshot)}