Gitea-Tools/webui/session_views.py

"""HTML views for the Runtime and session view (Phase 1, #641).

Read-only composition of runtime health (#430) and inventory sessions /
namespaces / worktrees (#636). Surfaces stale and contamination indicators
when detectable. Recovery links point only at sanctioned reconnect/restart
docs — never at manual process kill (#630).
"""

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,
    SessionRow,
    SessionViewSnapshot,
)


def _badge(text: str, css: str) -> str:
    return f'<span class="badge {css}">{escape(text)}</span>'


def _flags(flags: Sequence[str], *, css: str) -> str:
    if not flags:
        return '<span class="muted">—</span>'
    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
    stale_html = ""
    if stale:
        stale_html = (
            f'<div class="health-card health-stale" style="margin-top:0.75rem;">'
            f"<strong>Stale runtime:</strong> {escape(stale)}</div>"
        )
    identity = runtime.authenticated_username or "unresolved"
    if runtime.identity_error:
        identity = f"unresolved ({runtime.identity_error})"

    return f"""<div class="health-card">
  <h3>Runtime context</h3>
  <table class="detail">
    <tr><th>Profile</th><td><code>{escape(runtime.profile_name)}</code></td></tr>
    <tr><th>Role kind</th><td>{escape(runtime.role_kind)}</td></tr>
    <tr><th>Identity</th><td>{escape(str(identity))}</td></tr>
    <tr><th>Remote / host</th>
      <td><code>{escape(runtime.remote)}</code> · <code>{escape(runtime.host)}</code></td>
    </tr>
    <tr><th>Local HEAD</th>
      <td><code>{escape(runtime.repo_sha or "unknown")}</code></td>
    </tr>
    <tr><th>Remote master</th>
      <td><code>{escape(runtime.remote_master_sha or "unknown")}</code></td>
    </tr>
    <tr><th>Commits behind</th>
      <td>{escape(str(runtime.commits_behind_master if runtime.commits_behind_master is not None else "unknown"))}</td>
    </tr>
  </table>
  <p class="muted">Full runtime detail: <a href="/runtime">/runtime</a> ·
  Inventory API: <a href="/api/v1/inventory"><code>/api/v1/inventory</code></a></p>
  {stale_html}
</div>"""


