feat(webui): evolve application shell for Phase 1 console IA (Closes #638)

Introduce a single nav-config module (webui/nav.py) driving grouped
navigation across the epic #631 Phase 1 information architecture:
Health, Traffic, Runtime/Sessions, Projects, Inventory, Timeline,
Policy (placeholder), and Insights (placeholder). The shell header now
carries a read-only environment badge (local/remote from WEBUI_HOST), a
mode: read-only badge, and a Docs link. Home summarizes the console
purpose, lists the Phase 1 surfaces by group, and links the MVP legacy
pages.

Not-yet-implemented surfaces (/sessions, /inventory, /timeline,
/policy, /insights) resolve to graceful read-only stub pages rather than
404s; their backing views land in later child issues of #631 (inventory
surfaces backed by #636). Mutating methods on stub routes still fail
closed with read-only-mvp. No privileged action controls are added.

Adds tests/test_webui_shell.py covering nav groups, badge presence,
environment classification, docs link, stub routes (200 + read-only),
and home content. Updates docs/webui-local-dev.md with the new routes
and a Phase 1 shell section.

Co-Authored-By: Claude Opus 4.8 (1M context) <[email protected]>

anti_stomp_preflight.py

@@ -87,6 +87,7 @@ MUTATION_TASKS = frozenset({
     "edit_pr",
     "commit_files",
     "gitea_commit_files",
+    "publish_unpublished_branch",
     "delete_branch",
     "cleanup_merged_pr_branch",
     "cleanup_stale_claims",

branch_publish.py

@@ -0,0 +1,591 @@
+"""Publish an unpublished local commit on a registered issue worktree (#812 AC20).
+
+Entry point B of #812 is the state where an author's work has already advanced
+to a local commit: the worktree is registered, clean, on the issue branch, and
+carries the only copy of the implementation, but the branch has never been
+published. That state deadlocks, because two individually correct predicates
+close a cycle:
+
+* ``issue_lock_renewal.assess_exact_owner_lease_renewal`` refuses to renew an
+  expired lease without an observable remote head — an unpublished branch has
+  none.
+* Every publication path (``gitea_commit_files``, ``gitea_create_pr``) derives
+  its workspace from the author issue lock under #618, so nothing can create
+  that remote head without first holding the lock.
+
+This module supplies the missing operation: it publishes an *already committed*
+local head to the remote branch, so exact-owner renewal has the evidence it
+requires. It deliberately does **not** renew, reclaim, rebind, or clear any
+lock. Publication is the whole of its authority.
+
+Why this is not a lock bypass
+-----------------------------
+The operation can only publish a branch whose **durable issue-lock record
+already names the caller as claimant**. Ownership is read from the lock file on
+disk (``issue_lock_store``), never from a caller-supplied flag, so the tool
+cannot manufacture a claim it does not already hold. Nothing here weakens the
+#510/#618/#713 guards: a dirty tree, an unregistered worktree, a foreign
+claimant, a changed HEAD, or a divergent remote head each refuse, exactly as
+they do today. The only thing this adds is the ability to make an existing,
+owned, committed, clean branch observable on the remote.
+
+Separation of records (#812 AC23)
+---------------------------------
+The durable **issue-lock file** and the control-plane **workflow lease** are
+distinct records. This module reads the former as ownership evidence and writes
+neither. Publishing changes remote git state only; no lock is renewed,
+abandoned, reclaimed, or generation-bumped here.
+
+Process evidence (#812 AC24)
+----------------------------
+Liveness of the lock's recorded pid is **not consulted**. That is deliberate:
+the recorded pid routinely belongs to the long-running MCP daemon rather than to
+an active author client, and the existing reclaim predicate
+(``assess_expired_lock_reclaim``) can never be satisfied while that daemon runs.
+Publication does not require the recording process to be dead, so this module
+never asserts, infers, or depends on a process being dead. Ownership is proven
+by identity and profile match against the recorded claimant instead.
+"""
+
+from __future__ import annotations
+
+import hashlib
+import os
+import re
+import subprocess
+
+from reviewer_worktree import parse_dirty_tracked_files
+from stable_branch_push_guard import is_stable_ref, redact_command
+
+# Assessment outcomes.
+PUBLISH_SANCTIONED = "publish_sanctioned"
+ALREADY_PUBLISHED = "already_published"
+REFUSED = "refused"
+
+#: Implementation branches must stay traceable to their issue (#713 lineage).
+ISSUE_BRANCH_RE = re.compile(r"^(fix|feat|docs|chore)/issue-(\d+)-.+$")
+
+_SHA_RE = re.compile(r"^[0-9a-f]{40}$")
+
+
+def _text(value: object) -> str:
+    return value.strip() if isinstance(value, str) else ""
+
+
+def _realpath(value: str | None) -> str | None:
+    path = _text(value)
+    return os.path.realpath(path) if path else None
+
+
+def parse_untracked_files(porcelain_status: str) -> list[str]:
+    """Return untracked paths from ``git status --porcelain`` output.
+
+    ``reviewer_worktree.parse_dirty_tracked_files`` deliberately skips ``??``
+    entries. Publication needs both halves: an untracked file in the worktree is
+    unpublished content that the commit does not carry, so publishing would
+    silently leave it behind.
+    """
+    untracked: list[str] = []
+    for line in (porcelain_status or "").splitlines():
+        if not line.startswith("??"):
+            continue
+        path = line[2:].strip()
+        if path:
+            untracked.append(path)
+    return untracked
+
+
+def hash_worktree_files(worktree_path: str, paths) -> dict[str, str | None]:
+    """SHA-256 each path under *worktree_path*; ``None`` when unreadable."""
+    root = _text(worktree_path)
+    hashes: dict[str, str | None] = {}
+    for rel in paths or ():
+        rel_text = _text(rel)
+        if not rel_text:
+            continue
+        full = os.path.join(root, rel_text)
+        try:
+            with open(full, "rb") as handle:
+                digest = hashlib.sha256()
+                for chunk in iter(lambda: handle.read(65536), b""):
+                    digest.update(chunk)
+            hashes[rel_text] = digest.hexdigest()
+        except OSError:
+            hashes[rel_text] = None
+    return hashes
+
+
+def read_remote_branch_head(
+    worktree_path: str, remote_name: str, branch_name: str
+) -> dict:
+    """Observe the remote head for *branch_name*, read-only.
+
+    ``probe_ok`` False means git could not answer at all. That is kept distinct
+    from "the branch does not exist": an unobservable remote must fail closed
+    rather than be mistaken for an absent branch, because the two lead to
+    opposite dispositions.
+    """
+    path = _text(worktree_path)
+    remote = _text(remote_name)
+    branch = _text(branch_name)
+    result: dict = {
+        "probe_ok": False,
+        "remote_branch_exists": False,
+        "remote_head_sha": None,
+        "reasons": [],
+    }
+    if not (path and remote and branch):
+        result["reasons"].append(
+            "remote head probe requires a worktree path, remote name, and branch"
+        )
+        return result
+    try:
+        res = subprocess.run(
+            ["git", "-C", path, "ls-remote", remote, f"refs/heads/{branch}"],
+            capture_output=True,
+            text=True,
+            check=False,
+        )
+    except OSError as exc:  # git unavailable — fail closed, never assume absent
+        result["reasons"].append(f"remote head probe could not run: {exc}")
+        return result
+    if res.returncode != 0:
+        result["reasons"].append(
+            f"remote head probe failed for '{branch}' on remote '{remote}'"
+        )
+        return result
+
+    result["probe_ok"] = True
+    for line in (res.stdout or "").splitlines():
+        parts = line.split()
+        if len(parts) >= 2 and parts[1] == f"refs/heads/{branch}":
+            result["remote_branch_exists"] = True
+            result["remote_head_sha"] = parts[0].strip()
+            break
+    return result
+
+
+def read_is_ancestor(
+    worktree_path: str, ancestor_sha: str, descendant_sha: str
+) -> dict:
+    """Observe whether *ancestor_sha* is an ancestor of *descendant_sha*."""
+    path = _text(worktree_path)
+    ancestor = _text(ancestor_sha)
+    descendant = _text(descendant_sha)
+    result: dict = {"probe_ok": False, "is_ancestor": False, "reasons": []}
+    if not (path and ancestor and descendant):
+        result["reasons"].append(
+            "ancestry probe requires a worktree path and both commit SHAs"
+        )
+        return result
+    try:
+        present = subprocess.run(
+            ["git", "-C", path, "rev-parse", "--verify", "--quiet",
+             f"{ancestor}^{{commit}}"],
+            capture_output=True, text=True, check=False,
+        )
+        if present.returncode != 0:
+            result["reasons"].append(
+                f"remote head {ancestor} is not present locally, so it cannot be "
+                "proven an ancestor of the commit being published"
+            )
+            return result
+        res = subprocess.run(
+            ["git", "-C", path, "merge-base", "--is-ancestor", ancestor, descendant],
+            capture_output=True, text=True, check=False,
+        )
+    except OSError as exc:
+        result["reasons"].append(f"ancestry probe could not run: {exc}")
+        return result
+    result["probe_ok"] = res.returncode in (0, 1)
+    result["is_ancestor"] = res.returncode == 0
+    return result
+
+
+def assess_unpublished_commit_publication(
+    existing_lock,
+    *,
+    issue_number: int,
+    branch_name: str,
+    worktree_path: str,
+    expected_head: str,
+    remote: str,
+    org: str,
+    repo: str,
+    identity: str | None,
+    profile: str | None,
+    worktree_state,
+    worktree_registered: bool | None = None,
+    remote_probe=None,
+    ancestry=None,
+    competing_open_prs=(),
+    expected_file_hashes=None,
+    observed_file_hashes=None,
+) -> dict:
+    """Decide whether an unpublished local commit may be published (#812 AC20).
+
+    Pure predicate. Every input is either a caller-declared expectation that
+    must be *matched* against observation, or a server-side observation. No
+    caller-supplied boolean is accepted as proof of ownership, liveness, or
+    eligibility: ``existing_lock`` comes from the durable lock file and the
+    git/PR state is observed by the server.
+
+    The single mutating disposition it can return is "publish this exact commit
+    to this exact branch". It never sanctions renewal, reclamation, force
+    updates, history rewriting, or publication of uncommitted content.
+    """
+    reasons: list[str] = []
+    branch = _text(branch_name)
+    head = _text(expected_head).lower()
+    workspace = _realpath(worktree_path)
+    state = worktree_state if isinstance(worktree_state, dict) else {}
+    lock = existing_lock if isinstance(existing_lock, dict) else None
+
+    evidence: dict = {
+        "issue_number": issue_number,
+        "branch_name": branch or None,
+        "worktree_path": workspace,
+        "expected_head": head or None,
+        "remote": _text(remote) or None,
+        "org": _text(org) or None,
+        "repo": _text(repo) or None,
+        "identity": _text(identity) or None,
+        "profile": _text(profile) or None,
+        "lock_record_present": lock is not None,
+        "recorded_claimant": None,
+        "recorded_branch": None,
+        "recorded_worktree": None,
+        "lock_generation": None,
+        "local_head_sha": _text(state.get("head_sha")) or None,
+        "current_branch": _text(state.get("current_branch")) or None,
+        "dirty_tracked_files": [],
+        "untracked_files": [],
+        "worktree_registered": worktree_registered,
+        "remote_branch_exists": None,
+        "remote_head_sha": None,
+        "fast_forward_from_remote": None,
+        "competing_open_prs": [],
+        "file_hashes_verified": None,
+        "hash_mismatches": [],
+        # Recorded explicitly so no reader mistakes silence for a liveness
+        # claim, and so the audit shows which records were left alone (AC23/AC24).
+        "owner_pid_liveness_consulted": False,
+        "workflow_lease_touched": False,
+        "issue_lock_record_mutated": False,
+    }
+
+    # ── declared shape ────────────────────────────────────────────────────
+    if not branch:
+        reasons.append("branch name not declared; fail closed")
+    if not head:
+        reasons.append(
+            "expected_head not declared; publication must name the exact commit"
+        )
+    elif not _SHA_RE.match(head):
+        reasons.append(
+            f"expected_head '{head}' is not a full 40-character commit SHA; "
+            "abbreviated or symbolic revisions are refused"
+        )
+    if not workspace:
+        reasons.append("worktree path not declared; fail closed")
+
+    if branch:
+        match = ISSUE_BRANCH_RE.match(branch)
+        if not match:
+            reasons.append(
+                f"branch '{branch}' is not an issue-linked implementation branch "
+                "((fix|feat|docs|chore)/issue-<number>-<description>); fail closed"
+            )
+        elif int(match.group(2)) != int(issue_number):
+            reasons.append(
+                f"branch '{branch}' does not carry issue number {issue_number}; "
+                "fail closed"
+            )
+        if is_stable_ref(branch):
+            reasons.append(
+                f"refusing to publish stable branch '{branch}'; this operation "
+                "publishes issue branches only"
+            )
+
+    # ── ownership: durable issue-lock record only (AC8, AC20, AC24) ───────
+    if lock is None:
+        reasons.append(
+            "no durable issue-lock record for this issue; publication requires an "
+            "existing recorded claim naming the caller, so this operation cannot "
+            "be used to bypass the author lock"
+        )
+    else:
+        lease = lock.get("work_lease")
+        lease = lease if isinstance(lease, dict) else {}
+        claimant = lease.get("claimant")
+        claimant = claimant if isinstance(claimant, dict) else {}
+        recorded_user = _text(claimant.get("username"))
+        recorded_profile = _text(claimant.get("profile"))
+        recorded_branch = _text(lock.get("branch_name")) or _text(lease.get("branch"))
+        recorded_worktree = _realpath(
+            _text(lock.get("worktree_path")) or _text(lease.get("worktree_path"))
+        )
+        evidence["recorded_claimant"] = {
+            "username": recorded_user or None,
+            "profile": recorded_profile or None,
+        }
+        evidence["recorded_branch"] = recorded_branch or None
+        evidence["recorded_worktree"] = recorded_worktree
+        try:
+            evidence["lock_generation"] = int(lock.get("lock_generation") or 0)
+        except (TypeError, ValueError):
+            evidence["lock_generation"] = 0
+
+        try:
+            recorded_issue = int(lock.get("issue_number") or 0)
+        except (TypeError, ValueError):
+            recorded_issue = 0
+        if recorded_issue != int(issue_number):
+            reasons.append(
+                f"durable lock records issue {lock.get('issue_number')}, not "
+                f"{issue_number}; ambiguous ownership, fail closed"
+            )
+        for field, declared in (
+            ("remote", _text(remote)),
+            ("org", _text(org)),
+            ("repo", _text(repo)),
+        ):
+            recorded = _text(lock.get(field))
+            if recorded and declared and recorded != declared:
+                reasons.append(
+                    f"durable lock records {field} '{recorded}' but the request "
+                    f"declares '{declared}'; repository mismatch, fail closed"
+                )
+        if recorded_branch and branch and recorded_branch != branch:
+            reasons.append(
+                f"durable lock records branch '{recorded_branch}' but the request "
+                f"declares '{branch}'; fail closed"
+            )
+        if recorded_worktree and workspace and recorded_worktree != workspace:
+            reasons.append(
+                f"durable lock records worktree '{recorded_worktree}' but the "
+                f"request declares '{workspace}'; fail closed"
+            )
+        if not recorded_user or not recorded_profile:
+            reasons.append(
+                "durable lock does not record a claimant username and profile; "
+                "ownership cannot be proven, fail closed"
+            )
+        else:
+            if recorded_user != _text(identity):
+                reasons.append(
+                    f"durable lock claimant '{recorded_user}' is not the acting "
+                    f"identity '{_text(identity) or '(unknown)'}'; foreign claim, "
+                    "fail closed"
+                )
+            if recorded_profile != _text(profile):
+                reasons.append(
+                    f"durable lock claimant profile '{recorded_profile}' is not "
+                    f"the active profile '{_text(profile) or '(unknown)'}'; "
+                    "fail closed"
+                )
+
+    # ── worktree: registered, on-branch, clean, at the expected commit ────
+    if worktree_registered is False:
+        reasons.append(
+            f"worktree '{workspace}' is not listed in git worktree list; #713 "
+            "requires a genuinely registered worktree, fail closed"
+        )
+
+    current_branch = _text(state.get("current_branch"))
+    if not current_branch:
+        reasons.append("worktree branch could not be observed; fail closed")
+    elif branch and current_branch != branch:
+        reasons.append(
+            f"worktree is on branch '{current_branch}', not '{branch}'; fail closed"
+        )
+
+    porcelain = state.get("porcelain_status") or ""
+    dirty_tracked = parse_dirty_tracked_files(porcelain)
+    untracked = parse_untracked_files(porcelain)
+    evidence["dirty_tracked_files"] = dirty_tracked
+    evidence["untracked_files"] = untracked
+    if dirty_tracked:
+        reasons.append(
+            "worktree has dirty tracked files, so the commit is not the whole of "
+            f"the work: {', '.join(dirty_tracked)}. This operation publishes an "
+            "existing clean commit only; uncommitted content is out of scope"
+        )
+    if untracked:
+        reasons.append(
+            "worktree has untracked files that the commit does not carry: "
+            f"{', '.join(untracked)}. Publishing would silently leave them "
+            "behind; fail closed"
+        )
+
+    local_head = _text(state.get("head_sha")).lower()
+    if not local_head:
+        reasons.append("local HEAD could not be observed; fail closed")
+    elif head and local_head != head:
+        reasons.append(
+            f"worktree HEAD is {local_head} but the request declares {head}; the "
+            "local commit changed since it was recorded, fail closed"
+        )
+
+    # ── remote state ──────────────────────────────────────────────────────
+    probe = remote_probe if isinstance(remote_probe, dict) else {}
+    already_published = False
+    if not probe.get("probe_ok"):
+        reasons.append(
+            "remote branch head could not be observed; publication must not "
+            "proceed against an unknown remote state, fail closed"
+        )
+        reasons.extend(probe.get("reasons") or [])
+    else:
+        remote_exists = bool(probe.get("remote_branch_exists"))
+        remote_head = _text(probe.get("remote_head_sha")).lower() or None
+        evidence["remote_branch_exists"] = remote_exists
+        evidence["remote_head_sha"] = remote_head
+        if remote_exists and remote_head and head:
+            if remote_head == head:
+                already_published = True
+                evidence["fast_forward_from_remote"] = True
+            else:
+                anc = ancestry if isinstance(ancestry, dict) else {}
+                is_anc = bool(anc.get("probe_ok")) and bool(anc.get("is_ancestor"))
+                evidence["fast_forward_from_remote"] = is_anc
+                if not is_anc:
+                    reasons.append(
+                        f"remote branch '{branch}' already exists at {remote_head}, "
+                        f"which is not an ancestor of {head}; publishing would "
+                        "discard or rewrite published history, fail closed"
+                    )
+                    reasons.extend(anc.get("reasons") or [])
+        elif remote_exists and not remote_head:
+            reasons.append(
+                f"remote branch '{branch}' exists but its head could not be read; "
+                "fail closed"
+            )
+
+    # ── competing claims ──────────────────────────────────────────────────
+    competing = [p for p in (competing_open_prs or ()) if p]
+    evidence["competing_open_prs"] = list(competing)
+    if competing:
+        reasons.append(
+            f"open pull request(s) {competing} already claim issue {issue_number} "
+            "or this branch; ambiguous ownership, fail closed"
+        )
+
+    # ── content verification before publication ───────────────────────────
+    if expected_file_hashes:
+        observed = (
+            observed_file_hashes if isinstance(observed_file_hashes, dict) else {}
+        )
+        mismatches: list[str] = []
+        for path, expected_digest in dict(expected_file_hashes).items():
+            actual = observed.get(path)
+            if actual is None:
+                mismatches.append(f"{path}: missing or unreadable in the worktree")
+            elif _text(actual).lower() != _text(expected_digest).lower():
+                mismatches.append(
+                    f"{path}: expected {expected_digest}, observed {actual}"
+                )
+        evidence["hash_mismatches"] = mismatches
+        evidence["file_hashes_verified"] = not mismatches
+        if mismatches:
+            reasons.append(
+                "declared content hashes do not match the worktree: "
+                + "; ".join(mismatches)
+                + ". Refusing to publish content that is not what was recorded"
+            )
+
+    if reasons:
+        return {
+            "outcome": REFUSED,
+            "publish_sanctioned": False,
+            "already_published": False,
+            "reasons": reasons,
+            "evidence": evidence,
+        }
+    return {
+        "outcome": ALREADY_PUBLISHED if already_published else PUBLISH_SANCTIONED,
+        # Idempotent retry: a remote head that already equals the assessed commit
+        # needs no second push, so the caller verifies instead of acting.
+        "publish_sanctioned": not already_published,
+        "already_published": already_published,
+        "reasons": [],
+        "evidence": evidence,
+    }
+
+
+def publish_commit_to_remote_branch(
+    *,
+    worktree_path: str,
+    remote_name: str,
+    branch_name: str,
+    expected_head: str,
+) -> dict:
+    """Send exactly *expected_head* to ``refs/heads/<branch_name>``.
+
+    The refspec names the commit SHA explicitly rather than ``HEAD`` or the
+    local branch, so what lands is the commit that was assessed and nothing
+    else. No force, no lease, no ``+`` prefix: a non-fast-forward is rejected by
+    git itself, the last of several independent guards against overwriting
+    published history.
+    """
+    path = _text(worktree_path)
+    remote = _text(remote_name)
+    branch = _text(branch_name)
+    head = _text(expected_head)
+    result: dict = {
+        "success": False,
+        "pushed_ref": f"refs/heads/{branch}" if branch else None,
+        "pushed_sha": head or None,
+        "stderr": None,
+        "reasons": [],
+    }
+    if not (path and remote and branch and head):
+        result["reasons"].append(
+            "publication requires a worktree path, remote, branch, and commit SHA"
+        )
+        return result
+
+    refspec = f"{head}:refs/heads/{branch}"
+    try:
+        res = subprocess.run(
+            ["git", "-C", path, "push", remote, refspec],
+            capture_output=True,
+            text=True,
+            check=False,
+        )
+    except OSError as exc:
+        result["reasons"].append(f"publication could not run: {exc}")
+        return result
+
+    if res.returncode != 0:
+        # Redact before surfacing: failures can echo credentialed remote URLs.
+        result["stderr"] = redact_command(res.stderr or "")
+        result["reasons"].append(
+            f"publication of {head} to '{branch}' on remote '{remote}' failed"
+        )
+        return result
+
+    result["success"] = True
+    return result
+
+
+def verify_published_head(
+    *, worktree_path: str, remote_name: str, branch_name: str, expected_head: str
+) -> dict:
+    """Read-after-write: confirm the remote head equals *expected_head* (AC20)."""
+    probe = read_remote_branch_head(worktree_path, remote_name, branch_name)
+    head = _text(expected_head).lower()
+    observed = _text(probe.get("remote_head_sha")).lower() or None
+    verified = bool(head) and bool(probe.get("probe_ok")) and observed == head
+    reasons: list[str] = list(probe.get("reasons") or [])
+    if probe.get("probe_ok") and not verified:
+        reasons.append(
+            "read-after-write verification failed: remote head is "
+            f"{observed or '(absent)'}, expected {head}"
+        )
+    return {
+        "verified": verified,
+        "remote_head_sha": observed,
+        "expected_head": head or None,
+        "reasons": reasons,
+    }

docs/mcp-tool-inventory.md

@@ -120,6 +120,7 @@ that gates each call, not which tools exist.
 - `gitea_observability_list_projects`
 - `gitea_observability_reconcile_incident`
 - `gitea_post_heartbeat`
+- `gitea_publish_unpublished_issue_branch`
 - `gitea_quarantine_contaminated_review`
 - `gitea_reclaim_expired_workflow_lease`
 - `gitea_reconcile_already_landed_pr`

docs/webui-local-dev.md

@@ -67,6 +67,11 @@ that govern when a write path may open (#632, epic #631).
 | `/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) |
+| `/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 |
+| `/insights` | Phase 1 shell stub — operational insights placeholder |
 
 Most routes are GET-only. POST/PUT/PATCH/DELETE return `405` with
 `read-only-mvp`, except `/audit` and `/api/audit` which accept POST for
@@ -147,6 +152,26 @@ health, workflow/schema SHA-256 hashes, and stale-runtime warnings when the
 checkout is behind merged safety-gate changes. Restart guidance links to #420;
 no tokens or MCP restart actions are exposed.
 
+## Application shell — Phase 1 (#638)
+
+The console shell (`webui/layout.py`) renders a grouped navigation driven by a
+single nav-config module, `webui/nav.py`. Nav groups follow the epic #631
+Phase 1 information architecture: **Health, Traffic, Runtime/Sessions,
+Projects, Inventory, Timeline, Policy** (placeholder), and **Insights**
+(placeholder). Live views and Phase 1 placeholders (`stub`) are declared in one
+place so the layout and the route table cannot drift.
+
+The header carries two read-only status badges — an **environment** badge
+(`local` for loopback binds, `remote` otherwise, derived from `WEBUI_HOST`) and
+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`.
+
 ## Deployment boundary (#435)
 
 MVP serves on loopback by default. Binding `0.0.0.0` or `::` is **refused**

gitea_mcp_server.py

@@ -2024,6 +2024,7 @@ import review_quarantine  # noqa: E402  # #695 contaminated formal-review quaran
 import mcp_daemon_guard  # noqa: E402  # #695 native transport provenance
 import already_landed_reconcile  # noqa: E402
 import author_mutation_worktree  # noqa: E402
+import branch_publish  # noqa: E402  # #812 AC20 unpublished-commit publication
 import root_checkout_guard  # noqa: E402
 import workflow_scope_guard  # noqa: E402  # #683 production scope / force-on guards
 import stable_branch_push_guard  # noqa: E402
@@ -9082,6 +9083,237 @@ def gitea_commit_files(
     }
 
 
+def _publication_block(reasons: list[str], **extra) -> dict:
+    """Uniform fail-closed shape for publication refusals (#812 AC20)."""
+    payload = {
+        "success": False,
+        "performed": False,
+        "published": False,
+        "verified": False,
+        "outcome": branch_publish.REFUSED,
+        "reasons": reasons,
+        # State every record this operation left alone, so a refusal can never
+        # be misread as a lock mutation (#812 AC23).
+        "issue_lock_record_mutated": False,
+        "workflow_lease_touched": False,
+    }
+    payload.update(extra)
+    return payload
+
+
+@mcp.tool()
+def gitea_publish_unpublished_issue_branch(
+    issue_number: int,
+    branch_name: str,
+    worktree_path: str,
+    expected_head: str,
+    remote: str = "dadeschools",
+    host: str | None = None,
+    org: str | None = None,
+    repo: str | None = None,
+    git_remote_name: str | None = None,
+    expected_file_hashes: dict | None = None,
+    dry_run: bool = False,
+) -> dict:
+    """Publish an already-committed, unpublished issue branch (#812 AC20).
+
+    Creates the remote head for a branch whose work is *already* a local commit
+    on a registered, clean worktree, so exact-owner lease renewal
+    (``issue_lock_renewal``) has the published head its evidence model requires.
+    This is the one step of the entry point B deadlock that no existing tool can
+    perform: publication is otherwise lock-derived under #618, and the lock
+    itself is withheld until a remote head exists.
+
+    Not a lock bypass. The branch's **durable issue-lock record must already
+    name the caller as claimant** — ownership is read from the lock file, never
+    asserted by the caller — so this can only publish work the caller already
+    owns. It refuses a dirty or untracked-carrying worktree, an unregistered
+    worktree, a changed local HEAD, a remote head that is not an ancestor of the
+    commit, a competing open PR on another branch for the same issue, and any
+    declared-hash mismatch. It renews, reclaims, and clears nothing: the durable
+    issue-lock file and the control-plane workflow lease are both left untouched
+    (#812 AC23), and the recorded owner pid's liveness is never consulted or
+    asserted (#812 AC24).
+
+    Args:
+        issue_number: The issue whose recorded claim authorizes publication.
+        branch_name: Issue branch to publish, ``(fix|feat|docs|chore)/issue-N-…``.
+        worktree_path: Registered worktree holding the commit.
+        expected_head: Full 40-character SHA of the commit to publish. Required:
+            publication names the exact commit, and a mismatch fails closed.
+        remote: Known instance — 'dadeschools' or 'prgs'.
+        host: Override the Gitea host.
+        org: Override the owner/organization.
+        repo: Override the repository name.
+        git_remote_name: Git remote to publish to; defaults to *remote*.
+        expected_file_hashes: Optional ``{path: sha256}`` verified against the
+            worktree before publication. Any mismatch or missing file refuses.
+        dry_run: Report the decision and evidence, mutate nothing.
+
+    Returns:
+        dict with 'success', 'performed', 'published', 'verified', 'outcome',
+        'remote_head_sha', 'reasons', and 'evidence'.
+    """
+    task = "publish_unpublished_branch"
+    ok, block_reasons = role_session_router.check_author_mutation_after_reviewer_stop(
+        task
+    )
+    if not ok:
+        return _publication_block(block_reasons)
+
+    blocked = _namespace_mutation_block(task, remote=remote)
+    if blocked:
+        return blocked
+    blocked = _profile_permission_block(
+        task_capability_map.required_permission(task),
+        remote=remote, host=host, org=org, repo=repo,
+        org_explicit=org is not None,
+        repo_explicit=repo is not None,
+    )
+    if blocked:
+        return blocked
+
+    verify_preflight_purity(remote, task=task, org=org, repo=repo)
+
+    h, o, r = _resolve(remote, host, org, repo)
+    git_remote = (git_remote_name or remote or "").strip()
+    workspace = os.path.realpath(os.path.abspath((worktree_path or "").strip() or "."))
+
+    existing_lock = issue_lock_store.load_issue_lock(
+        remote=remote, org=o, repo=r, issue_number=int(issue_number)
+    )
+    git_state = issue_lock_worktree.read_worktree_git_state(workspace)
+    registered = author_mutation_worktree.path_in_git_worktree_list(
+        workspace, PROJECT_ROOT
+    )
+
+    remote_probe = branch_publish.read_remote_branch_head(
+        workspace, git_remote, branch_name
+    )
+    ancestry = None
+    probe_head = (remote_probe.get("remote_head_sha") or "").strip()
+    if probe_head and probe_head.lower() != (expected_head or "").strip().lower():
+        ancestry = branch_publish.read_is_ancestor(workspace, probe_head, expected_head)
+
+    # A PR on this very branch is this work's own PR, not a rival claim. Only an
+    # open PR for the same issue on a *different* branch is a competing claim.
+    competing: list = []
+    for pull in _list_open_pulls(h, o, r, _auth(h)):
+        ref = str((pull.get("head") or {}).get("ref") or "")
+        if not ref or ref == branch_name:
+            continue
+        if issue_lock_adoption.branch_carries_issue_marker(ref, int(issue_number)):
+            competing.append(pull.get("number"))
+
+    observed_hashes = None
+    if expected_file_hashes:
+        observed_hashes = branch_publish.hash_worktree_files(
+            workspace, list(expected_file_hashes.keys())
+        )
+
+    claimant = _work_lease_claimant(h)
+    assessment = branch_publish.assess_unpublished_commit_publication(
+        existing_lock,
+        issue_number=int(issue_number),
+        branch_name=branch_name,
+        worktree_path=workspace,
+        expected_head=expected_head,
+        remote=remote,
+        org=o,
+        repo=r,
+        identity=claimant.get("username"),
+        profile=claimant.get("profile"),
+        worktree_state=git_state,
+        worktree_registered=registered,
+        remote_probe=remote_probe,
+        ancestry=ancestry,
+        competing_open_prs=competing,
+        expected_file_hashes=expected_file_hashes,
+        observed_file_hashes=observed_hashes,
+    )
+
+    if assessment["outcome"] == branch_publish.REFUSED:
+        return _publication_block(
+            assessment["reasons"], evidence=assessment["evidence"]
+        )
+
+    if dry_run:
+        return {
+            "success": True,
+            "performed": False,
+            "published": False,
+            "verified": False,
+            "dry_run": True,
+            "outcome": assessment["outcome"],
+            "would_publish": assessment["publish_sanctioned"],
+            "remote_head_sha": assessment["evidence"].get("remote_head_sha"),
+            "reasons": [],
+            "evidence": assessment["evidence"],
+            "issue_lock_record_mutated": False,
+            "workflow_lease_touched": False,
+        }
+
+    performed = False
+    if assessment["publish_sanctioned"]:
+        with _audited(
+            task, host=h, remote=remote, org=o, repo=r,
+            target_branch=branch_name,
+            request_metadata={
+                "issue_number": int(issue_number),
+                "expected_head": expected_head,
+                "git_remote": git_remote,
+            },
+        ):
+            push = branch_publish.publish_commit_to_remote_branch(
+                worktree_path=workspace,
+                remote_name=git_remote,
+                branch_name=branch_name,
+                expected_head=expected_head,
+            )
+        if not push.get("success"):
+            return _publication_block(
+                push.get("reasons") or ["publication failed"],
+                evidence=assessment["evidence"],
+                stderr=push.get("stderr"),
+            )
+        performed = True
+
+    verification = branch_publish.verify_published_head(
+        worktree_path=workspace,
+        remote_name=git_remote,
+        branch_name=branch_name,
+        expected_head=expected_head,
+    )
+    if not verification.get("verified"):
+        return _publication_block(
+            verification.get("reasons") or ["read-after-write verification failed"],
+            evidence=assessment["evidence"],
+            performed=performed,
+            published=performed,
+            remote_head_sha=verification.get("remote_head_sha"),
+        )
+
+    return {
+        "success": True,
+        "performed": performed,
+        "published": True,
+        "verified": True,
+        "outcome": assessment["outcome"],
+        "remote_head_sha": verification.get("remote_head_sha"),
+        "reasons": [],
+        "evidence": assessment["evidence"],
+        # Publication is the whole of this operation's authority (#812 AC23).
+        "issue_lock_record_mutated": False,
+        "workflow_lease_touched": False,
+        "exact_next_action": (
+            f"Remote head for '{branch_name}' is now observable. Call "
+            f"gitea_lock_issue(issue_number={int(issue_number)}, "
+            f"branch_name='{branch_name}', worktree_path='{workspace}') to renew "
+            "the exact-owner lease, then gitea_create_pr."
+        ),
+    }
+
+
 # Merge methods supported by the Gitea merge API.
 _MERGE_METHODS = ("merge", "squash", "rebase")
 

task_capability_map.py

@@ -62,6 +62,14 @@ TASK_CAPABILITY_MAP: dict[str, dict[str, str]] = {
         "permission": "gitea.branch.push",
         "role": "author",
     },
+    # #812 AC20: publish an already-committed, unpublished local head so
+    # exact-owner lease renewal has an observable remote head to reason about.
+    # Same authority as any other author push — deliberately not a new
+    # operation name, so it cannot widen an already-configured author profile.
+    "publish_unpublished_branch": {
+        "permission": "gitea.branch.push",
+        "role": "author",
+    },
     "create_pr": {
         "permission": "gitea.pr.create",
         "role": "author",
@@ -500,6 +508,7 @@ ROLE_EXCLUSIVE_TASKS: frozenset[str] = frozenset(
         "gitea_release_merger_pr_lease",
         "create_branch",
         "push_branch",
+        "publish_unpublished_branch",
         "create_pr",
         "commit_files",
         "gitea_commit_files",

tests/test_issue_812_publish_unpublished_commit.py

@@ -0,0 +1,650 @@
+"""Publication of an unpublished local commit (#812 AC20).
+
+Entry point B of #812: a registered worktree, clean, on its issue branch,
+holding a local commit that has never been published. Exact-owner lease renewal
+refuses such a claim for want of an observable remote head, and every existing
+publication path is lock-derived, so the two predicates close a cycle around
+work that is otherwise complete.
+
+These tests exercise the disposition through its *evidence*, never through any
+particular issue number: every case uses an arbitrary issue number against a
+synthetic repository, and the same assertions hold for any other. Nothing here
+reads, writes, or references the live protected worktree named in #812 AC17 —
+that content is preserved evidence for the duration of this work, so the
+fixtures below build their own repositories from scratch.
+
+The remote is a local bare repository, so publication and read-after-write
+verification are genuinely executed rather than mocked.
+"""
+
+from __future__ import annotations
+
+import os
+import subprocess
+import sys
+import tempfile
+import unittest
+from datetime import datetime, timedelta, timezone
+from unittest.mock import patch
+
+sys.path.insert(0, os.path.dirname(os.path.dirname(os.path.abspath(__file__))))
+sys.path.insert(0, os.path.dirname(os.path.abspath(__file__)))
+
+import branch_publish  # noqa: E402
+import issue_lock_provenance  # noqa: E402
+import issue_lock_renewal  # noqa: E402
+import issue_lock_store  # noqa: E402
+import mcp_server  # noqa: E402
+from mutation_profile_fixture import shared_mutation_env  # noqa: E402
+
+ISSUE = 9812
+BRANCH = f"feat/issue-{ISSUE}-publish-fixture"
+IDENTITY = "example-user"
+PROFILE = "test-author-prgs"
+ORG = "Scaled-Tech-Consulting"
+REPO = "Gitea-Tools"
+GIT_REMOTE = "prgs"
+
+
+def _ts(hours: int) -> str:
+    return (
+        (datetime.now(timezone.utc) + timedelta(hours=hours))
+        .isoformat()
+        .replace("+00:00", "Z")
+    )
+
+
+class _PublishBase(unittest.TestCase):
+    """Real git repo + real bare remote + durable lock naming the caller.
+
+    The recorded owner pid is deliberately **this live process**. That mirrors
+    the production shape #812 documents, where the pid belongs to a long-running
+    MCP daemon rather than to a dead author client, and it proves publication
+    never depends on a dead process (#812 AC24).
+    """
+
+    def setUp(self):
+        self.lock_dir = tempfile.TemporaryDirectory()
+        self.addCleanup(self.lock_dir.cleanup)
+        self.origin = tempfile.mkdtemp(prefix="issue812-origin-")
+        self.repo = tempfile.mkdtemp(prefix="issue812-work-")
+        for path in (self.origin, self.repo):
+            self.addCleanup(
+                lambda p=path: subprocess.run(["rm", "-rf", p], check=False)
+            )
+        self._init_repos()
+        self.remotes = patch.dict(
+            mcp_server.REMOTES,
+            {"prgs": {"host": "gitea.prgs.cc", "org": ORG, "repo": REPO}},
+        )
+        self.remotes.start()
+        self.addCleanup(patch.stopall)
+        mcp_server._IDENTITY_CACHE.clear()
+
+    # ── fixture construction ─────────────────────────────────────────────
+    def _git(self, *args, cwd=None):
+        return subprocess.run(
+            ["git", "-C", cwd or self.repo, *args],
+            capture_output=True,
+            text=True,
+            check=True,
+        )
+
+    def _init_repos(self):
+        subprocess.run(
+            ["git", "init", "-q", "--bare", "-b", "master", self.origin], check=True
+        )
+        self._git("init", "-q", "-b", "master")
+        self._git("config", "user.email", "[email protected]")
+        self._git("config", "user.name", "Test")
+        self._git("remote", "add", GIT_REMOTE, self.origin)
+
+        with open(os.path.join(self.repo, "seed.txt"), "w") as fh:
+            fh.write("seed\n")
+        self._git("add", "seed.txt")
+        self._git("commit", "-q", "-m", "seed")
+        self.base_sha = self._git("rev-parse", "HEAD").stdout.strip()
+        self._git("push", "-q", GIT_REMOTE, "master")
+
+        self._git("checkout", "-q", "-b", BRANCH)
+        with open(os.path.join(self.repo, "work.txt"), "w") as fh:
+            fh.write("unpublished implementation\n")
+        self._git("add", "work.txt")
+        self._git("commit", "-q", "-m", "unpublished implementation")
+        self.head_sha = self._git("rev-parse", "HEAD").stdout.strip()
+        self.worktree = os.path.realpath(self.repo)
+
+    def lock_path(self):
+        return issue_lock_store.lock_file_path(
+            remote="prgs", org=ORG, repo=REPO, issue_number=ISSUE,
+            lock_dir=self.lock_dir.name,
+        )
+
+    def write_lock(self, **overrides):
+        path = self.lock_path()
+        claimant = overrides.pop(
+            "claimant", {"username": IDENTITY, "profile": PROFILE}
+        )
+        pid = overrides.pop("session_pid", os.getpid())
+        lease = {
+            "operation_type": issue_lock_store.AUTHOR_ISSUE_WORK_LEASE,
+            "issue_number": ISSUE,
+            "pr_number": None,
+            "branch": overrides.get("branch_name", BRANCH),
+            "worktree_path": overrides.get("worktree_path", self.worktree),
+            "claimant": claimant,
+            "created_at": _ts(-2),
+            "last_heartbeat_at": _ts(-2),
+            # Expired: entry point B's lease has lapsed, which is precisely why
+            # renewal — and therefore a published head — is needed.
+            "expires_at": _ts(-1),
+        }
+        lease.update(overrides.pop("work_lease", {}))
+        data = {
+            "issue_number": ISSUE,
+            "branch_name": BRANCH,
+            "remote": "prgs",
+            "org": ORG,
+            "repo": REPO,
+            "worktree_path": self.worktree,
+            "session_pid": pid,
+            "pid": pid,
+            "lock_generation": 1,
+            "work_lease": lease,
+            "lock_provenance": issue_lock_provenance.build_sanctioned_lock_provenance(
+                tool="gitea_lock_issue", claimant=claimant
+            ),
+        }
+        data.update(overrides)
+        data["lock_file_path"] = path
+        issue_lock_store.save_lock_file(path, data)
+        return path
+
+    def _tool_env(self):
+        env = shared_mutation_env(
+            PROFILE, include_example_repo=True,
+            GITEA_ISSUE_LOCK_DIR=self.lock_dir.name,
+        )
+        env["GITEA_ISSUE_LOCK_DIR"] = self.lock_dir.name
+        # These tests repoint PROJECT_ROOT at a synthetic repository so the
+        # registered-worktree proof runs for real. Pin the parity gate to the
+        # server's own startup head so that repointing does not read as a stale
+        # daemon; the gate itself stays live and enforced.
+        startup_head = mcp_server._STARTUP_PARITY.get("startup_head") or ""
+        env["GITEA_TEST_CURRENT_HEAD"] = startup_head
+        env["GITEA_TEST_LIVE_REMOTE_HEAD"] = startup_head
+        return env
+
+    # ── tool driver ──────────────────────────────────────────────────────
+    def run_publish(self, *, open_prs=None, expected_head=None, **kwargs):
+        """Drive the public publication tool against the synthetic fixture."""
+        env = self._tool_env()
+        with patch(
+            "mcp_server._list_open_pulls", return_value=list(open_prs or [])
+        ), patch(
+            "mcp_server._auth", return_value="token x"
+        ), patch(
+            "mcp_server.get_auth_header", return_value="token x"
+        ), patch(
+            "mcp_server._work_lease_claimant",
+            return_value={"username": IDENTITY, "profile": PROFILE},
+        ), patch.object(
+            mcp_server, "PROJECT_ROOT", self.repo
+        ), patch.dict(os.environ, env, clear=True):
+            os.environ["GITEA_ISSUE_LOCK_DIR"] = self.lock_dir.name
+            return mcp_server.gitea_publish_unpublished_issue_branch(
+                issue_number=kwargs.pop("issue_number", ISSUE),
+                branch_name=kwargs.pop("branch_name", BRANCH),
+                worktree_path=kwargs.pop("worktree_path", self.worktree),
+                expected_head=expected_head or self.head_sha,
+                remote="prgs",
+                git_remote_name=kwargs.pop("git_remote_name", GIT_REMOTE),
+                **kwargs,
+            )
+
+    def remote_head(self, branch=BRANCH):
+        res = subprocess.run(
+            ["git", "-C", self.origin, "rev-parse", "--verify", "--quiet", branch],
+            capture_output=True, text=True, check=False,
+        )
+        return (res.stdout or "").strip() or None
+
+
+class TestSuccessfulPublication(_PublishBase):
+    """AC20 — the branch becomes observable and is verified after the write."""
+
+    def test_publishes_clean_unpublished_commit(self):
+        self.write_lock()
+        self.assertIsNone(self.remote_head(), "fixture must start unpublished")
+
+        result = self.run_publish()
+
+        self.assertTrue(result["success"], result.get("reasons"))
+        self.assertTrue(result["performed"])
+        self.assertTrue(result["published"])
+        self.assertTrue(result["verified"], "read-after-write must be proven")
+        self.assertEqual(result["remote_head_sha"], self.head_sha)
+        self.assertEqual(self.remote_head(), self.head_sha)
+
+    def test_publication_does_not_rewrite_the_commit(self):
+        self.write_lock()
+        self.run_publish()
+        # The published object is the same commit, not a copy or a rewrite.
+        self.assertEqual(self.remote_head(), self.head_sha)
+        self.assertEqual(
+            self._git("rev-parse", "HEAD").stdout.strip(), self.head_sha
+        )
+
+    def test_exact_next_action_names_the_lock_call(self):
+        self.write_lock()
+        result = self.run_publish()
+        self.assertIn("gitea_lock_issue", result["exact_next_action"])
+
+
+class TestFailsClosed(_PublishBase):
+    """AC20/AC9 — each refusal reason, exercised independently."""
+
+    def test_changed_local_head_refuses(self):
+        self.write_lock()
+        stale = self.base_sha  # a real commit, but not the declared head
+        result = self.run_publish(expected_head=stale)
+        self.assertFalse(result["success"])
+        self.assertTrue(
+            any("local commit changed" in r for r in result["reasons"]),
+            result["reasons"],
+        )
+        self.assertIsNone(self.remote_head(), "refusal must not publish")
+
+    def test_abbreviated_sha_refuses(self):
+        self.write_lock()
+        result = self.run_publish(expected_head=self.head_sha[:8])
+        self.assertFalse(result["success"])
+        self.assertTrue(
+            any("40-character" in r for r in result["reasons"]), result["reasons"]
+        )
+
+    def test_dirty_tracked_worktree_refuses(self):
+        self.write_lock()
+        with open(os.path.join(self.repo, "work.txt"), "a") as fh:
+            fh.write("uncommitted edit\n")
+
+        result = self.run_publish()
+
+        self.assertFalse(result["success"])
+        self.assertTrue(
+            any("dirty tracked files" in r for r in result["reasons"]),
+            result["reasons"],
+        )
+        self.assertIn("work.txt", result["evidence"]["dirty_tracked_files"])
+        self.assertIsNone(self.remote_head())
+
+    def test_untracked_file_refuses(self):
+        self.write_lock()
+        with open(os.path.join(self.repo, "stray.txt"), "w") as fh:
+            fh.write("not committed\n")
+
+        result = self.run_publish()
+
+        self.assertFalse(result["success"])
+        self.assertTrue(
+            any("untracked files" in r for r in result["reasons"]), result["reasons"]
+        )
+        self.assertIn("stray.txt", result["evidence"]["untracked_files"])
+        self.assertIsNone(self.remote_head())
+
+    def test_unexpected_remote_head_refuses(self):
+        """A remote head that is not an ancestor must never be overwritten."""
+        self.write_lock()
+        # Publish a divergent commit to the branch from a separate line.
+        self._git("checkout", "-q", "-b", "divergent", self.base_sha)
+        with open(os.path.join(self.repo, "other.txt"), "w") as fh:
+            fh.write("someone else's work\n")
+        self._git("add", "other.txt")
+        self._git("commit", "-q", "-m", "divergent")
+        divergent = self._git("rev-parse", "HEAD").stdout.strip()
+        self._git("push", "-q", GIT_REMOTE, f"{divergent}:refs/heads/{BRANCH}")
+        self._git("checkout", "-q", BRANCH)
+
+        result = self.run_publish()
+
+        self.assertFalse(result["success"])
+        self.assertTrue(
+            any("not an ancestor" in r for r in result["reasons"]), result["reasons"]
+        )
+        self.assertEqual(
+            self.remote_head(), divergent, "the other head must survive intact"
+        )
+
+    def test_fast_forward_remote_head_is_allowed(self):
+        """An ancestor head is an honest fast-forward, not a conflict."""
+        self.write_lock()
+        self._git("push", "-q", GIT_REMOTE, f"{self.base_sha}:refs/heads/{BRANCH}")
+
+        result = self.run_publish()
+
+        self.assertTrue(result["success"], result.get("reasons"))
+        self.assertTrue(result["evidence"]["fast_forward_from_remote"])
+        self.assertEqual(self.remote_head(), self.head_sha)
+
+    def test_content_hash_mismatch_refuses(self):
+        self.write_lock()
+        wrong = {"work.txt": "0" * 64}
+
+        result = self.run_publish(expected_file_hashes=wrong)
+
+        self.assertFalse(result["success"])
+        self.assertTrue(
+            any("declared content hashes" in r for r in result["reasons"]),
+            result["reasons"],
+        )
+        self.assertFalse(result["evidence"]["file_hashes_verified"])
+        self.assertIsNone(self.remote_head())
+
+    def test_matching_content_hashes_publish(self):
+        self.write_lock()
+        digests = branch_publish.hash_worktree_files(self.worktree, ["work.txt"])
+
+        result = self.run_publish(expected_file_hashes=digests)
+
+        self.assertTrue(result["success"], result.get("reasons"))
+        self.assertTrue(result["evidence"]["file_hashes_verified"])
+
+    def test_missing_declared_file_refuses(self):
+        self.write_lock()
+        result = self.run_publish(expected_file_hashes={"absent.txt": "0" * 64})
+        self.assertFalse(result["success"])
+        self.assertTrue(
+            any("missing or unreadable" in r for r in result["reasons"]),
+            result["reasons"],
+        )
+
+    def test_foreign_claimant_refuses(self):
+        """Ownership comes from the durable record, not from the caller."""
+        self.write_lock(claimant={"username": "someone-else", "profile": PROFILE})
+
+        result = self.run_publish()
+
+        self.assertFalse(result["success"])
+        self.assertTrue(
+            any("foreign claim" in r for r in result["reasons"]), result["reasons"]
+        )
+        self.assertIsNone(self.remote_head())
+
+    def test_foreign_profile_refuses(self):
+        self.write_lock(
+            claimant={"username": IDENTITY, "profile": "test-reviewer-prgs"}
+        )
+        result = self.run_publish()
+        self.assertFalse(result["success"])
+        self.assertTrue(
+            any("claimant profile" in r for r in result["reasons"]), result["reasons"]
+        )
+
+    def test_absent_lock_record_refuses(self):
+        """No recorded claim means this cannot be used to bypass the lock."""
+        result = self.run_publish()  # no write_lock()
+
+        self.assertFalse(result["success"])
+        self.assertTrue(
+            any("no durable issue-lock record" in r for r in result["reasons"]),
+            result["reasons"],
+        )
+        self.assertIsNone(self.remote_head())
+
+    def test_branch_mismatch_against_lock_refuses(self):
+        self.write_lock(branch_name=f"feat/issue-{ISSUE}-different")
+        result = self.run_publish()
+        self.assertFalse(result["success"])
+        self.assertTrue(
+            any("records branch" in r for r in result["reasons"]), result["reasons"]
+        )
+
+    def test_worktree_mismatch_against_lock_refuses(self):
+        self.write_lock(worktree_path="/tmp/some/other/worktree")
+        result = self.run_publish()
+        self.assertFalse(result["success"])
+        self.assertTrue(
+            any("records worktree" in r for r in result["reasons"]), result["reasons"]
+        )
+
+    def test_competing_open_pr_on_another_branch_refuses(self):
+        self.write_lock()
+        competing = [{"number": 4242, "head": {"ref": f"fix/issue-{ISSUE}-rival"}}]
+
+        result = self.run_publish(open_prs=competing)
+
+        self.assertFalse(result["success"])
+        self.assertTrue(
+            any("already claim issue" in r for r in result["reasons"]),
+            result["reasons"],
+        )
+        self.assertIsNone(self.remote_head())
+
+    def test_open_pr_on_the_same_branch_is_not_competing(self):
+        """This branch's own PR is not a rival claim against itself."""
+        self.write_lock()
+        own = [{"number": 77, "head": {"ref": BRANCH}}]
+
+        result = self.run_publish(open_prs=own)
+
+        self.assertTrue(result["success"], result.get("reasons"))
+
+
+class TestGuardStrictnessPreserved(_PublishBase):
+    """AC15 — publication is an operation, never a weakening of the guards."""
+
+    def test_non_issue_branch_refuses(self):
+        self._git("checkout", "-q", "-b", "scratch/not-issue-linked")
+        self.write_lock(branch_name="scratch/not-issue-linked")
+
+        result = self.run_publish(branch_name="scratch/not-issue-linked")
+
+        self.assertFalse(result["success"])
+        self.assertTrue(
+            any("issue-linked" in r for r in result["reasons"]), result["reasons"]
+        )
+
+    def test_stable_branch_refuses(self):
+        self.write_lock(branch_name="master")
+        result = self.run_publish(branch_name="master")
+        self.assertFalse(result["success"])
+        self.assertTrue(
+            any("issue-linked" in r or "stable branch" in r for r in result["reasons"]),
+            result["reasons"],
+        )
+
+    def test_branch_number_must_match_the_issue(self):
+        other = "feat/issue-7777-mismatched"
+        self._git("checkout", "-q", "-b", other)
+        self.write_lock(branch_name=other)
+        result = self.run_publish(branch_name=other)
+        self.assertFalse(result["success"])
+        self.assertTrue(
+            any("does not carry issue number" in r for r in result["reasons"]),
+            result["reasons"],
+        )
+
+    def test_unregistered_worktree_refuses(self):
+        """#713 — an improvised directory is not a registered worktree."""
+        path = self.write_lock()
+        assessment = branch_publish.assess_unpublished_commit_publication(
+            issue_lock_store.read_lock_file(path),
+            issue_number=ISSUE, branch_name=BRANCH, worktree_path=self.worktree,
+            expected_head=self.head_sha, remote="prgs", org=ORG, repo=REPO,
+            identity=IDENTITY, profile=PROFILE,
+            worktree_state={
+                "current_branch": BRANCH, "porcelain_status": "",
+                "head_sha": self.head_sha,
+            },
+            worktree_registered=False,
+            remote_probe={"probe_ok": True, "remote_branch_exists": False},
+        )
+        self.assertEqual(assessment["outcome"], branch_publish.REFUSED)
+        self.assertTrue(
+            any("not listed in git worktree list" in r
+                for r in assessment["reasons"]),
+            assessment["reasons"],
+        )
+
+    def test_unobservable_remote_refuses(self):
+        """An unknown remote state must not be mistaken for an absent branch."""
+        self.write_lock()
+        result = self.run_publish(git_remote_name="no-such-remote")
+        self.assertFalse(result["success"])
+        self.assertTrue(
+            any("could not be observed" in r for r in result["reasons"]),
+            result["reasons"],
+        )
+
+
+class TestRecordSeparation(_PublishBase):
+    """AC23 — the durable issue lock and the workflow lease are distinct."""
+
+    def test_publication_leaves_the_issue_lock_byte_identical(self):
+        path = self.write_lock()
+        with open(path, "rb") as fh:
+            before = fh.read()
+
+        result = self.run_publish()
+
+        self.assertTrue(result["success"], result.get("reasons"))
+        with open(path, "rb") as fh:
+            after = fh.read()
+        self.assertEqual(before, after, "publication must not mutate the lock record")
+        self.assertFalse(result["issue_lock_record_mutated"])
+        self.assertFalse(result["workflow_lease_touched"])
+
+    def test_refusal_also_reports_untouched_records(self):
+        result = self.run_publish()  # refuses: no lock record
+        self.assertFalse(result["issue_lock_record_mutated"])
+        self.assertFalse(result["workflow_lease_touched"])
+
+    def test_lock_generation_is_not_advanced(self):
+        path = self.write_lock()
+        self.run_publish()
+        lock = issue_lock_store.read_lock_file(path)
+        self.assertEqual(lock["lock_generation"], 1)
+
+
+class TestTruthfulProcessEvidence(_PublishBase):
+    """AC24 — a live daemon pid is never represented as a dead process."""
+
+    def test_live_recorded_pid_does_not_block_publication(self):
+        # The recorded pid is this live process, standing in for the live MCP
+        # daemon. Reclaim would refuse here; publication legitimately does not.
+        path = self.write_lock(session_pid=os.getpid())
+        lock = issue_lock_store.read_lock_file(path)
+        self.assertEqual(lock["pid"], os.getpid())
+
+        result = self.run_publish()
+
+        self.assertTrue(result["success"], result.get("reasons"))
+        self.assertEqual(self.remote_head(), self.head_sha)
+
+    def test_liveness_is_not_consulted_as_evidence(self):
+        self.write_lock(session_pid=os.getpid())
+        result = self.run_publish()
+        self.assertFalse(result["evidence"]["owner_pid_liveness_consulted"])
+
+    def test_reclaim_still_refuses_for_the_same_live_pid(self):
+        """Publication does not soften the reclaim predicate it routes around."""
+        path = self.write_lock(session_pid=os.getpid())
+        lock = issue_lock_store.read_lock_file(path)
+        reclaim = issue_lock_store.assess_expired_lock_reclaim(lock)
+        self.assertFalse(reclaim["reclaim_allowed"])
+
+
+class TestIdempotentRetry(_PublishBase):
+    """AC20 — retry is safe and read-after-write is proven every time."""
+
+    def test_second_publication_reports_already_published(self):
+        self.write_lock()
+        first = self.run_publish()
+        self.assertTrue(first["performed"])
+
+        second = self.run_publish()
+
+        self.assertTrue(second["success"], second.get("reasons"))
+        self.assertFalse(second["performed"], "no second push is needed")
+        self.assertTrue(second["published"])
+        self.assertTrue(second["verified"])
+        self.assertEqual(second["outcome"], branch_publish.ALREADY_PUBLISHED)
+        self.assertEqual(self.remote_head(), self.head_sha)
+
+
+class TestDryRun(_PublishBase):
+    """AC12 — dry run reports the decision and mutates nothing."""
+
+    def test_dry_run_reports_intent_without_publishing(self):
+        self.write_lock()
+
+        result = self.run_publish(dry_run=True)
+
+        self.assertTrue(result["success"])
+        self.assertTrue(result["dry_run"])
+        self.assertTrue(result["would_publish"])
+        self.assertFalse(result["performed"])
+        self.assertIsNone(self.remote_head(), "dry run must not publish")
+
+    def test_dry_run_and_apply_agree_on_a_refusal(self):
+        """AC11 — the reported decision does not depend on which mode ran."""
+        self.write_lock(claimant={"username": "someone-else", "profile": PROFILE})
+
+        dry = self.run_publish(dry_run=True)
+        applied = self.run_publish()
+
+        self.assertFalse(dry["success"])
+        self.assertFalse(applied["success"])
+        self.assertEqual(dry["reasons"], applied["reasons"])
+
+
+class TestRenewalUnblocked(_PublishBase):
+    """AC20/AC21 — renewal is permitted only after verified publication."""
+
+    def _renewal(self, remote_head):
+        return issue_lock_renewal.assess_exact_owner_lease_renewal(
+            issue_lock_store.read_lock_file(self.lock_path()),
+            issue_number=ISSUE, branch_name=BRANCH, worktree_path=self.worktree,
+            remote="prgs", org=ORG, repo=REPO,
+            identity=IDENTITY, profile=PROFILE,
+            current_branch=BRANCH, porcelain_status="", worktree_exists=True,
+            head_sha=self.head_sha, remote_head_sha=remote_head,
+        )
+
+    def test_renewal_refuses_before_publication(self):
+        self.write_lock()
+        decision = self._renewal(None)
+        self.assertFalse(decision["renewal_sanctioned"])
+        self.assertTrue(
+            any("unpublished branch" in r for r in decision["reasons"]),
+            decision["reasons"],
+        )
+
+    def test_renewal_is_sanctioned_after_publication(self):
+        self.write_lock()
+        result = self.run_publish()
+        self.assertTrue(result["verified"], result.get("reasons"))
+
+        decision = self._renewal(self.remote_head())
+
+        self.assertTrue(decision["renewal_sanctioned"], decision["reasons"])
+
+
+class TestProtectedAssetUntouched(unittest.TestCase):
+    """AC17 — no test or fixture may reference the protected worktree."""
+
+    def test_no_reference_to_the_protected_worktree(self):
+        here = os.path.dirname(os.path.abspath(__file__))
+        root = os.path.dirname(here)
+        needle = "issue-635-project-registry" + "-api"
+        for path in (
+            os.path.join(here, "test_issue_812_publish_unpublished_commit.py"),
+            os.path.join(root, "branch_publish.py"),
+        ):
+            with open(path, "r", encoding="utf-8") as fh:
+                body = fh.read()
+            self.assertNotIn(needle, body)
+
+
+if __name__ == "__main__":  # pragma: no cover
+    unittest.main()

tests/test_task_capability_role_invariants.py

@@ -139,6 +139,9 @@ EXPECTED_ROLE_EXCLUSIVE_TASKS = frozenset(
         "gitea_release_merger_pr_lease",
         "create_branch",
         "push_branch",
+        # #812 AC20: publishing an unpublished local head is author-only for the
+        # same reason every other push is — it writes a branch to the remote.
+        "publish_unpublished_branch",
         "create_pr",
         "commit_files",
         "gitea_commit_files",

tests/test_webui_shell.py

@@ -0,0 +1,135 @@
+"""Tests for the Phase 1 operator console application shell (#638)."""
+import sys
+import unittest
+from pathlib import Path
+
+sys.path.insert(0, str(Path(__file__).resolve().parent.parent))
+
+from starlette.routing import Route
+from starlette.testclient import TestClient
+
+from webui import layout
+from webui.app import create_app
+from webui.nav import NAV_GROUPS, STUB_PAGES, nav_hrefs
+
+
+class TestShellNav(unittest.TestCase):
+    def setUp(self):
+        self.client = TestClient(create_app())
+
+    def test_nav_group_labels_present(self):
+        text = self.client.get("/").text
+        for group in NAV_GROUPS:
+            with self.subTest(group=group.label):
+                self.assertIn(f">{group.label}<", text)
+
+    def test_phase1_group_labels_cover_expected_ia(self):
+        labels = {group.label for group in NAV_GROUPS}
+        for expected in (
+            "Health",
+            "Traffic",
+            "Runtime/Sessions",
+            "Projects",
+            "Inventory",
+            "Timeline",
+            "Policy",
+            "Insights",
+        ):
+            with self.subTest(label=expected):
+                self.assertIn(expected, labels)
+
+    def test_every_nav_href_resolves_to_a_get_route(self):
+        app = create_app()
+        get_paths = {
+            route.path
+            for route in app.routes
+            if isinstance(route, Route) and "GET" in route.methods
+        }
+        for href in nav_hrefs():
+            with self.subTest(href=href):
+                self.assertIn(href, get_paths, f"nav href {href} has no GET route")
+
+    def test_legacy_hrefs_still_navigable(self):
+        text = self.client.get("/").text
+        for href in ("/queue", "/projects", "/prompts", "/runtime",
+                     "/audit", "/worktrees", "/leases", "/actions"):
+            with self.subTest(href=href):
+                self.assertIn(f'href="{href}"', text)
+
+
+class TestShellBadges(unittest.TestCase):
+    def setUp(self):
+        self.client = TestClient(create_app())
+
+    def test_mode_badge_present(self):
+        self.assertIn("mode: read-only", self.client.get("/").text)
+
+    def test_environment_badge_present(self):
+        self.assertIn("env:", self.client.get("/").text)
+
+    def test_default_environment_is_local(self):
+        self.assertEqual(layout.environment_label(), "local")
+
+    def test_remote_bind_reports_remote_environment(self):
+        import os
+
+        prior = os.environ.get("WEBUI_HOST")
+        os.environ["WEBUI_HOST"] = "10.0.0.5"
+        try:
+            self.assertEqual(layout.environment_label(), "remote")
+        finally:
+            if prior is None:
+                os.environ.pop("WEBUI_HOST", None)
+            else:
+                os.environ["WEBUI_HOST"] = prior
+
+    def test_docs_link_present(self):
+        text = self.client.get("/").text
+        self.assertIn(layout.DOCS_URL, text)
+        self.assertIn(">Docs<", text)
+
+
+class TestShellStubs(unittest.TestCase):
+    def setUp(self):
+        self.client = TestClient(create_app())
+
+    def test_stub_routes_render_200(self):
+        for path, (title, _desc) in STUB_PAGES.items():
+            with self.subTest(path=path):
+                response = self.client.get(path)
+                self.assertEqual(response.status_code, 200, path)
+                self.assertIn(title, response.text)
+                self.assertIn("placeholder", response.text)
+
+    def test_stub_routes_are_read_only(self):
+        for path in STUB_PAGES:
+            with self.subTest(path=path):
+                response = self.client.post(path)
+                self.assertEqual(response.status_code, 405)
+                self.assertEqual(response.json()["error"], "read-only-mvp")
+
+    def test_stub_pages_carry_nav_and_badges(self):
+        response = self.client.get("/inventory")
+        self.assertIn("mode: read-only", response.text)
+        self.assertIn('href="/queue"', response.text)
+
+
+class TestShellHome(unittest.TestCase):
+    def setUp(self):
+        self.client = TestClient(create_app())
+
+    def test_home_summarizes_console(self):
+        text = self.client.get("/").text
+        self.assertIn("Operator console", text)
+        self.assertIn("Phase 1", text)
+
+    def test_home_links_legacy_pages(self):
+        text = self.client.get("/").text
+        self.assertIn("MVP legacy pages", text)
+        for href in ("/queue", "/audit", "/leases"):
+            with self.subTest(href=href):
+                self.assertIn(f'href="{href}"', text)
+
+
+if __name__ == "__main__":
+    unittest.main()

webui/app.py

@@ -11,6 +11,7 @@ from starlette.routing import Route
 
 from webui.deployment_boundary import deployment_snapshot
 from webui.layout import render_page
+from webui.nav import NAV_GROUPS, STUB_PAGES
 from webui.project_registry import find_project, load_registry, registry_to_dict
 from webui.project_views import render_project_detail, render_projects_list
 from webui.prompt_library import find_prompt, library_to_dict
@@ -43,24 +44,62 @@ def _stub_page(title: str, description: str) -> HTMLResponse:
     return HTMLResponse(render_page(title=title, body_html=body))
 
 
+_LEGACY_PAGES = (
+    ("/queue", "Queue", "live PR and issue dashboard (#429)"),
+    ("/projects", "Projects", "registry and onboarding (#427)"),
+    ("/prompts", "Prompts", "canonical workflow prompt library (#428)"),
+    ("/runtime", "Runtime", "MCP health and stale-runtime detection (#430)"),
+    ("/audit", "Audit", "final-report paste and validator preview (#431)"),
+    ("/worktrees", "Worktrees", "branch hygiene dashboard (#432)"),
+    ("/leases", "Leases", "collision and lease visibility (#433)"),
+    ("/actions", "Actions", "gated write-action framework (#434)"),
+)
+
+
+def _render_home_nav_groups() -> str:
+    groups = []
+    for group in NAV_GROUPS:
+        items = "".join(
+            f'<li><a href="{item.href}">{item.label}</a>'
+            + ("" if item.status == "live" else " <span class=\"muted\">(stub)</span>")
+            + "</li>"
+            for item in group.items
+        )
+        groups.append(f"<h3>{group.label}</h3><ul>{items}</ul>")
+    return "".join(groups)
+
+
 async def home(_request: Request) -> HTMLResponse:
+    legacy = "".join(
+        f"<li><strong>{label}</strong> — {desc} "
+        f'(<a href="{href}">{href}</a>)</li>'
+        for href, label, desc in _LEGACY_PAGES
+    )
     body = (
         "<h2>Operator console</h2>"
-        "<p>Local entry point for MCP Control Plane operational views.</p>"
-        "<ul>"
-        "<li><strong>Queue</strong> — live PR and issue dashboard (#429)</li>"
-        "<li><strong>Projects</strong> — registry and onboarding (#427)</li>"
-        "<li><strong>Prompts</strong> — canonical workflow prompt library (#428)</li>"
-        "<li><strong>Runtime</strong> — MCP health and stale-runtime detection (#430)</li>"
-        "<li><strong>Audit</strong> — final-report paste and validator preview (#431)</li>"
-        "<li><strong>Worktrees</strong> — branch hygiene dashboard (#432)</li>"
-        "<li><strong>Leases</strong> — collision and lease visibility (#433)</li>"
-        "<li><strong>Actions</strong> — gated write-action framework (#434)</li>"
-        "</ul>"
+        "<p>Read-only home for the MCP Control Plane Phase 1 operator console. "
+        "Gitea, MCP capability gates, and canonical workflows remain the source "
+        "of truth; this console never mutates them.</p>"
+        "<h2>Phase 1 surfaces</h2>"
+        + _render_home_nav_groups()
+        + "<h2>MVP legacy pages</h2>"
+        "<ul>" + legacy + "</ul>"
     )
     return HTMLResponse(render_page(title="Home", body_html=body))
 
 
+async def phase_stub(request: Request) -> HTMLResponse:
+    """Graceful read-only placeholder for a not-yet-implemented Phase 1 surface."""
+    title, description = STUB_PAGES[request.url.path]
+    body = (
+        f"<h2>{title}</h2>"
+        f'<div class="stub"><p>{description}</p>'
+        "<p>Phase 1 shell placeholder — no write actions. Tracked under "
+        "epic #631.</p></div>"
+    )
+    return HTMLResponse(render_page(title=title, body_html=body))
+
+
 async def health(_request: Request) -> JSONResponse:
     bind_host = _request.app.state.webui_bind_host
     return JSONResponse({
@@ -291,6 +330,10 @@ def create_app(*, bind_host: str | None = None) -> Starlette:
                 methods=["POST"],
             ),
             Route("/api/leases", api_leases, methods=["GET"]),
+            *[
+                Route(path, phase_stub, methods=["GET"])
+                for path in STUB_PAGES
+            ],
         ],
         exception_handlers={405: method_not_allowed},
     )

webui/layout.py

@@ -2,28 +2,66 @@
 
 from __future__ import annotations
 
-NAV_ITEMS = (
-    ("/", "Home"),
-    ("/queue", "Queue"),
-    ("/projects", "Projects"),
-    ("/prompts", "Prompts"),
-    ("/runtime", "Runtime"),
-    ("/audit", "Audit"),
-    ("/worktrees", "Worktrees"),
-    ("/leases", "Leases"),
-    ("/actions", "Actions"),
-)
+import os
+
+from webui.nav import NAV_GROUPS
 
 MVP_NOTICE = (
     "Read-only MVP — Gitea, MCP tools, and canonical workflows remain the "
     "source of truth. No mutation endpoints."
 )
 
+# Canonical docs entry point surfaced from the shell header (#638).
+DOCS_URL = (
+    "https://gitea.prgs.cc/Scaled-Tech-Consulting/Gitea-Tools/src/branch/"
+    "master/docs/webui-local-dev.md"
+)
+
+_LOCAL_HOSTS = frozenset({"", "127.0.0.1", "localhost", "::1"})
+
+
+def environment_label() -> str:
+    """Classify the serving environment as ``local`` or ``remote`` (#638).
+
+    Derived from the same ``WEBUI_HOST`` default the app binds to; loopback
+    hosts are ``local``, anything else is ``remote``. Read-only signal only.
+    """
+    host = (os.environ.get("WEBUI_HOST", "127.0.0.1") or "").strip().lower()
+    return "local" if host in _LOCAL_HOSTS else "remote"
+
+
+def _render_nav() -> str:
+    groups_html = []
+    for group in NAV_GROUPS:
+        links = "".join(
+            f'<a href="{item.href}"'
+            + (' class="nav-stub"' if item.status == "stub" else "")
+            + f'>{item.label}</a>'
+            for item in group.items
+        )
+        groups_html.append(
+            '<div class="nav-group">'
+            f'<span class="nav-group-label">{group.label}</span>'
+            f'<span class="nav-group-links">{links}</span>'
+            "</div>"
+        )
+    return "".join(groups_html)
+
+
+def _render_badges() -> str:
+    env = environment_label()
+    return (
+        '<div class="header-badges">'
+        f'<span class="badge env-badge env-{env}">env: {env}</span>'
+        '<span class="badge mode-badge">mode: read-only</span>'
+        f'<a class="badge docs-link" href="{DOCS_URL}">Docs</a>'
+        "</div>"
+    )
+
 
 def render_page(*, title: str, body_html: str, extra_head: str = "") -> str:
-    nav_links = "".join(
-        f'<a href="{href}">{label}</a>' for href, label in NAV_ITEMS
-    )
+    nav_links = _render_nav()
+    header_badges = _render_badges()
     return f"""<!DOCTYPE html>
 <html lang="en">
 <head>
@@ -53,21 +91,58 @@ def render_page(*, title: str, body_html: str, extra_head: str = "") -> str:
       padding: 0.75rem 1.25rem;
     }}
     header h1 {{
-      margin: 0 0 0.5rem;
+      margin: 0;
       font-size: 1.1rem;
       font-weight: 600;
     }}
+    .header-top {{
+      display: flex;
+      flex-wrap: wrap;
+      align-items: center;
+      justify-content: space-between;
+      gap: 0.5rem 1rem;
+      margin-bottom: 0.6rem;
+    }}
+    .header-badges {{ display: inline-flex; flex-wrap: wrap; gap: 0.4rem; }}
+    .env-badge.env-local {{ color: #8fd19e; border-color: #3d6b4a; }}
+    .env-badge.env-remote {{ color: #e0c27a; border-color: #6b5730; }}
+    .mode-badge {{ color: #9ec8f0; border-color: #3d5f7a; }}
+    a.docs-link {{
+      color: var(--accent);
+      border-color: var(--accent);
+      text-decoration: none;
+      text-transform: none;
+    }}
+    a.docs-link:hover {{ filter: brightness(1.12); }}
     nav {{
       display: flex;
       flex-wrap: wrap;
-      gap: 0.75rem 1rem;
+      gap: 0.5rem 1.25rem;
     }}
+    .nav-group {{
+      display: flex;
+      flex-direction: column;
+      gap: 0.15rem;
+    }}
+    .nav-group-label {{
+      font-size: 0.68rem;
+      text-transform: uppercase;
+      letter-spacing: 0.04em;
+      color: var(--muted);
+    }}
+    .nav-group-links {{ display: inline-flex; flex-wrap: wrap; gap: 0.6rem; }}
     nav a {{
       color: var(--accent);
       text-decoration: none;
       font-size: 0.9rem;
     }}
     nav a:hover {{ text-decoration: underline; }}
+    nav a.nav-stub {{ color: var(--muted); }}
+    nav a.nav-stub::after {{
+      content: " ·stub";
+      font-size: 0.7rem;
+      color: var(--muted);
+    }}
     main {{
       max-width: 52rem;
       margin: 0 auto;
@@ -166,7 +241,10 @@ def render_page(*, title: str, body_html: str, extra_head: str = "") -> str:
 </head>
 <body>
   <header>
-    <h1>MCP Control Plane</h1>
+    <div class="header-top">
+      <h1>MCP Control Plane</h1>
+      {header_badges}
+    </div>
     <nav>{nav_links}</nav>
   </header>
   <main>

webui/nav.py

@@ -0,0 +1,111 @@
+"""Navigation IA for the Phase 1 operator console shell (#638).
+
+Single source of truth for the console navigation so ``webui/layout.py`` and
+the ``webui/app.py`` route table stay aligned with epic #631. Read-only: every
+destination is a GET view or a Phase 1 placeholder. No mutation links.
+
+Nav groups follow the #631 Phase 1 information architecture: Health, Traffic,
+Runtime/Sessions, Projects, Inventory, Timeline, Policy (placeholder), and
+Insights (placeholder). Later-phase surfaces are declared as ``stub`` items and
+backed by ``STUB_PAGES`` so their nav links resolve to a graceful placeholder
+instead of a 404.
+"""
+
+from __future__ import annotations
+
+from dataclasses import dataclass
+
+
+@dataclass(frozen=True)
+class NavItem:
+    """A single navigation destination.
+
+    ``status`` is ``"live"`` for implemented views and ``"stub"`` for Phase 1
+    placeholders whose backing view lands in a later child issue.
+    """
+
+    href: str
+    label: str
+    status: str = "live"
+
+
+@dataclass(frozen=True)
+class NavGroup:
+    label: str
+    items: tuple[NavItem, ...]
+
+
+NAV_GROUPS: tuple[NavGroup, ...] = (
+    NavGroup("Health", (
+        NavItem("/health", "Liveness"),
+    )),
+    NavGroup("Traffic", (
+        NavItem("/queue", "Queue"),
+        NavItem("/leases", "Leases"),
+        NavItem("/actions", "Actions"),
+    )),
+    NavGroup("Runtime/Sessions", (
+        NavItem("/runtime", "Runtime health"),
+        NavItem("/sessions", "Sessions", "stub"),
+    )),
+    NavGroup("Projects", (
+        NavItem("/projects", "Projects"),
+    )),
+    NavGroup("Inventory", (
+        NavItem("/inventory", "Inventory", "stub"),
+        NavItem("/worktrees", "Worktrees"),
+    )),
+    NavGroup("Timeline", (
+        NavItem("/timeline", "Timeline", "stub"),
+    )),
+    NavGroup("Policy", (
+        NavItem("/policy", "Policy", "stub"),
+        NavItem("/prompts", "Prompts"),
+    )),
+    NavGroup("Insights", (
+        NavItem("/insights", "Insights", "stub"),
+        NavItem("/audit", "Audit"),
+    )),
+)
+
+
+# Phase 1 placeholder destinations whose backing views land in later child
+# issues of epic #631. Each maps a path to (title, description). Routes are
+# registered so nav links resolve to a graceful, read-only stub page.
+STUB_PAGES: dict[str, tuple[str, str]] = {
+    "/sessions": (
+        "Sessions",
+        "Active session, capability, and role inventory. Backed by the unified "
+        "inventory API (#636) once it lands.",
+    ),
+    "/inventory": (
+        "Inventory",
+        "Unified sessions, leases, locks, namespaces, and worktree inventory. "
+        "Backed by the Phase 1 inventory API (#636).",
+    ),
+    "/timeline": (
+        "Timeline",
+        "Workflow event timeline across issues and PRs. A later Phase 1 surface.",
+    ),
+    "/policy": (
+        "Policy",
+        "Capability and role policy surface. Placeholder until a later phase.",
+    ),
+    "/insights": (
+        "Insights",
+        "Aggregate operational insights and trends. Placeholder until a later "
+        "phase.",
+    ),
+}
+
+
+def iter_nav_items():
+    """Yield every ``NavItem`` across all groups in declared order."""
+    for group in NAV_GROUPS:
+        for item in group.items:
+            yield item
+
+
+def nav_hrefs() -> tuple[str, ...]:
+    """Return every navigation href in declared order."""
+    return tuple(item.href for item in iter_nav_items())