def _summary_bar(snapshot: SessionViewSnapshot) -> str:
    total = len(snapshot.sessions)
    stale = snapshot.stale_session_count
    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")
    pid_css = (
        "badge-health-ok"
        if row.pid_alive is True
        else ("badge-blocked" if row.pid_alive is False else "badge-health-skipped")
    )
    # 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>
    <div><code>{escape(str(row.role or "—"))}</code> / <code>{escape(str(row.profile or "—"))}</code></div>
    <div class="muted" style="font-size:0.82rem;">ns: <code>{escape(str(row.namespace or "—"))}</code></div>
  </td>
  <td>
    <code>{escape(pid)}</code>
    {_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>{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(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>"
        )
    body = "".join(_render_session_row(row) for row in rows)
    return f"""<table class="registry">
  <thead>
    <tr>
      <th>Session</th>
      <th>Role / profile / namespace</th>
      <th>PID</th>
      <th>Status</th>
      <th>Leases / work</th>
      <th>Worktree binding</th>
      <th>Stale</th>
      <th>Contamination</th>
    </tr>
  </thead>
  <tbody>
    {body}
  </tbody>
</table>"""


def _namespaces_section(snapshot: SessionViewSnapshot) -> str:
    section = snapshot.inventory.section("namespaces")
    if section is None:
        return (
            '<div class="prompt-card"><h3>Namespaces</h3>'
            '<p class="muted">Namespaces section not loaded.</p></div>'
        )
    if not section.ok:
        return f"""<div class="prompt-card">
  <h3>Namespaces {_badge(section.status, "badge-health-degraded")}</h3>
  <p class="muted">{escape(section.reason or "unavailable")}</p>
</div>"""

    rows = []
    for item in section.items:
        caps = item.get("capability_summary") or {}
        cap_bits = ", ".join(
            name for name, ok in sorted(caps.items()) if ok
        ) or "none"
        rows.append(
            "<tr>"
            f"<td><code>{escape(str(item.get('mcp_namespace') or '—'))}</code></td>"
            f"<td><code>{escape(str(item.get('profile_name') or '—'))}</code></td>"
            f"<td>{escape(str(item.get('role') or '—'))}</td>"
            f"<td>{escape(cap_bits)}</td>"
            f"<td>{'yes' if item.get('active') else 'no'}</td>"
            "</tr>"
        )
    reason = (
        f'<p class="muted">{escape(section.reason)}</p>'
        if section.reason
        else ""
    )
    return f"""<div class="prompt-card">
  <h3>Namespaces / capabilities</h3>
  {reason}
  <table class="registry">
    <thead>
      <tr>
        <th>Namespace</th>
        <th>Profile</th>
        <th>Role</th>
        <th>Capabilities</th>
        <th>Active in process</th>
      </tr>
    </thead>
    <tbody>
      {"".join(rows) if rows else '<tr><td colspan="5" class="muted">No namespace rows.</td></tr>'}
    </tbody>
  </table>
</div>"""


def _worktrees_section(snapshot: SessionViewSnapshot) -> str:
    section = snapshot.inventory.section("worktrees")
    if section is None:
        return ""
    if not section.ok and not section.items:
        return f"""<div class="prompt-card">
  <h3>Worktrees {_badge(section.status, "badge-health-degraded")}</h3>
  <p class="muted">{escape(section.reason or "unavailable")}</p>
</div>"""

    rows = []
    for item in section.items[:50]:
        rows.append(
            "<tr>"
            f"<td><code>{escape(str(item.get('rel_path') or item.get('path') or '—'))}</code></td>"
            f"<td><code>{escape(str(item.get('branch') or '—'))}</code></td>"
            f"<td>{escape(str(item.get('classification') or '—'))}</td>"
            f"<td>{'yes' if item.get('registered_worktree') else 'no'}</td>"
            f"<td>{'dirty' if item.get('dirty') else 'clean'}</td>"
            "</tr>"
        )
    more = ""
    if len(section.items) > 50:
        more = f'<p class="muted">Showing 50 of {len(section.items)}. Full list: <a href="/worktrees">/worktrees</a>.</p>'
    return f"""<div class="prompt-card">
  <h3>Worktree bindings</h3>
  <p class="muted">Registered issue worktrees under <code>branches/</code>. Hygiene detail: <a href="/worktrees">/worktrees</a>.</p>
  <table class="registry">
    <thead>
      <tr>
        <th>Path</th>
        <th>Branch</th>
        <th>Classification</th>
        <th>Registered</th>
        <th>State</th>
      </tr>
    </thead>
    <tbody>
      {"".join(rows) if rows else '<tr><td colspan="5" class="muted">No worktrees recorded.</td></tr>'}
    </tbody>
  </table>
  {more}
</div>"""


def _contamination_section(markers: Sequence[ContaminationMarker]) -> str:
    if not markers:
        return (
            '<div class="prompt-card"><h3>Contamination markers</h3>'
            '<p class="muted">No contamination kinds inspected.</p></div>'
        )
    rows = []
    for marker in markers:
        active = marker.to_dict()["active"]
        status = "ACTIVE" if active else ("cleared" if marker.cleared else "absent")
        css = "badge-blocked" if active else "badge-health-ok"
        rows.append(
            "<tr>"
            f"<td><code>{escape(marker.kind)}</code></td>"
            f"<td>{_badge(status, css)}</td>"
            f"<td>{escape(marker.reason_class or '—')}</td>"
            f"<td><code>{escape(marker.session_id or '—')}</code></td>"
            f"<td>{escape(marker.command_summary or marker.summary)}</td>"
            "</tr>"
        )
    return f"""<div class="prompt-card">
  <h3>Contamination markers (#630 / #671)</h3>
  <p class="muted">Durable markers only — never silent when present. Clearance is reconciler-only.</p>
  <table class="registry">
    <thead>
      <tr>
        <th>Kind</th>
        <th>State</th>
        <th>Reason class</th>
        <th>Session</th>
        <th>Summary</th>
      </tr>
    </thead>
    <tbody>
      {"".join(rows)}
    </tbody>
  </table>
</div>"""


def _recovery_section(snapshot: SessionViewSnapshot) -> str:
    items = []
    for doc in snapshot.recovery_docs:
        items.append(
            "<li>"
            f"<code>{escape(doc['path'])}</code> — "
            f"<strong>{escape(doc['label'])}</strong>: {escape(doc['note'])}"
            "</li>"
        )
    return f"""<div class="prompt-card">
  <h3>Sanctioned recovery (read-only)</h3>
  <p class="muted">This view does <strong>not</strong> restart, kill, or take over sessions.
  Manual <code>pkill</code> / <code>kill</code> of MCP daemons is contamination (#630), not recovery.</p>
  <ul class="reasons">
    {"".join(items)}
    <li>Prefer IDE/client reconnect (<code>/mcp reconnect</code>) or an operator-owned restart recorded in the restart inventory.</li>
  </ul>
</div>"""


def render_sessions_page(snapshot: SessionViewSnapshot) -> str:
    """Render the full HTML body for the runtime/session view."""
    error_block = ""
    if snapshot.fetch_error:
        error_block = (
            f'<div class="health-card health-stale"><strong>Partial load:</strong> '
            f"{escape(snapshot.fetch_error)}</div>"
        )
    if snapshot.runtime.fetch_error:
        error_block += (
            f'<div class="health-card health-stale"><strong>Runtime note:</strong> '
            f"{escape(snapshot.runtime.fetch_error)}</div>"
        )

    body = f"""
{error_block}
{_runtime_banner(snapshot)}
{_summary_bar(snapshot)}

<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.
  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)}
{_worktrees_section(snapshot)}
{_contamination_section(snapshot.contamination_markers)}
{_recovery_section(snapshot)}
"""
    return render_page(title="Sessions", body_html=f"""<h2>Runtime and sessions</h2>
<p class="meta">Phase 1 read-only view (#641). Combines runtime health (#430) with
unified inventory sessions/namespaces/worktrees (#636). No restart or session-takeover controls.</p>
{body}""")