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]>

docs/webui-local-dev.md

@@ -43,10 +43,6 @@ for the console architecture: layer and authority boundaries, the redaction
 boundary, `/api/v1/...` versioning, the target page map, and the phase gates
 that govern when a write path may open (#632, epic #631).
 
-See [webui-project-registry-api.md](webui-project-registry-api.md) for the
-versioned project registry contract: registry schema versions 1 and 2, project
-status, onboarding checklist state, and the fail-closed error payloads (#635).
-
 ## Routes (MVP)
 
 | Path | Description |
@@ -55,11 +51,9 @@ status, onboarding checklist state, and the fail-closed error payloads (#635).
 | `/health` | JSON liveness (`status`, `service`, `mode`, `timestamp`) |
 | `/queue` | Live PR and issue queue dashboard (#429) |
 | `/api/queue` | JSON queue export with pagination metadata |
-| `/projects` | Project registry list with status and onboarding progress (#427, #635) |
+| `/projects` | Project registry list (#427) |
 | `/projects/{id}` | Project detail + onboarding checklist |
-| `/api/v1/projects` | Versioned JSON registry export (#635) |
+| `/api/projects` | JSON registry export |
-| `/api/v1/projects/{id}` | Versioned JSON project detail (#635) |
-| `/api/projects` | JSON registry export — unversioned Phase 1 alias of `/api/v1/projects` |
 | `/prompts` | Prompt library with per-prompt copy buttons (#428) |
 | `/api/prompts` | JSON prompt export with workflow hashes |
 | `/runtime` | MCP runtime health and stale detection (#430) |
@@ -73,6 +67,11 @@ status, onboarding checklist state, and the fail-closed error payloads (#635).
 | `/api/actions/{id}/preview` | Mutation ledger preview (GET, read-only) |
 | `/leases` | Lease and collision visibility (#433) |
 | `/api/leases` | JSON lease/collision export |
+| `/sessions` | Phase 1 shell stub — session inventory (backed by #636) |
+| `/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
@@ -153,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**
@@ -212,53 +231,6 @@ 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.
 
-## Inventory API (#636)
-
-`GET /api/v1/inventory` returns one versioned, read-only snapshot that unifies
-what the lease (#433), worktree (#432), and runtime (#430) MVP views each show
-separately, so traffic-control and recovery consumers read the same source.
-`GET /api/v1/inventory/{section}` returns a single section under the identical
-schema (`sessions`, `leases`, `locks`, `worktrees`, `namespaces`); an unknown
-section is a `404` with `error: unknown_section`. Both routes are `GET`-only.
-
-Each section carries its own `status` (`ok` / `degraded` / `unavailable`), a
-`reason` when not `ok`, and a `scan_ms`. A subsystem that cannot be read
-degrades to a reasoned section; it never raises and never emits an empty list
-that would read as "nothing is there".
-
-### Field authority
-
-Every section names where its rows came from; authorities are never blended.
-
-| Section | Authority | Source |
-|---|---|---|
-| `sessions` | `control_plane_db` | #613 control-plane DB (`mode=ro`), authoritative for exclusive ownership (#600/#601) |
-| `leases` | `control_plane_db` | #613 control-plane DB; degrades if the `work_items` table is absent |
-| `locks` | `filesystem` | durable per-issue lock files (`issue_lock_store`) |
-| `worktrees` | `filesystem` | registered git worktrees via the #432 hygiene scanner |
-| `namespaces` | `filesystem` | the active profile serving this web process (others are not enumerable) |
-
-The payload restates this map under `field_authority` for machine consumers.
-
-### Ownership safety
-
-`ownership_authority_complete` is true only when every ownership-bearing section
-(`sessions`, `leases`, `locks`) read cleanly. While it is false, nothing is
-reported as unowned and no collision is asserted from a degraded source —
-absence of evidence is reported as absence of evidence, never as free work.
-
-`collisions` surfaces detectable conflicts, each with a `kind` and `severity`:
-`lock-without-worktree`, `duplicate-live-lock`, `live-lock-dead-owner` (unexpired
-lease, dead pid — a #753 recovery candidate that would read as live to a naive
-timestamp check), `stale-lock-dead-owner`, `expired-lock-live-owner` (the
-#635/#760 daemon-pid deadlock), `concurrent-active-lease`, `active-lease-past-expiry`,
-and `orphan-lease`. Collisions are emitted only from sections that read cleanly.
-
-The control-plane DB is opened through a `mode=ro` URI so a read never creates
-or migrates it; paths are collapsed against `$HOME`, URLs lose userinfo and
-query strings, and credential-shaped values are redacted at the boundary. Lease
-steal/release and worktree deletion are Phase 2+ and have no representation here.
-
 ## Tests
 
 ```bash

docs/webui-project-registry-api.md

@@ -1,213 +0,0 @@
-# Project registry API (#635)
-
-Phase 1 of the [console architecture ADR](architecture/webui-control-plane-console-architecture-adr.md)
-gives the project registry a versioned, read-only API. This document is the
-field-by-field contract for that API and for the registry file behind it.
-
-Everything here is **read-only**. The console never writes the registry; an
-operator edits the JSON file, and an invalid file fails closed rather than
-rendering a partial inventory.
-
-## Routes
-
-| Route | Method | Description |
-|-------|--------|-------------|
-| `/api/v1/projects` | GET | Versioned registry export: all projects, with provenance |
-| `/api/v1/projects/{project_id}` | GET | Single project; `404` with `project_not_found` when unknown |
-| `/api/projects` | GET | Unversioned MVP alias (#427), retained for all of Phase 1 |
-| `/projects` | GET | HTML list — status and onboarding progress per project |
-| `/projects/{project_id}` | GET | HTML detail — identity, profiles, paths, checklist |
-
-Per ADR section 6 the unversioned alias may be retired no earlier than Phase 2,
-and only after this document and `webui-local-dev.md` record the swap. The alias
-returns the same payload as `/api/v1/projects`, including the legacy `version`
-and `source_path` keys #427 consumers already read.
-
-The HTML views render from the same DTO the JSON routes serialize
-(`project_to_dict`), so the console and the API cannot disagree about a
-project's status or onboarding progress.
-
-## Registry file
-
-Default location: `webui/data/projects.registry.json`. Override with the
-`WEBUI_PROJECT_REGISTRY` environment variable.
-
-Schema versions: **1** and **2** are accepted; **2** is current. A version 1
-file loads unchanged and is normalized with the documented defaults, so an
-existing operator registry keeps working without edits.
-
-### Root
-
-| Field | Type | Required | Notes |
-|-------|------|----------|-------|
-| `version` | int | yes | `1` or `2`. Anything else fails closed |
-| `projects` | array | yes | Must be non-empty |
-
-### Project
-
-| Field | Type | Required | Default | Notes |
-|-------|------|----------|---------|-------|
-| `id` | string | yes | — | Stable registry id used in URLs |
-| `repo_name` | string | yes | — | Gitea repository name |
-| `gitea_owner` | string | yes | — | Owning org or user |
-| `remote_host` | string | yes | — | Instance base URL, no credentials |
-| `remote_name` | string | no | `null` | Logical remote label, e.g. `prgs` (v2) |
-| `default_branch` | string | yes | — | Stable branch name |
-| `local_checkout_path` | string | yes | — | Control checkout path |
-| `status` | string | no | `active` | `active`, `onboarding`, `paused`, `archived` (v2) |
-| `profiles` | object | yes | — | Must map `author`, `reviewer`, `reconciler` |
-| `workflow_paths` | object | yes | — | Non-empty; label to repo-relative path |
-| `schema_paths` | object | no | `{}` | Label to repo-relative path |
-| `onboarding_checklist` | array | no | `[]` | See below |
-| `last_seen_health` | object | no | `null` | Redacted health only (v2) |
-
-### Onboarding step
-
-| Field | Type | Required | Default | Notes |
-|-------|------|----------|---------|-------|
-| `id` | string | yes | — | Stable step id |
-| `title` | string | yes | — | Short operator-facing label |
-| `description` | string | yes | — | Self-contained; assumes no chat history |
-| `state` | string | no | `pending` | `complete`, `pending`, `blocked`, `not_applicable` (v2) |
-| `required` | bool | no | `true` | Optional steps never block readiness (v2) |
-
-### Last-seen health
-
-| Field | Type | Required | Notes |
-|-------|------|----------|-------|
-| `status` | string | no (default `unknown`) | `healthy`, `degraded`, `unreachable`, `unknown` |
-| `checked_at` | string | no | ISO-8601 UTC timestamp, e.g. `2026-01-01T00:00:00Z` |
-| `detail` | string | no | Short redacted note |
-
-Health is recorded metadata, not a live probe: Phase 1 performs no outbound
-health checks. Endpoints, tokens, and keychain identifiers must never appear
-here.
-
-## Response shape
-
-`GET /api/v1/projects`:
-
-```json
-{
-  "api_version": "v1",
-  "schema_version": 2,
-  "version": 2,
-  "source_path": "/path/to/webui/data/projects.registry.json",
-  "source": {
-    "kind": "file",
-    "path": "/path/to/webui/data/projects.registry.json",
-    "inventory_complete": true
-  },
-  "project_count": 1,
-  "projects": [
-    {
-      "id": "example",
-      "repo_name": "Example",
-      "gitea_owner": "Org",
-      "repo_full_name": "Org/Example",
-      "remote_host": "https://gitea.example.invalid",
-      "remote_name": "example-remote",
-      "default_branch": "main",
-      "local_checkout_path": ".",
-      "status": "active",
-      "profiles": {"author": "...", "reviewer": "...", "reconciler": "..."},
-      "workflow_paths": {"skill": "skills/..."},
-      "schema_paths": {},
-      "onboarding_checklist": [
-        {
-          "id": "profiles",
-          "title": "Configure execution profiles",
-          "description": "...",
-          "state": "complete",
-          "required": true
-        }
-      ],
-      "onboarding_summary": {
-        "total": 1,
-        "complete": 1,
-        "pending": 0,
-        "blocked": 0,
-        "not_applicable": 0,
-        "required_outstanding": 0,
-        "onboarding_complete": true
-      },
-      "last_seen_health": null
-    }
-  ]
-}
-```
-
-`GET /api/v1/projects/{project_id}` returns `api_version`, `schema_version`,
-`source`, and a single `project` object with the same fields.
-
-The `source` block satisfies the ADR section 6 provenance rule: every payload
-states where the data came from and whether the inventory is complete. A
-file-backed registry is always complete — there is no pagination to truncate it.
-
-`onboarding_summary` is derived, never stored. `required_outstanding` counts
-steps that are `required` **and** in state `pending` or `blocked`;
-`onboarding_complete` is true when that count is zero.
-
-## Fail-closed errors
-
-Validation failures raise `RegistryError`, which routes render instead of a
-traceback.
-
-`404` — unknown project id on `/api/v1/projects/{project_id}`:
-
-```json
-{
-  "error": "project_not_found",
-  "project_id": "not-registered",
-  "known_project_ids": ["example"],
-  "remediation": "Request one of the known project ids, or add the project ...",
-  "source": {"kind": "file", "path": "...", "inventory_complete": true}
-}
-```
-
-`500` — invalid registry, on both the versioned route and the alias:
-
-```json
-{
-  "error": "registry_invalid",
-  "detail": "unsupported registry version: 42",
-  "remediation": "Set 'version' to one of 1, 2 (current schema is 2) ...",
-  "field_path": "version",
-  "source_path": "/path/to/registry.json"
-}
-```
-
-`field_path` points at the offending location (`projects[0].profiles.reconciler`,
-`projects[0].onboarding_checklist[2].state`, and so on). The HTML routes render
-the same detail, field, source, and remediation on a "Project registry
-unavailable" page.
-
-Conditions that fail closed:
-
-* file missing or unreadable;
-* invalid JSON (the remediation names line and column);
-* root not an object, or `projects` missing/empty;
-* unsupported `version`;
-* a credential-shaped key anywhere in the file (`token`, `*_secret`, `auth_*`, and similar);
-* a project missing a required field, or missing an `author`/`reviewer`/`reconciler` profile;
-* an unknown `status`, onboarding `state`, or health `status`.
-
-## Credential rule
-
-The registry stores redacted metadata only. Credential-shaped keys are
-rejected at load time, before any DTO is built, consistent with
-[safety-model.md](safety-model.md) and
-[credential-isolation.md](credential-isolation.md). Tokens live in the keychain
-and are resolved server-side by `gitea_auth`.
-
-## Migrating a version 1 registry
-
-1. Set `"version": 2`.
-2. Optionally add `"status"` per project (omitted means `active`).
-3. Optionally add `"remote_name"` per project.
-4. Optionally add `"state"` and `"required"` to each onboarding step (omitted
-   means `pending` and `true`).
-5. Optionally add `"last_seen_health"`.
-
-No step is mandatory: a version 1 file keeps loading. Bumping the version only
-declares that the file may use the v2 fields.

gitea_mcp_server.py

@@ -9173,27 +9173,11 @@ def gitea_publish_unpublished_issue_branch(
     if blocked:
         return blocked
 
-    # #815: resolve the caller's worktree *before* preflight and forward it, so
+    verify_preflight_purity(remote, task=task, org=org, repo=repo)
-    # every workspace-resolution layer behind verify_preflight_purity — including
-    # the #618 branches-only guard — judges the registered issue worktree this
-    # publication actually operates on. Resolving it afterwards let preflight
-    # fall back to the MCP process root, so a daemon rooted at the stable control
-    # checkout refused a valid explicit worktree before the assessor ever ran.
-    # A caller supplying nothing usable forwards None and keeps the ordinary
-    # fail-closed fallback.
-    explicit_worktree = (worktree_path or "").strip()
-    workspace = os.path.realpath(os.path.abspath(explicit_worktree or "."))
-
-    verify_preflight_purity(
-        remote,
-        worktree_path=workspace if explicit_worktree else None,
-        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)

tests/test_issue_815_preflight_worktree_forwarding.py

@@ -1,622 +0,0 @@
-"""Publication preflight must receive the caller's worktree (#815).
-
-``gitea_publish_unpublished_issue_branch`` takes a **required** ``worktree_path``
-but resolved it only *after* ``verify_preflight_purity`` had already run. Every
-workspace-resolution layer behind that preflight — canonical root, root checkout,
-create-issue bootstrap, the #618 branches-only guard, issue scope, and anti-stomp
-— therefore received ``None`` and fell back to the MCP process root. A daemon
-rooted at the stable control checkout refused a valid registered issue worktree
-that the caller had explicitly supplied, before the publication assessor ever ran.
-
-The #812 suite could not see this. Its fixture sets ``self.worktree =
-os.path.realpath(self.repo)`` and patches ``PROJECT_ROOT`` to that same path, so
-the fallback resolved to the very worktree the argument named. The production
-topology — control checkout on a stable branch, issue worktree somewhere else —
-was never constructed, and preflight additionally no-ops under pytest unless
-production guards are forced on.
-
-These tests build that topology honestly:
-
-* ``PROJECT_ROOT`` is a control checkout sitting on ``master``;
-* the registered issue worktree is a genuinely separate path under ``branches/``;
-* ``GITEA_TEST_FORCE_PRODUCTION_GUARDS`` is set so the #618 guard really runs;
-* no patch makes the issue worktree appear to be ``PROJECT_ROOT``.
-
-Nothing here reads, writes, or references the protected worktree named in #812
-AC17 and #815 AC9. Every fixture is built from scratch against a local bare
-remote, so publication and read-after-write verification genuinely execute.
-"""
-
-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 issue_lock_provenance  # noqa: E402
-import issue_lock_store  # noqa: E402
-import mcp_server  # noqa: E402
-from mutation_profile_fixture import shared_mutation_env  # noqa: E402
-
-ISSUE = 9815
-BRANCH = f"feat/issue-{ISSUE}-forwarding-fixture"
-WORKTREE_DIRNAME = BRANCH.replace("/", "-")
-IDENTITY = "example-user"
-PROFILE = "test-author-prgs"
-ORG = "Scaled-Tech-Consulting"
-REPO = "Gitea-Tools"
-GIT_REMOTE = "prgs"
-
-AUTHOR_PROFILE = {
-    "profile_name": "prgs-author",
-    "role": "author",
-    "allowed_operations": [
-        "gitea.read", "gitea.issue.create", "gitea.issue.comment",
-        "gitea.pr.create", "gitea.repo.commit", "gitea.branch.push",
-    ],
-    "forbidden_operations": [],
-    "audit_label": "prgs-author",
-}
-
-
-def _ts(hours: int) -> str:
-    return (
-        (datetime.now(timezone.utc) + timedelta(hours=hours))
-        .isoformat()
-        .replace("+00:00", "Z")
-    )
-
-
-class TestPreflightReceivesTheWorktree(unittest.TestCase):
-    """AC1 — the supplied path reaches ``verify_preflight_purity`` itself.
-
-    Follows the #735 capture pattern: replace preflight with a recorder that
-    raises, so the argument can be proven forwarded without performing the
-    mutation. This is the direct unit-level statement of the defect.
-    """
-
-    def _capture_preflight(self, **kwargs):
-        captured: dict = {}
-
-        def _capture(*a, **kw):
-            captured.update(kw)
-            captured["_args"] = a
-            raise RuntimeError("capture-only")
-
-        with patch.object(
-            mcp_server, "verify_preflight_purity", side_effect=_capture
-        ), patch.object(
-            mcp_server, "get_profile", return_value=AUTHOR_PROFILE
-        ), patch.object(
-            mcp_server, "_resolve",
-            return_value=("gitea.prgs.cc", ORG, REPO),
-        ), patch.object(
-            mcp_server, "_auth", return_value="token fake",
-        ), patch.object(
-            mcp_server.role_session_router,
-            "check_author_mutation_after_reviewer_stop",
-            return_value=(True, []),
-        ), patch.object(
-            mcp_server, "_namespace_mutation_block", return_value=None
-        ), patch.object(
-            mcp_server, "_profile_permission_block", return_value=None
-        ):
-            try:
-                mcp_server.gitea_publish_unpublished_issue_branch(**kwargs)
-            except RuntimeError as exc:
-                if "capture-only" not in str(exc) and not captured:
-                    raise
-        self.assertTrue(
-            captured,
-            "gitea_publish_unpublished_issue_branch never called "
-            "verify_preflight_purity",
-        )
-        return captured
-
-    def _base_kwargs(self, **overrides):
-        kwargs = {
-            "issue_number": ISSUE,
-            "branch_name": BRANCH,
-            "worktree_path": "/tmp/issue-815-explicit-worktree",
-            "expected_head": "a" * 40,
-            "remote": "prgs",
-            "org": ORG,
-            "repo": REPO,
-            "git_remote_name": GIT_REMOTE,
-        }
-        kwargs.update(overrides)
-        return kwargs
-
-    def test_explicit_worktree_path_reaches_preflight(self):
-        captured = self._capture_preflight(**self._base_kwargs())
-        self.assertEqual(
-            captured.get("worktree_path"),
-            os.path.realpath(os.path.abspath("/tmp/issue-815-explicit-worktree")),
-            "the authoritative worktree_path must be forwarded into preflight",
-        )
-
-    def test_forwarded_path_is_the_one_publication_uses(self):
-        """AC4 — preflight and publication must judge the same resolved path."""
-        raw = "/tmp/issue-815-explicit-worktree/./"
-        captured = self._capture_preflight(**self._base_kwargs(worktree_path=raw))
-        expected = os.path.realpath(os.path.abspath(raw.strip()))
-        self.assertEqual(captured.get("worktree_path"), expected)
-
-    def test_blank_worktree_path_forwards_none(self):
-        """AC5/AC8 — nothing usable supplied keeps the fail-closed fallback."""
-        for blank in ("", "   "):
-            with self.subTest(blank=repr(blank)):
-                captured = self._capture_preflight(
-                    **self._base_kwargs(worktree_path=blank)
-                )
-                self.assertIsNone(
-                    captured.get("worktree_path"),
-                    "a blank worktree must not resolve to the process cwd",
-                )
-
-    def test_org_repo_and_task_forwarding_are_not_regressed(self):
-        """AC6 — #735's org/repo forwarding and the task name still hold."""
-        captured = self._capture_preflight(**self._base_kwargs())
-        self.assertEqual(captured.get("org"), ORG)
-        self.assertEqual(captured.get("repo"), REPO)
-        self.assertEqual(captured.get("task"), "publish_unpublished_branch")
-
-
-class _ProductionTopologyBase(unittest.TestCase):
-    """Control checkout on master + a distinct registered issue worktree.
-
-    This is the shape the production daemon runs in and the shape the #812
-    fixture never built. ``PROJECT_ROOT`` is the control checkout; the issue
-    worktree is a real registered worktree at a different path; production
-    guards are forced on so the #618 branches-only guard genuinely evaluates.
-    """
-
-    def setUp(self):
-        self.lock_dir = tempfile.TemporaryDirectory()
-        self.addCleanup(self.lock_dir.cleanup)
-        self.origin = tempfile.mkdtemp(prefix="issue815-origin-")
-        self.control = tempfile.mkdtemp(prefix="issue815-control-")
-        for path in (self.origin, self.control):
-            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()
-
-    def _git(self, *args, cwd=None):
-        return subprocess.run(
-            ["git", "-C", cwd or self.control, *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.control, "seed.txt"), "w") as fh:
-            fh.write("seed\n")
-        # The real repository gitignores branches/, so a registered worktree
-        # living there does not dirty the stable control checkout. Mirror that,
-        # or the #615 dirty-runtime block fires on the worktree we just created.
-        with open(os.path.join(self.control, ".gitignore"), "w") as fh:
-            fh.write("branches/\n")
-        self._git("add", "seed.txt", ".gitignore")
-        self._git("commit", "-q", "-m", "seed")
-        self.base_sha = self._git("rev-parse", "HEAD").stdout.strip()
-        self._git("push", "-q", GIT_REMOTE, "master")
-
-        # The control checkout STAYS on master. This is the whole point: the
-        # daemon's process root is the stable control checkout, never the
-        # worktree the publication targets.
-        self.worktree = os.path.realpath(
-            os.path.join(self.control, "branches", WORKTREE_DIRNAME)
-        )
-        self._git("worktree", "add", "-q", "-b", BRANCH, self.worktree, "master")
-
-        with open(os.path.join(self.worktree, "work.txt"), "w") as fh:
-            fh.write("unpublished implementation\n")
-        self._git("add", "work.txt", cwd=self.worktree)
-        self._git("commit", "-q", "-m", "unpublished implementation", cwd=self.worktree)
-        self.head_sha = self._git("rev-parse", "HEAD", cwd=self.worktree).stdout.strip()
-
-        self.control_branch = self._git(
-            "rev-parse", "--abbrev-ref", "HEAD"
-        ).stdout.strip()
-
-    # ── durable lock naming the caller and the issue worktree ────────────
-    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, *, bind_session=True, **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),
-            "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)
-        # Bind the session pointer so the #683 issue-scope guard resolves an
-        # owning issue for this author session. In real production the publish
-        # task does not require a session lock — require_author_lock is keyed on
-        # the test-only production_guards_forced() flag, which this suite must
-        # set to make preflight run at all — so this pointer is fixture
-        # scaffolding to clear a guard production would not apply here, never a
-        # softening of the worktree-forwarding behaviour under test. The
-        # preflight-negative cases below leave it unbound precisely so the #618
-        # guard is reached with no session fallback to rescue a bad worktree.
-        if bind_session:
-            pointer = {
-                "pid": os.getpid(),
-                "lock_file_path": path,
-                "issue_number": ISSUE,
-                "branch_name": data["branch_name"],
-                "remote": "prgs",
-                "org": ORG,
-                "repo": REPO,
-            }
-            issue_lock_store.save_lock_file(
-                issue_lock_store.session_pointer_path(self.lock_dir.name), pointer
-            )
-        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
-        # The defect only exists where preflight actually runs. Under pytest the
-        # production root/branches/scope guards are skipped unless forced on, so
-        # force them: this test exists to exercise the #618 guard, not to bypass
-        # it. Parity is pinned to the server's own startup head so repointing
-        # PROJECT_ROOT does not read as a stale daemon.
-        env["GITEA_TEST_FORCE_PRODUCTION_GUARDS"] = "1"
-        # Production is a promoted stable-control runtime. The pytest process
-        # itself runs from a branches/ worktree, which the #615 runtime-mode
-        # gate correctly classifies as dev-test; declaring the sanctioned mode
-        # models the production daemon rather than defeating the gate. Without
-        # this, forcing production guards on would trip the *runtime-mode* block
-        # for a reason unrelated to the #815 worktree-forwarding defect.
-        env["GITEA_MCP_RUNTIME_MODE"] = "stable-control"
-        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
-
-    def run_publish(self, *, open_prs=None, expected_head=None, **kwargs):
-        """Drive the public tool with PROJECT_ROOT pinned to the CONTROL checkout."""
-        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(
-            # NOTE: the control checkout — deliberately NOT self.worktree.
-            mcp_server, "PROJECT_ROOT", self.control
-        ), 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 TestForwardingClearsThe618Guard(_ProductionTopologyBase):
-    """AC2 — the faithful production reproduction, and the sharpest fix proof.
-
-    The production recovery worker had **no** session issue lock — acquiring one
-    was the very thing the deadlock prevented — so preflight had nothing but the
-    explicit ``worktree_path`` argument to resolve the workspace from. This class
-    reproduces exactly that: no session pointer is bound, so there is no
-    author-lock fallback to rescue a dropped argument.
-
-    With the argument forwarded (fixed source) the #618 branches-only guard
-    accepts the registered issue worktree and the call advances to the next
-    guard. With the argument dropped (the buggy source this issue reports)
-    preflight falls back to ``PROJECT_ROOT`` — the stable control checkout — and
-    the #618 guard traps the call there. The two outcomes are told apart by the
-    guard that fired, on its own error text.
-
-    This test therefore *fails* against the unpatched source (the call is trapped
-    at #618 instead of clearing it), which is what makes it a regression rather
-    than a smoke test.
-    """
-
-    _CONTROL_CHECKOUT_MARKERS = ("stable control checkout", "#618")
-
-    def test_explicit_worktree_clears_618_without_a_session_lock(self):
-        # No write_lock(): the session is deliberately unbound, as in production.
-        with self.assertRaises(RuntimeError) as ctx:
-            self.run_publish()
-        message = str(ctx.exception)
-        # The workspace guard is satisfied — the failure is the *later* scope
-        # guard (no owning issue), never the control-checkout refusal. If the
-        # argument were dropped, this call would be trapped at #618 instead.
-        for marker in self._CONTROL_CHECKOUT_MARKERS:
-            self.assertNotIn(
-                marker, message,
-                f"the explicit worktree must clear #618; got a control-checkout "
-                f"refusal instead: {message}",
-            )
-        self.assertIn(
-            "owning issue", message,
-            f"expected the downstream scope guard to fire, got: {message}",
-        )
-        self.assertIsNone(self.remote_head())
-
-    def test_dropped_argument_would_be_trapped_at_618(self):
-        # Simulate the buggy call shape directly: no session lock, and preflight
-        # given no worktree, exactly as the unpatched source left it. This pins
-        # the control-checkout refusal that the fix eliminates, so the pair of
-        # tests brackets the defect from both sides regardless of which source
-        # version is loaded.
-        env = self._tool_env()
-        with patch(
-            "mcp_server._list_open_pulls", return_value=[]
-        ), 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.control
-        ), patch.dict(os.environ, env, clear=True):
-            os.environ["GITEA_ISSUE_LOCK_DIR"] = self.lock_dir.name
-            with self.assertRaises(RuntimeError) as ctx:
-                # Drive verify_preflight_purity the way the buggy body did:
-                # no worktree_path forwarded at all.
-                mcp_server.verify_preflight_purity(
-                    "prgs",
-                    task="publish_unpublished_branch",
-                    org=ORG,
-                    repo=REPO,
-                )
-        message = str(ctx.exception)
-        self.assertTrue(
-            any(m in message for m in self._CONTROL_CHECKOUT_MARKERS),
-            f"a dropped worktree must trap at the control checkout: {message}",
-        )
-        self.assertIsNone(self.remote_head())
-
-
-class TestProductionTopologyPublishes(_ProductionTopologyBase):
-    """AC2/AC4/AC7 — the explicit registered worktree is what preflight validates."""
-
-    def test_fixture_is_genuinely_the_production_topology(self):
-        """Guard the guard: if this drifts, the regression stops meaning anything."""
-        self.assertNotEqual(
-            os.path.realpath(self.control), self.worktree,
-            "the issue worktree must not be PROJECT_ROOT",
-        )
-        self.assertEqual(
-            self.control_branch, "master",
-            "the control checkout must sit on a stable branch",
-        )
-        self.assertTrue(
-            os.path.realpath(self.worktree).startswith(
-                os.path.realpath(os.path.join(self.control, "branches")) + os.sep
-            ),
-            "the issue worktree must live under branches/",
-        )
-        listed = subprocess.run(
-            ["git", "-C", self.control, "worktree", "list"],
-            capture_output=True, text=True, check=True,
-        ).stdout
-        self.assertIn(
-            self.worktree, listed, "the issue worktree must be genuinely registered"
-        )
-
-    def test_publishes_from_a_control_rooted_daemon(self):
-        """The exact production failure: this refused with #618 before the fix."""
-        self.write_lock()
-        self.assertIsNone(self.remote_head(), "fixture must start unpublished")
-        res = self.run_publish()
-        self.assertTrue(res.get("success"), res)
-        self.assertTrue(res.get("performed"), res)
-        self.assertEqual(self.remote_head(), self.head_sha)
-
-    def test_dry_run_uses_the_explicit_worktree(self):
-        """AC4 — dry-run reaches the same decision without publishing."""
-        self.write_lock()
-        res = self.run_publish(dry_run=True)
-        self.assertTrue(res.get("success"), res)
-        self.assertFalse(res.get("performed"), res)
-        self.assertTrue(res.get("would_publish"), res)
-        self.assertIsNone(self.remote_head(), "dry-run must not publish")
-
-    def test_dry_run_and_apply_agree_on_the_same_worktree(self):
-        """AC4 — both paths resolve the same workspace, so both succeed."""
-        self.write_lock()
-        dry = self.run_publish(dry_run=True)
-        self.assertTrue(dry.get("would_publish"), dry)
-        applied = self.run_publish()
-        self.assertTrue(applied.get("performed"), applied)
-        self.assertEqual(self.remote_head(), self.head_sha)
-
-    def test_read_after_write_verification_still_runs(self):
-        """AC6 — PR #814's post-publication verification is unchanged."""
-        self.write_lock()
-        res = self.run_publish()
-        self.assertTrue(res.get("verified"), res)
-        self.assertEqual(res.get("remote_head_sha"), self.head_sha)
-
-
-class TestProductionTopologyFailsClosed(_ProductionTopologyBase):
-    """AC3/AC5/AC8 — the fix does not weaken any refusal.
-
-    A refusal reaches the caller by one of two mechanisms, and this class holds
-    them apart deliberately. A bad *workspace* is caught by the #618 preflight
-    guard, which raises before the assessor is built. A bad *content/ownership*
-    fact passes preflight (the worktree itself is fine) and is then refused by
-    the publication assessor, which returns ``success: False``. Both are
-    fail-closed; asserting the wrong mechanism would hide a regression.
-    """
-
-    # ── #618 preflight refusals: no session lock, so nothing rescues a bad
-    #    workspace and the guard fires exactly as it does in production ──────
-    def _assert_preflight_raises(self, **kwargs):
-        with self.assertRaises(RuntimeError) as ctx:
-            self.run_publish(**kwargs)
-        self.assertIsNone(
-            self.remote_head(), "a blocked publication must not reach the remote"
-        )
-        return str(ctx.exception)
-
-    def test_blank_worktree_path_fails_closed_via_618(self):
-        """AC5 — a blank path forwards None, so preflight sees the control root."""
-        for blank in ("", "   "):
-            with self.subTest(blank=repr(blank)):
-                message = self._assert_preflight_raises(worktree_path=blank)
-                self.assertIn("618", message)
-
-    def test_control_checkout_as_worktree_fails_closed_via_618(self):
-        """AC5 — naming the stable control checkout explicitly is still refused."""
-        message = self._assert_preflight_raises(worktree_path=self.control)
-        self.assertIn("618", message)
-
-    def test_unregistered_directory_fails_closed(self):
-        """AC3 — a plain directory under branches/ is not a registered worktree."""
-        bogus = os.path.join(self.control, "branches", "not-a-worktree")
-        os.makedirs(bogus, exist_ok=True)
-        self._assert_preflight_raises(worktree_path=bogus)
-
-    def test_missing_worktree_path_fails_closed(self):
-        """AC3 — a path that does not exist is refused, not silently replaced."""
-        missing = os.path.join(self.control, "branches", "absent-worktree")
-        self._assert_preflight_raises(worktree_path=missing)
-
-    # ── assessor refusals: preflight passes on a valid worktree, then the
-    #    publication assessor refuses on content/ownership evidence ──────────
-    def _assert_assessor_refuses(self, **kwargs):
-        res = self.run_publish(**kwargs)
-        self.assertFalse(res.get("success"), res)
-        self.assertFalse(res.get("performed"), res)
-        self.assertIsNone(self.remote_head())
-        return res
-
-    def test_changed_local_head_still_refuses(self):
-        """AC6 — the declared expected_head remains authoritative."""
-        self.write_lock()
-        self._assert_assessor_refuses(expected_head="b" * 40)
-
-    def test_foreign_claimant_still_refuses(self):
-        """AC6 — ownership still comes from the durable lock record."""
-        self.write_lock(claimant={"username": "someone-else", "profile": PROFILE})
-        self._assert_assessor_refuses()
-
-    def test_dirty_worktree_still_refuses(self):
-        """AC6 — cleanliness enforcement survives the forwarding change."""
-        self.write_lock()
-        with open(os.path.join(self.worktree, "work.txt"), "a") as fh:
-            fh.write("uncommitted drift\n")
-        self._assert_assessor_refuses()
-
-    def test_competing_open_pr_still_refuses(self):
-        """AC6 — a rival claim on another branch still blocks."""
-        self.write_lock()
-        self._assert_assessor_refuses(
-            open_prs=[{"number": 4242, "head": {"ref": f"fix/issue-{ISSUE}-rival"}}]
-        )
-
-    def test_issue_lock_record_is_not_mutated_by_a_refusal(self):
-        """AC6 — record separation (#812 AC23) is unaffected by this change."""
-        path = self.write_lock()
-        with open(path, "rb") as fh:
-            before = fh.read()
-        self._assert_assessor_refuses(expected_head="c" * 40)
-        with open(path, "rb") as fh:
-            self.assertEqual(before, fh.read())
-
-
-class TestProtectedFixtureNotReferenced(unittest.TestCase):
-    """AC9 — this regression never names the protected #635 fixture.
-
-    The forbidden tokens are reconstructed from fragments so this assertion
-    file does not itself contain them and produce a false positive.
-    """
-
-    def test_no_reference_to_the_protected_worktree(self):
-        forbidden = [
-            "issue-635-" + "project-registry-api",
-            "b2f6e9a6dc40e9651ef8" + "76f322dd0a68bddebfd8",
-        ]
-        here = os.path.abspath(__file__)
-        with open(here, "r", encoding="utf-8") as fh:
-            text = fh.read()
-        for token in forbidden:
-            self.assertNotIn(
-                token, text,
-                f"the protected #635 fixture must not be referenced: {token}",
-            )
-
-
-if __name__ == "__main__":
-    unittest.main()

tests/test_webui_inventory.py

@@ -1,468 +0,0 @@
-"""Tests for the unified web-console inventory API (#636).
-
-Covers the four cases the issue names — empty, populated, partial failure, and
-the no-false-unowned invariant — plus redaction, collision detection, the
-resource-split routes, and read-only guarantees against a real control-plane
-database and real durable lock files.
-"""
-import json
-import os
-import sys
-import tempfile
-import unittest
-from datetime import datetime, timedelta, timezone
-from pathlib import Path
-
-sys.path.insert(0, str(Path(__file__).resolve().parent.parent))
-
-from starlette.testclient import TestClient
-
-import control_plane_db
-from webui.app import create_app
-from webui import inventory
-
-
-def _iso(dt: datetime) -> str:
-    return dt.astimezone(timezone.utc).isoformat()
-
-
-def _write_lock(lock_dir: str, name: str, payload: dict) -> str:
-    path = os.path.join(lock_dir, name)
-    with open(path, "w", encoding="utf-8") as handle:
-        json.dump(payload, handle)
-    return path
-
-
-def _live_lock_payload(
-    *,
-    issue_number: int,
-    branch: str,
-    worktree_path: str,
-    pid: int,
-    username: str = "jcwalker3",
-    profile: str = "prgs-author",
-) -> dict:
-    now = datetime.now(timezone.utc)
-    future = now + timedelta(hours=2)
-    return {
-        "branch_name": branch,
-        "issue_number": issue_number,
-        "org": "Scaled-Tech-Consulting",
-        "repo": "Gitea-Tools",
-        "remote": "prgs",
-        "pid": pid,
-        "session_pid": pid,
-        "lock_generation": 1,
-        "worktree_path": worktree_path,
-        "claimant": {"username": username, "profile": profile},
-        "work_lease": {
-            "branch": branch,
-            "issue_number": issue_number,
-            "operation_type": "author_issue_work",
-            "created_at": _iso(now),
-            "expires_at": _iso(future),
-            "last_heartbeat_at": _iso(now),
-            "claimant": {"username": username, "profile": profile},
-        },
-    }
-
-
-class _FixtureMixin(unittest.TestCase):
-    def setUp(self) -> None:
-        self._tmp = tempfile.TemporaryDirectory()
-        self.tmp = self._tmp.name
-        self.lock_dir = os.path.join(self.tmp, "locks")
-        os.makedirs(self.lock_dir, mode=0o700)
-        self.db_path = os.path.join(self.tmp, "control_plane.db")
-        self.addCleanup(self._tmp.cleanup)
-
-    def _seed_db(self) -> control_plane_db.ControlPlaneDB:
-        db = control_plane_db.ControlPlaneDB(self.db_path)
-        db.upsert_session(
-            session_id="prgs-author-1",
-            role="author",
-            profile="prgs-author",
-            namespace="gitea-author",
-            pid=os.getpid(),
-        )
-        db.upsert_work_item(
-            remote="prgs",
-            org="Scaled-Tech-Consulting",
-            repo="Gitea-Tools",
-            kind="issue",
-            number=636,
-        )
-        db.assign_and_lease(
-            session_id="prgs-author-1",
-            role="author",
-            remote="prgs",
-            org="Scaled-Tech-Consulting",
-            repo="Gitea-Tools",
-            kind="issue",
-            number=636,
-        )
-        return db
-
-
-class TestRedaction(unittest.TestCase):
-    def test_redact_path_collapses_home(self):
-        home = os.path.expanduser("~")
-        self.assertEqual(
-            inventory.redact_path(f"{home}/Development/Gitea-Tools"),
-            "~/Development/Gitea-Tools",
-        )
-
-    def test_redact_url_strips_userinfo_and_query(self):
-        self.assertEqual(
-            inventory.redact_url("https://user:[email protected]/api?token=abc"),
-            "https://gitea.prgs.cc/api",
-        )
-
-    def test_scrub_drops_credential_keys(self):
-        scrubbed = inventory.scrub(
-            {"token": "abc123", "api_key": "k", "profile": "prgs-author"}
-        )
-        self.assertEqual(scrubbed["token"], "[redacted]")
-        self.assertEqual(scrubbed["api_key"], "[redacted]")
-        self.assertEqual(scrubbed["profile"], "prgs-author")
-
-    def test_scrub_is_recursive_and_never_raises(self):
-        class Weird:
-            def __repr__(self) -> str:
-                return "weird-obj"
-
-        out = inventory.scrub({"nested": [{"password": "p", "obj": Weird()}]})
-        self.assertEqual(out["nested"][0]["password"], "[redacted]")
-        self.assertEqual(out["nested"][0]["obj"], "weird-obj")
-
-
-class TestEmptyInventory(_FixtureMixin):
-    def test_empty_db_and_locks_degrade_without_raising(self):
-        # No DB file, no locks: sessions/leases unavailable, locks ok+empty.
-        snap = inventory.load_inventory_snapshot(
-            db_path=self.db_path,
-            lock_dir=self.lock_dir,
-            load_hygiene=lambda: _StubHygiene(entries=()),
-        )
-        sessions = snap.section("sessions")
-        leases = snap.section("leases")
-        locks = snap.section("locks")
-        self.assertEqual(sessions.status, inventory.STATUS_UNAVAILABLE)
-        self.assertEqual(leases.status, inventory.STATUS_UNAVAILABLE)
-        self.assertEqual(locks.status, inventory.STATUS_OK)
-        self.assertEqual(len(locks.items), 0)
-        # Ownership authority is incomplete because the DB is missing.
-        self.assertFalse(snap.ownership_authority_complete)
-        self.assertEqual(snap.collisions, ())
-
-    def test_empty_db_present_but_unpopulated(self):
-        control_plane_db.ControlPlaneDB(self.db_path)  # creates schema, no rows
-        snap = inventory.load_inventory_snapshot(
-            db_path=self.db_path,
-            lock_dir=self.lock_dir,
-            load_hygiene=lambda: _StubHygiene(entries=()),
-        )
-        self.assertEqual(snap.section("sessions").status, inventory.STATUS_OK)
-        self.assertEqual(len(snap.section("sessions").items), 0)
-        self.assertEqual(snap.section("leases").status, inventory.STATUS_OK)
-        self.assertTrue(snap.ownership_authority_complete)
-
-
-class TestPopulatedInventory(_FixtureMixin):
-    def test_sections_populated_and_correlated(self):
-        self._seed_db()
-        wt = f"{self.tmp}/branches/issue-636-inventory-api"
-        _write_lock(
-            self.lock_dir,
-            "prgs-Scaled-Tech-Consulting-Gitea-Tools-636.json",
-            _live_lock_payload(
-                issue_number=636,
-                branch="feat/issue-636-inventory-api",
-                worktree_path=wt,
-                pid=os.getpid(),
-            ),
-        )
-        hygiene = _StubHygiene(
-            entries=(
-                _StubEntry(
-                    rel_path="branches/issue-636-inventory-api",
-                    branch="feat/issue-636-inventory-api",
-                    classification="active-issue",
-                ),
-            )
-        )
-        snap = inventory.load_inventory_snapshot(
-            db_path=self.db_path,
-            lock_dir=self.lock_dir,
-            load_hygiene=lambda: hygiene,
-        )
-        self.assertTrue(snap.ownership_authority_complete)
-        self.assertEqual(len(snap.section("sessions").items), 1)
-        self.assertEqual(len(snap.section("leases").items), 1)
-        self.assertEqual(len(snap.section("locks").items), 1)
-        self.assertEqual(len(snap.section("worktrees").items), 1)
-
-        # The lease, lock, and worktree for #636 correlate onto one row.
-        row = next(r for r in snap.correlations if r["issue_number"] == 636)
-        self.assertEqual(row["branch"], "feat/issue-636-inventory-api")
-        self.assertTrue(row["lock_live"])
-        self.assertEqual(row["worktree_classification"], "active-issue")
-        self.assertEqual(len(row["lease_ids"]), 1)
-        # No collision: live lock, live pid, matching worktree.
-        self.assertEqual(snap.collisions, ())
-
-    def test_serialized_payload_declares_field_authority(self):
-        self._seed_db()
-        snap = inventory.load_inventory_snapshot(
-            db_path=self.db_path,
-            lock_dir=self.lock_dir,
-            load_hygiene=lambda: _StubHygiene(entries=()),
-        )
-        payload = inventory.snapshot_to_dict(snap)
-        self.assertEqual(payload["api_version"], "v1")
-        self.assertEqual(payload["schema_version"], 1)
-        self.assertEqual(payload["field_authority"]["sessions"], "control_plane_db")
-        self.assertEqual(payload["field_authority"]["locks"], "filesystem")
-        self.assertIn("sessions", payload["sections"])
-
-
-class TestPartialFailure(_FixtureMixin):
-    def test_worktree_scan_failure_degrades_only_that_section(self):
-        self._seed_db()
-
-        def _boom():
-            raise RuntimeError("git worktree list exploded")
-
-        snap = inventory.load_inventory_snapshot(
-            db_path=self.db_path,
-            lock_dir=self.lock_dir,
-            load_hygiene=_boom,
-        )
-        self.assertEqual(
-            snap.section("worktrees").status, inventory.STATUS_UNAVAILABLE
-        )
-        self.assertIn("exploded", snap.section("worktrees").reason)
-        # DB-backed sections still healthy.
-        self.assertEqual(snap.section("sessions").status, inventory.STATUS_OK)
-        self.assertIn("worktrees", snap.degraded_sections)
-
-    def test_degraded_ownership_suppresses_unowned_claim(self):
-        # DB absent → sessions/leases unavailable → ownership incomplete even
-        # though a lock exists and could look "unclaimed" by the DB alone.
-        _write_lock(
-            self.lock_dir,
-            "prgs-Scaled-Tech-Consulting-Gitea-Tools-636.json",
-            _live_lock_payload(
-                issue_number=636,
-                branch="feat/issue-636-inventory-api",
-                worktree_path=f"{self.tmp}/wt",
-                pid=os.getpid(),
-            ),
-        )
-        snap = inventory.load_inventory_snapshot(
-            db_path=self.db_path,
-            lock_dir=self.lock_dir,
-            load_hygiene=lambda: _StubHygiene(entries=()),
-        )
-        self.assertFalse(snap.ownership_authority_complete)
-        payload = inventory.snapshot_to_dict(snap)
-        self.assertIn("may be treated as unowned", payload["ownership_note"])
-
-
-class TestCollisionDetection(_FixtureMixin):
-    def test_live_lock_dead_owner_flagged(self):
-        _write_lock(
-            self.lock_dir,
-            "prgs-Scaled-Tech-Consulting-Gitea-Tools-700.json",
-            _live_lock_payload(
-                issue_number=700,
-                branch="feat/issue-700-x",
-                worktree_path=f"{self.tmp}/wt700",
-                pid=999_999_999,  # not a running pid
-            ),
-        )
-        control_plane_db.ControlPlaneDB(self.db_path)  # empty but present
-        snap = inventory.load_inventory_snapshot(
-            db_path=self.db_path,
-            lock_dir=self.lock_dir,
-            load_hygiene=lambda: _StubHygiene(entries=()),
-        )
-        kinds = {c.kind for c in snap.collisions}
-        self.assertIn("live-lock-dead-owner", kinds)
-        # Also lock-without-worktree, since no worktree carries the branch.
-        self.assertIn("lock-without-worktree", kinds)
-
-    def test_duplicate_live_lock_on_same_branch(self):
-        for issue in (800, 801):
-            _write_lock(
-                self.lock_dir,
-                f"prgs-Scaled-Tech-Consulting-Gitea-Tools-{issue}.json",
-                _live_lock_payload(
-                    issue_number=issue,
-                    branch="feat/issue-800-shared",
-                    worktree_path=f"{self.tmp}/wt{issue}",
-                    pid=os.getpid(),
-                ),
-            )
-        control_plane_db.ControlPlaneDB(self.db_path)
-        snap = inventory.load_inventory_snapshot(
-            db_path=self.db_path,
-            lock_dir=self.lock_dir,
-            load_hygiene=lambda: _StubHygiene(entries=()),
-        )
-        self.assertIn(
-            "duplicate-live-lock", {c.kind for c in snap.collisions}
-        )
-
-    def test_no_collision_when_sections_degraded(self):
-        # locks ok but worktrees unavailable → lock-without-worktree must NOT
-        # be asserted (a missing scan is not a missing worktree).
-        _write_lock(
-            self.lock_dir,
-            "prgs-Scaled-Tech-Consulting-Gitea-Tools-636.json",
-            _live_lock_payload(
-                issue_number=636,
-                branch="feat/issue-636-inventory-api",
-                worktree_path=f"{self.tmp}/wt",
-                pid=os.getpid(),
-            ),
-        )
-        control_plane_db.ControlPlaneDB(self.db_path)
-
-        def _boom():
-            raise RuntimeError("scan down")
-
-        snap = inventory.load_inventory_snapshot(
-            db_path=self.db_path,
-            lock_dir=self.lock_dir,
-            load_hygiene=_boom,
-        )
-        self.assertNotIn(
-            "lock-without-worktree", {c.kind for c in snap.collisions}
-        )
-
-
-class TestSectionInclude(_FixtureMixin):
-    def test_include_restricts_scanned_sections(self):
-        self._seed_db()
-        snap = inventory.load_inventory_snapshot(
-            db_path=self.db_path,
-            lock_dir=self.lock_dir,
-            include=("locks",),
-        )
-        self.assertIsNotNone(snap.section("locks"))
-        self.assertIsNone(snap.section("sessions"))
-        self.assertIsNone(snap.section("worktrees"))
-
-
-class TestRoutes(_FixtureMixin):
-    def setUp(self) -> None:
-        super().setUp()
-        # Point the loaders at the fixture DB and lock dir via env, and stub
-        # the worktree scan so the route does not shell out to git.
-        self._prev_env = {
-            "GITEA_CONTROL_PLANE_DB": os.environ.get("GITEA_CONTROL_PLANE_DB"),
-            "GITEA_ISSUE_LOCK_DIR": os.environ.get("GITEA_ISSUE_LOCK_DIR"),
-            "WEBUI_TEST_OFFLINE": os.environ.get("WEBUI_TEST_OFFLINE"),
-        }
-        os.environ["GITEA_CONTROL_PLANE_DB"] = self.db_path
-        os.environ["GITEA_ISSUE_LOCK_DIR"] = self.lock_dir
-        os.environ["WEBUI_TEST_OFFLINE"] = "1"
-        self._seed_db()
-        self.client = TestClient(create_app())
-
-    def tearDown(self) -> None:
-        for key, value in self._prev_env.items():
-            if value is None:
-                os.environ.pop(key, None)
-            else:
-                os.environ[key] = value
-
-    def test_inventory_route_returns_versioned_payload(self):
-        resp = self.client.get("/api/v1/inventory")
-        self.assertEqual(resp.status_code, 200)
-        body = resp.json()
-        self.assertEqual(body["api_version"], "v1")
-        self.assertIn("sessions", body["sections"])
-        self.assertIn("field_authority", body)
-
-    def test_section_route_restricts_and_labels(self):
-        resp = self.client.get("/api/v1/inventory/locks")
-        self.assertEqual(resp.status_code, 200)
-        body = resp.json()
-        self.assertEqual(body["requested_section"], "locks")
-        self.assertIn("locks", body["sections"])
-        self.assertNotIn("sessions", body["sections"])
-
-    def test_unknown_section_is_404(self):
-        resp = self.client.get("/api/v1/inventory/bogus")
-        self.assertEqual(resp.status_code, 404)
-        self.assertEqual(resp.json()["error"], "unknown_section")
-
-    def test_inventory_route_rejects_post(self):
-        resp = self.client.post("/api/v1/inventory")
-        self.assertEqual(resp.status_code, 405)
-
-
-class TestReadOnly(_FixtureMixin):
-    def test_snapshot_does_not_create_db_file(self):
-        missing = os.path.join(self.tmp, "does-not-exist.db")
-        inventory.load_inventory_snapshot(
-            db_path=missing,
-            lock_dir=self.lock_dir,
-            load_hygiene=lambda: _StubHygiene(entries=()),
-        )
-        self.assertFalse(os.path.exists(missing))
-
-    def test_readonly_connection_refuses_write(self):
-        self._seed_db()
-        conn = inventory._open_readonly(self.db_path)
-        try:
-            with self.assertRaises(Exception):
-                conn.execute(
-                    "INSERT INTO sessions(session_id, role, started_at, "
-                    "last_heartbeat_at, status) VALUES ('x','author',"
-                    "'t','t','active')"
-                )
-                conn.commit()
-        finally:
-            conn.close()
-
-
-# ── lightweight stand-ins for the #432 hygiene snapshot ──────────────────────
-
-
-class _StubEntry:
-    def __init__(
-        self,
-        *,
-        rel_path: str,
-        branch: str | None = None,
-        classification: str = "stale-clean",
-        head_sha: str | None = "abc123",
-        dirty_tracked: int = 0,
-        dirty_untracked: bool = False,
-        detached: bool = False,
-        registered_worktree: bool = True,
-        notes: str = "",
-    ) -> None:
-        self.rel_path = rel_path
-        self.folder_name = rel_path.split("/", 1)[-1]
-        self.branch = branch
-        self.classification = classification
-        self.head_sha = head_sha
-        self.dirty_tracked = dirty_tracked
-        self.dirty_untracked = dirty_untracked
-        self.detached = detached
-        self.registered_worktree = registered_worktree
-        self.notes = notes
-
-
-class _StubHygiene:
-    def __init__(self, *, entries=(), scan_error=None) -> None:
-        self.entries = tuple(entries)
-        self.scan_error = scan_error
-
-
-if __name__ == "__main__":
-    unittest.main()

tests/test_webui_project_registry.py

@@ -1,4 +1,4 @@
-"""Tests for web UI project registry (#427) and its API evolution (#635)."""
+"""Tests for web UI project registry (#427)."""
 import json
 import sys
 import tempfile
@@ -11,246 +11,57 @@ from starlette.testclient import TestClient
 
 from webui.app import create_app
 from webui.project_registry import (
-    CURRENT_SCHEMA_VERSION,
-    REGISTRY_API_VERSION,
-    SUPPORTED_SCHEMA_VERSIONS,
-    RegistryError,
     default_registry_path,
     load_registry,
-    onboarding_summary,
     project_to_dict,
 )
-from webui.registry_safety import is_forbidden_key
-
-_REPO_ROOT = Path(__file__).resolve().parent.parent
-_API_DOC = _REPO_ROOT / "docs" / "webui-project-registry-api.md"
 
 
-def _valid_project(**overrides):
+class TestProjectRegistryLoader(unittest.TestCase):
-    project = {
-        "id": "example",
-        "repo_name": "Example",
-        "gitea_owner": "Org",
-        "remote_host": "https://gitea.example.invalid",
-        "default_branch": "main",
-        "local_checkout_path": ".",
-        "profiles": {"author": "a", "reviewer": "r", "reconciler": "c"},
-        "workflow_paths": {"skill": "skills/x.md"},
-    }
-    project.update(overrides)
-    return project
-
-
-def _write_registry(payload) -> Path:
-    with tempfile.NamedTemporaryFile("w", suffix=".json", delete=False) as handle:
-        json.dump(payload, handle)
-        return Path(handle.name)
-
-
-class RegistryFileCase(unittest.TestCase):
-    """Base class that cleans up temporary registry files."""
-
-    def setUp(self):
-        self._temp_paths: list[Path] = []
-
-    def tearDown(self):
-        for path in self._temp_paths:
-            path.unlink(missing_ok=True)
-
-    def write_registry(self, payload) -> Path:
-        path = _write_registry(payload)
-        self._temp_paths.append(path)
-        return path
-
-
-class TestProjectRegistryLoader(RegistryFileCase):
     def test_default_registry_loads_gitea_tools(self):
         registry = load_registry()
-        self.assertEqual(registry.version, CURRENT_SCHEMA_VERSION)
+        self.assertEqual(registry.version, 1)
-        self.assertEqual(registry.schema_version, CURRENT_SCHEMA_VERSION)
-        self.assertEqual(registry.api_version, REGISTRY_API_VERSION)
         self.assertEqual(len(registry.projects), 1)
         project = registry.projects[0]
         self.assertEqual(project.id, "gitea-tools")
         self.assertEqual(project.repo_name, "Gitea-Tools")
         self.assertEqual(project.gitea_owner, "Scaled-Tech-Consulting")
-        self.assertEqual(project.repo_full_name, "Scaled-Tech-Consulting/Gitea-Tools")
         self.assertEqual(project.remote_host, "https://gitea.prgs.cc")
-        self.assertEqual(project.remote_name, "prgs")
-        self.assertEqual(project.status, "active")
         self.assertEqual(project.profiles["author"], "prgs-author")
         self.assertEqual(project.profiles["reviewer"], "prgs-reviewer")
         self.assertEqual(project.profiles["reconciler"], "prgs-reconciler")
         self.assertIn("skill", project.workflow_paths)
         self.assertGreaterEqual(len(project.onboarding_checklist), 4)
 
-    def test_default_registry_onboarding_summary_is_complete(self):
-        summary = onboarding_summary(load_registry().projects[0])
-        self.assertEqual(summary.total, summary.complete)
-        self.assertEqual(summary.required_outstanding, 0)
-        self.assertTrue(summary.onboarding_complete)
-
-    def test_version_1_registry_still_loads_with_defaults(self):
-        path = self.write_registry({
-            "version": 1,
-            "projects": [
-                _valid_project(
-                    onboarding_checklist=[
-                        {"id": "step", "title": "Step", "description": "Do it"}
-                    ]
-                )
-            ],
-        })
-        registry = load_registry(path)
-        self.assertEqual(registry.schema_version, 1)
-        self.assertIn(1, SUPPORTED_SCHEMA_VERSIONS)
-        project = registry.projects[0]
-        self.assertEqual(project.status, "active")
-        self.assertIsNone(project.remote_name)
-        self.assertIsNone(project.last_seen_health)
-        step = project.onboarding_checklist[0]
-        self.assertEqual(step.state, "pending")
-        self.assertTrue(step.required)
-        self.assertFalse(onboarding_summary(project).onboarding_complete)
-
-    def test_onboarding_summary_counts_states(self):
-        path = self.write_registry({
-            "version": 2,
-            "projects": [
-                _valid_project(
-                    onboarding_checklist=[
-                        {"id": "a", "title": "A", "description": "d", "state": "complete"},
-                        {"id": "b", "title": "B", "description": "d", "state": "blocked"},
-                        {
-                            "id": "c",
-                            "title": "C",
-                            "description": "d",
-                            "state": "pending",
-                            "required": False,
-                        },
-                        {
-                            "id": "d",
-                            "title": "D",
-                            "description": "d",
-                            "state": "not_applicable",
-                        },
-                    ]
-                )
-            ],
-        })
-        summary = onboarding_summary(load_registry(path).projects[0])
-        self.assertEqual(summary.total, 4)
-        self.assertEqual(summary.complete, 1)
-        self.assertEqual(summary.blocked, 1)
-        self.assertEqual(summary.pending, 1)
-        self.assertEqual(summary.not_applicable, 1)
-        # Only the blocked step is both required and outstanding.
-        self.assertEqual(summary.required_outstanding, 1)
-        self.assertFalse(summary.onboarding_complete)
-
-    def test_last_seen_health_is_parsed_when_present(self):
-        path = self.write_registry({
-            "version": 2,
-            "projects": [
-                _valid_project(
-                    last_seen_health={
-                        "status": "degraded",
-                        "checked_at": "2026-01-01T00:00:00Z",
-                        "detail": "daemon restart pending",
-                    }
-                )
-            ],
-        })
-        health = load_registry(path).projects[0].last_seen_health
-        self.assertIsNotNone(health)
-        self.assertEqual(health.status, "degraded")
-        self.assertEqual(health.checked_at, "2026-01-01T00:00:00Z")
-
     def test_registry_rejects_credential_keys(self):
-        path = self.write_registry({
+        payload = {
             "version": 1,
-            "projects": [_valid_project(id="bad", api_token="redacted-placeholder")],
-        })
-        with self.assertRaises(RegistryError) as ctx:
-            load_registry(path)
-        self.assertIn("credential", ctx.exception.remediation.lower())
-        self.assertEqual(ctx.exception.field_path, "projects[0].api_token")
-
-    def test_unsupported_version_fails_closed_with_remediation(self):
-        path = self.write_registry({"version": 99, "projects": [_valid_project()]})
-        with self.assertRaises(RegistryError) as ctx:
-            load_registry(path)
-        self.assertIn("unsupported registry version", ctx.exception.message)
-        self.assertIn(str(CURRENT_SCHEMA_VERSION), ctx.exception.remediation)
-        self.assertEqual(ctx.exception.field_path, "version")
-
-    def test_missing_required_field_fails_closed(self):
-        broken = _valid_project()
-        del broken["default_branch"]
-        path = self.write_registry({"version": 2, "projects": [broken]})
-        with self.assertRaises(RegistryError) as ctx:
-            load_registry(path)
-        self.assertIn("default_branch", ctx.exception.message)
-        self.assertEqual(ctx.exception.field_path, "projects[0]")
-
-    def test_unknown_status_fails_closed(self):
-        path = self.write_registry({
-            "version": 2,
-            "projects": [_valid_project(status="mystery")],
-        })
-        with self.assertRaises(RegistryError) as ctx:
-            load_registry(path)
-        self.assertEqual(ctx.exception.field_path, "projects[0].status")
-        self.assertIn("active", ctx.exception.remediation)
-
-    def test_unknown_onboarding_state_fails_closed(self):
-        path = self.write_registry({
-            "version": 2,
             "projects": [
-                _valid_project(
+                {
-                    onboarding_checklist=[
+                    "id": "bad",
-                        {"id": "a", "title": "A", "description": "d", "state": "almost"}
+                    "repo_name": "Bad",
-                    ]
+                    "gitea_owner": "Org",
-                )
+                    "remote_host": "https://gitea.example.invalid",
+                    "default_branch": "main",
+                    "local_checkout_path": ".",
+                    "profiles": {
+                        "author": "a",
+                        "reviewer": "r",
+                        "reconciler": "c",
+                    },
+                    "workflow_paths": {"skill": "skills/x.md"},
+                    "api_token": "secret",
+                }
             ],
-        })
+        }
-        with self.assertRaises(RegistryError) as ctx:
-            load_registry(path)
-        self.assertEqual(
-            ctx.exception.field_path,
-            "projects[0].onboarding_checklist[0].state",
-        )
-
-    def test_missing_profile_role_fails_closed(self):
-        path = self.write_registry({
-            "version": 2,
-            "projects": [_valid_project(profiles={"author": "a", "reviewer": "r"})],
-        })
-        with self.assertRaises(RegistryError) as ctx:
-            load_registry(path)
-        self.assertEqual(ctx.exception.field_path, "projects[0].profiles.reconciler")
-
-    def test_empty_projects_fails_closed(self):
-        path = self.write_registry({"version": 2, "projects": []})
-        with self.assertRaises(RegistryError) as ctx:
-            load_registry(path)
-        self.assertEqual(ctx.exception.field_path, "projects")
-
-    def test_invalid_json_fails_closed_with_location(self):
         with tempfile.NamedTemporaryFile("w", suffix=".json", delete=False) as handle:
-            handle.write("{not json")
+            json.dump(payload, handle)
             path = Path(handle.name)
-        self._temp_paths.append(path)
+        try:
-        with self.assertRaises(RegistryError) as ctx:
+            with self.assertRaises(ValueError):
-            load_registry(path)
+                load_registry(path)
-        self.assertIn("not valid JSON", ctx.exception.message)
+        finally:
-        self.assertIn("line", ctx.exception.remediation)
+            path.unlink(missing_ok=True)
-
-    def test_missing_file_fails_closed(self):
-        missing = Path(tempfile.gettempdir()) / "webui-registry-does-not-exist.json"
-        with self.assertRaises(RegistryError) as ctx:
-            load_registry(missing)
-        self.assertIn("could not be read", ctx.exception.message)
 
     def test_default_registry_path_points_at_packaged_data(self):
         path = default_registry_path()
@@ -270,147 +81,31 @@ class TestProjectRegistryRoutes(unittest.TestCase):
         self.assertIn("prgs-author", response.text)
         self.assertNotIn("child issue", response.text.lower())
 
-    def test_projects_page_shows_status_and_progress(self):
-        response = self.client.get("/projects")
-        self.assertIn("Status", response.text)
-        self.assertIn("Onboarding", response.text)
-        self.assertIn("4/4 complete", response.text)
-
     def test_project_detail_renders_checklist(self):
         response = self.client.get("/projects/gitea-tools")
         self.assertEqual(response.status_code, 200)
         self.assertIn("Onboarding checklist", response.text)
         self.assertIn("Configure execution profiles", response.text)
         self.assertIn("branches/", response.text)
-        self.assertIn("Complete", response.text)
-        self.assertIn("required outstanding 0", response.text)
 
     def test_project_detail_404(self):
         response = self.client.get("/projects/unknown-repo")
         self.assertEqual(response.status_code, 404)
 
-    def test_api_projects_alias_stays_compatible(self):
+    def test_api_projects_json(self):
         response = self.client.get("/api/projects")
         self.assertEqual(response.status_code, 200)
         data = response.json()
-        # #427 consumers keep these keys.
+        self.assertEqual(data["version"], 1)
-        self.assertEqual(data["version"], CURRENT_SCHEMA_VERSION)
-        self.assertIn("source_path", data)
         self.assertEqual(len(data["projects"]), 1)
         self.assertEqual(data["projects"][0]["id"], "gitea-tools")
         self.assertIn("onboarding_checklist", data["projects"][0])
 
-    def test_api_v1_projects_payload(self):
-        response = self.client.get("/api/v1/projects")
-        self.assertEqual(response.status_code, 200)
-        data = response.json()
-        self.assertEqual(data["api_version"], REGISTRY_API_VERSION)
-        self.assertEqual(data["schema_version"], CURRENT_SCHEMA_VERSION)
-        self.assertEqual(data["project_count"], 1)
-        self.assertEqual(data["source"]["kind"], "file")
-        self.assertTrue(data["source"]["inventory_complete"])
-        project = data["projects"][0]
-        self.assertEqual(project["status"], "active")
-        self.assertEqual(project["remote_name"], "prgs")
-        self.assertEqual(
-            project["repo_full_name"], "Scaled-Tech-Consulting/Gitea-Tools"
-        )
-        self.assertTrue(project["onboarding_summary"]["onboarding_complete"])
-        self.assertEqual(project["onboarding_checklist"][0]["state"], "complete")
-        self.assertIsNone(project["last_seen_health"])
-
-    def test_api_v1_project_detail(self):
-        response = self.client.get("/api/v1/projects/gitea-tools")
-        self.assertEqual(response.status_code, 200)
-        data = response.json()
-        self.assertEqual(data["api_version"], REGISTRY_API_VERSION)
-        self.assertEqual(data["project"]["id"], "gitea-tools")
-        self.assertEqual(data["source"]["kind"], "file")
-
-    def test_api_v1_project_detail_missing_fails_closed(self):
-        response = self.client.get("/api/v1/projects/not-registered")
-        self.assertEqual(response.status_code, 404)
-        data = response.json()
-        self.assertEqual(data["error"], "project_not_found")
-        self.assertEqual(data["project_id"], "not-registered")
-        self.assertIn("gitea-tools", data["known_project_ids"])
-        self.assertIn("remediation", data)
-
-    def test_api_v1_projects_is_read_only(self):
-        response = self.client.post("/api/v1/projects", json={})
-        self.assertEqual(response.status_code, 405)
-        self.assertEqual(response.json()["error"], "read-only-mvp")
-
     def test_project_to_dict_is_json_safe(self):
         registry = load_registry()
-        dto = project_to_dict(registry.projects[0])
+        encoded = json.dumps(project_to_dict(registry.projects[0]))
-        encoded = json.dumps(dto)
         self.assertIn("gitea-tools", encoded)
-        # Prose may mention tokens; no serialized *key* may look like a secret.
-        for key in dto:
-            with self.subTest(key=key):
-                self.assertFalse(is_forbidden_key(key))
-
-
-class TestInvalidRegistryFailsClosedOverHttp(RegistryFileCase):
-    def setUp(self):
-        super().setUp()
-        self.path = self.write_registry({"version": 42, "projects": []})
-        self.client = TestClient(create_app())
-
-    def _with_bad_registry(self, url: str):
-        import os
-        from unittest import mock
-
-        with mock.patch.dict(
-            os.environ, {"WEBUI_PROJECT_REGISTRY": str(self.path)}, clear=False
-        ):
-            return self.client.get(url)
-
-    def test_api_v1_reports_actionable_error(self):
-        response = self._with_bad_registry("/api/v1/projects")
-        self.assertEqual(response.status_code, 500)
-        data = response.json()
-        self.assertEqual(data["error"], "registry_invalid")
-        self.assertIn("unsupported registry version", data["detail"])
-        self.assertTrue(data["remediation"])
-        self.assertEqual(data["field_path"], "version")
-
-    def test_unversioned_alias_reports_actionable_error(self):
-        response = self._with_bad_registry("/api/projects")
-        self.assertEqual(response.status_code, 500)
-        self.assertEqual(response.json()["error"], "registry_invalid")
-
-    def test_html_page_reports_actionable_error(self):
-        response = self._with_bad_registry("/projects")
-        self.assertEqual(response.status_code, 500)
-        self.assertIn("Project registry unavailable", response.text)
-        self.assertIn("Remediation", response.text)
-
-
-class TestProjectRegistryApiDocs(unittest.TestCase):
-    def test_api_contract_is_documented(self):
-        self.assertTrue(_API_DOC.is_file(), f"missing {_API_DOC}")
-        text = _API_DOC.read_text(encoding="utf-8")
-        for token in (
-            "/api/v1/projects",
-            "/api/v1/projects/{project_id}",
-            "/api/projects",
-            "onboarding_summary",
-            "last_seen_health",
-            "registry_invalid",
-            "#635",
-        ):
-            with self.subTest(token=token):
-                self.assertIn(token, text)
-
-    def test_route_table_lists_versioned_routes(self):
-        local_dev = (_REPO_ROOT / "docs" / "webui-local-dev.md").read_text(
-            encoding="utf-8"
-        )
-        self.assertIn("/api/v1/projects", local_dev)
-        self.assertIn("webui-project-registry-api.md", local_dev)
 
 
 if __name__ == "__main__":
-    unittest.main()
+    unittest.main()

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,20 +11,9 @@ from starlette.routing import Route
 
 from webui.deployment_boundary import deployment_snapshot
 from webui.layout import render_page
-from webui.project_registry import (
+from webui.nav import NAV_GROUPS, STUB_PAGES
-    ProjectRegistry,
+from webui.project_registry import find_project, load_registry, registry_to_dict
-    RegistryError,
+from webui.project_views import render_project_detail, render_projects_list
-    find_project,
-    known_project_ids,
-    load_registry,
-    project_detail_to_dict,
-    registry_to_dict,
-)
-from webui.project_views import (
-    render_project_detail,
-    render_projects_list,
-    render_registry_error,
-)
 from webui.prompt_library import find_prompt, library_to_dict
 from webui.prompt_views import render_prompt_detail, render_prompts_page
 from final_report_validator import FINAL_REPORT_TASK_KINDS
@@ -41,11 +30,6 @@ from webui.worktree_scanner import load_hygiene_snapshot, snapshot_to_dict as wo
 from webui.worktree_views import render_worktrees_page
 from webui.runtime_health import load_runtime_snapshot, snapshot_to_dict as runtime_snapshot_to_dict
 from webui.runtime_views import render_runtime_page
-from webui.inventory import (
-    SECTION_NAMES as _INVENTORY_SECTIONS,
-    load_inventory_snapshot,
-    snapshot_to_dict as inventory_snapshot_to_dict,
-)
 
 _READ_ONLY_METHODS = frozenset({"GET", "HEAD", "OPTIONS"})
 _AUDIT_MUTATION_PATHS = frozenset({"/audit", "/api/audit"})
@@ -60,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>"
+        "<p>Read-only home for the MCP Control Plane Phase 1 operator console. "
-        "<ul>"
+        "Gitea, MCP capability gates, and canonical workflows remain the source "
-        "<li><strong>Queue</strong> — live PR and issue dashboard (#429)</li>"
+        "of truth; this console never mutates them.</p>"
-        "<li><strong>Projects</strong> — registry and onboarding (#427)</li>"
+        "<h2>Phase 1 surfaces</h2>"
-        "<li><strong>Prompts</strong> — canonical workflow prompt library (#428)</li>"
+        + _render_home_nav_groups()
-        "<li><strong>Runtime</strong> — MCP health and stale-runtime detection (#430)</li>"
+        + "<h2>MVP legacy pages</h2>"
-        "<li><strong>Audit</strong> — final-report paste and validator preview (#431)</li>"
+        "<ul>" + legacy + "</ul>"
-        "<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>"
     )
     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({
@@ -98,26 +120,14 @@ async def api_queue(_request: Request) -> JSONResponse:
     return JSONResponse(queue_snapshot_to_dict(load_queue_snapshot()))
 
 
-def _load_project_registry() -> tuple[ProjectRegistry | None, RegistryError | None]:
-    """Load the registry, converting validation failure into a fail-closed pair."""
-    try:
-        return load_registry(), None
-    except RegistryError as exc:
-        return None, exc
-
-
 async def projects(_request: Request) -> HTMLResponse:
-    registry, error = _load_project_registry()
+    registry = load_registry()
-    if error is not None:
-        return HTMLResponse(render_registry_error(error), status_code=500)
     return HTMLResponse(render_projects_list(registry))
 
 
 async def project_detail(request: Request) -> HTMLResponse:
     project_id = request.path_params["project_id"]
-    registry, error = _load_project_registry()
+    registry = load_registry()
-    if error is not None:
-        return HTMLResponse(render_registry_error(error), status_code=500)
     project = find_project(registry, project_id)
     if project is None:
         return HTMLResponse(
@@ -135,47 +145,10 @@ async def project_detail(request: Request) -> HTMLResponse:
 
 
 async def api_projects(_request: Request) -> JSONResponse:
-    """Unversioned MVP alias, retained through Phase 1 (#632 section 6)."""
+    registry = load_registry()
-    registry, error = _load_project_registry()
-    if error is not None:
-        return JSONResponse(error.to_dict(), status_code=500)
     return JSONResponse(registry_to_dict(registry))
 
 
-async def api_v1_projects(_request: Request) -> JSONResponse:
-    registry, error = _load_project_registry()
-    if error is not None:
-        return JSONResponse(error.to_dict(), status_code=500)
-    return JSONResponse(registry_to_dict(registry))
-
-
-async def api_v1_project_detail(request: Request) -> JSONResponse:
-    project_id = request.path_params["project_id"]
-    registry, error = _load_project_registry()
-    if error is not None:
-        return JSONResponse(error.to_dict(), status_code=500)
-    project = find_project(registry, project_id)
-    if project is None:
-        return JSONResponse(
-            {
-                "error": "project_not_found",
-                "project_id": project_id,
-                "known_project_ids": known_project_ids(registry),
-                "remediation": (
-                    "Request one of the known project ids, or add the project to the "
-                    "registry file named in 'source'."
-                ),
-                "source": {
-                    "kind": "file",
-                    "path": str(registry.source_path),
-                    "inventory_complete": True,
-                },
-            },
-            status_code=404,
-        )
-    return JSONResponse(project_detail_to_dict(registry, project))
-
-
 async def prompts(_request: Request) -> HTMLResponse:
     return HTMLResponse(render_prompts_page())
 
@@ -307,30 +280,6 @@ async def api_action_attempt(request: Request) -> JSONResponse:
     return JSONResponse(result, status_code=status)
 
 
-async def api_inventory(_request: Request) -> JSONResponse:
-    """Unified read-only session/lease/lock/worktree inventory (#636)."""
-    snapshot = load_inventory_snapshot()
-    return JSONResponse(inventory_snapshot_to_dict(snapshot))
-
-
-async def api_inventory_section(request: Request) -> JSONResponse:
-    """Resource-split view: one inventory section under the shared schema."""
-    section = request.path_params["section"]
-    if section not in _INVENTORY_SECTIONS:
-        return JSONResponse(
-            {
-                "error": "unknown_section",
-                "detail": f"no inventory section named {section!r}",
-                "available": sorted(_INVENTORY_SECTIONS),
-            },
-            status_code=404,
-        )
-    snapshot = load_inventory_snapshot(include=(section,))
-    payload = inventory_snapshot_to_dict(snapshot)
-    payload["requested_section"] = section
-    return JSONResponse(payload)
-
-
 async def method_not_allowed(request: Request, _exc: Exception) -> Response:
     path = request.url.path
     if path in _AUDIT_MUTATION_PATHS and request.method == "POST":
@@ -358,12 +307,6 @@ def create_app(*, bind_host: str | None = None) -> Starlette:
             Route("/projects", projects, methods=["GET"]),
             Route("/projects/{project_id}", project_detail, methods=["GET"]),
             Route("/api/projects", api_projects, methods=["GET"]),
-            Route("/api/v1/projects", api_v1_projects, methods=["GET"]),
-            Route(
-                "/api/v1/projects/{project_id}",
-                api_v1_project_detail,
-                methods=["GET"],
-            ),
             Route("/prompts", prompts, methods=["GET"]),
             Route("/prompts/{prompt_id}", prompt_detail, methods=["GET"]),
             Route("/api/prompts", api_prompts, methods=["GET"]),
@@ -387,12 +330,10 @@ def create_app(*, bind_host: str | None = None) -> Starlette:
                 methods=["POST"],
             ),
             Route("/api/leases", api_leases, methods=["GET"]),
-            Route("/api/v1/inventory", api_inventory, methods=["GET"]),
+            *[
-            Route(
+                Route(path, phase_stub, methods=["GET"])
-                "/api/v1/inventory/{section}",
+                for path in STUB_PAGES
-                api_inventory_section,
+            ],
-                methods=["GET"],
-            ),
         ],
         exception_handlers={405: method_not_allowed},
     )

webui/data/projects.registry.json

@@ -1,15 +1,13 @@
 {
-  "version": 2,
+  "version": 1,
   "projects": [
     {
       "id": "gitea-tools",
       "repo_name": "Gitea-Tools",
       "gitea_owner": "Scaled-Tech-Consulting",
-      "remote_name": "prgs",
       "remote_host": "https://gitea.prgs.cc",
       "default_branch": "master",
       "local_checkout_path": ".",
-      "status": "active",
       "profiles": {
         "author": "prgs-author",
         "reviewer": "prgs-reviewer",
@@ -28,32 +26,24 @@
         {
           "id": "profiles",
           "title": "Configure execution profiles",
-          "description": "Install author, reviewer, and reconciler MCP profiles (prgs-author, prgs-reviewer, prgs-reconciler) in separate namespaces. Tokens stay in keychain — never in this registry.",
+          "description": "Install author, reviewer, and reconciler MCP profiles (prgs-author, prgs-reviewer, prgs-reconciler) in separate namespaces. Tokens stay in keychain — never in this registry."
-          "state": "complete",
-          "required": true
         },
         {
           "id": "mcp_config",
           "title": "Wire MCP v2 contexts",
-          "description": "Copy and customize gitea-mcp.v2-contexts.example.json for your machine. Map this repo path under projects with default_owner Scaled-Tech-Consulting and default_repo Gitea-Tools.",
+          "description": "Copy and customize gitea-mcp.v2-contexts.example.json for your machine. Map this repo path under projects with default_owner Scaled-Tech-Consulting and default_repo Gitea-Tools."
-          "state": "complete",
-          "required": true
         },
         {
           "id": "wiki_gate",
           "title": "Wiki publication readiness",
-          "description": "For wiki-tracked work, satisfy the live Gitea Wiki proof gate (#224) before closing issues. See docs/wiki/Safety-and-Gates.md.",
+          "description": "For wiki-tracked work, satisfy the live Gitea Wiki proof gate (#224) before closing issues. See docs/wiki/Safety-and-Gates.md."
-          "state": "complete",
-          "required": true
         },
         {
           "id": "branches_layout",
           "title": "Isolate work under branches/",
-          "description": "All LLM task edits happen in worktrees under branches/. Main checkout stays clean; use skills/llm-project-workflow templates for start-issue and review flows.",
+          "description": "All LLM task edits happen in worktrees under branches/. Main checkout stays clean; use skills/llm-project-workflow templates for start-issue and review flows."
-          "state": "complete",
-          "required": true
         }
       ]
     }
   ]
-}
+}

webui/inventory.py

@@ -1,952 +0,0 @@
-"""Unified session/lease/lock/worktree inventory for the web console (#636).
-
-Leases (#433), worktrees (#432), and runtime (#430) each ship their own MVP
-view, each with its own shape and its own idea of what "owned" means. A
-traffic-control or recovery operator has to read all three and correlate them
-by hand, which is exactly the step that goes wrong under collision pressure.
-
-This module aggregates them into one versioned, read-only snapshot so the
-console, and any worker asking "what is safe to do next", read the same
-inventory from the same authority.
-
-Field authority is explicit and never blended. Every section declares where its
-rows came from:
-
-* ``control_plane_db`` — the #613 substrate: sessions, leases, assignments.
-  Authoritative for *exclusive ownership* (#600/#601).
-* ``filesystem`` — durable per-issue lock files (:mod:`issue_lock_store`) and
-  registered git worktrees. Authoritative for *what exists on this machine*.
-* ``gitea`` — remote issue/PR state, reached only through existing loaders.
-
-Safety invariants:
-
-* **Read-only.** The control-plane database is opened through a ``mode=ro``
-  URI. :class:`control_plane_db.ControlPlaneDB` creates directories and runs
-  migrations in its constructor, which an inventory read must never do, so this
-  module talks to sqlite directly rather than through that class.
-* **Fail-soft, never fail-silent.** A subsystem that cannot be read degrades to
-  a section carrying ``status`` and ``reason``. It never raises, and it never
-  produces an empty list that reads like "nothing is there".
-* **Never invent active ownership.** This is the invariant that matters most.
-  A degraded ownership source sets ``ownership_authority_complete`` false, and
-  while that flag is false no work item is reported unowned and no collision is
-  asserted. Absence of evidence is reported as absence of evidence.
-* **Redaction at the boundary.** Absolute paths are collapsed against the home
-  directory, URLs lose userinfo and query strings, and no credential-shaped
-  value is emitted. No session token exists in these sources and none is read.
-
-Phase 1 is read-only. Lease steal/release and worktree deletion are Phase 2+
-and deliberately have no representation here, not even a disabled one.
-"""
-
-from __future__ import annotations
-
-import os
-import re
-import sqlite3
-import time
-from dataclasses import dataclass, field
-from datetime import datetime, timezone
-from typing import Any, Callable
-from urllib.parse import urlparse
-
-import control_plane_db
-import issue_lock_store
-
-SCHEMA_VERSION = 1
-API_VERSION = "v1"
-
-#: Sections whose absence would make an ownership claim unprovable. If any of
-#: these is not ``ok``, the snapshot refuses to describe anything as unowned.
-OWNERSHIP_SECTIONS = ("sessions", "leases", "locks")
-
-SECTION_NAMES = ("sessions", "leases", "locks", "worktrees", "namespaces")
-
-STATUS_OK = "ok"
-STATUS_DEGRADED = "degraded"
-STATUS_UNAVAILABLE = "unavailable"
-
-AUTHORITY_CONTROL_PLANE_DB = "control_plane_db"
-AUTHORITY_FILESYSTEM = "filesystem"
-AUTHORITY_GITEA = "gitea"
-
-_CREDENTIAL_KEY_RE = re.compile(
-    r"(token|secret|password|passwd|api[_-]?key|authorization|bearer|credential)",
-    re.IGNORECASE,
-)
-_REDACTED = "[redacted]"
-
-
-@dataclass(frozen=True)
-class InventorySection:
-    """One subsystem's contribution, with its authority and health."""
-
-    name: str
-    authority: str
-    status: str
-    items: tuple[dict[str, Any], ...] = ()
-    reason: str | None = None
-    scan_ms: float | None = None
-
-    @property
-    def ok(self) -> bool:
-        return self.status == STATUS_OK
-
-    def to_dict(self) -> dict[str, Any]:
-        return {
-            "name": self.name,
-            "authority": self.authority,
-            "status": self.status,
-            "count": len(self.items),
-            "reason": self.reason,
-            "scan_ms": self.scan_ms,
-            "items": [dict(item) for item in self.items],
-        }
-
-
-@dataclass(frozen=True)
-class CollisionSignal:
-    """A detected conflict between two ownership records."""
-
-    kind: str
-    message: str
-    severity: str = "warning"
-    issue_number: int | None = None
-    branch: str | None = None
-    worktree_path: str | None = None
-    session_ids: tuple[str, ...] = ()
-
-    def to_dict(self) -> dict[str, Any]:
-        return {
-            "kind": self.kind,
-            "severity": self.severity,
-            "message": self.message,
-            "issue_number": self.issue_number,
-            "branch": self.branch,
-            "worktree_path": self.worktree_path,
-            "session_ids": list(self.session_ids),
-        }
-
-
-@dataclass(frozen=True)
-class InventorySnapshot:
-    """Versioned aggregate of every inventory section."""
-
-    generated_at: str
-    sections: tuple[InventorySection, ...]
-    collisions: tuple[CollisionSignal, ...] = ()
-    correlations: tuple[dict[str, Any], ...] = ()
-    schema_version: int = SCHEMA_VERSION
-    api_version: str = API_VERSION
-    scan_ms: float | None = None
-    _section_index: dict[str, InventorySection] = field(
-        default_factory=dict, repr=False, compare=False
-    )
-
-    def section(self, name: str) -> InventorySection | None:
-        return self._section_index.get(name)
-
-    @property
-    def degraded_sections(self) -> tuple[str, ...]:
-        return tuple(s.name for s in self.sections if not s.ok)
-
-    @property
-    def ownership_authority_complete(self) -> bool:
-        """True only when every ownership-bearing section read cleanly.
-
-        While this is false the snapshot must not describe any work item as
-        unowned: a lease the reader could not load is not an absent lease.
-        """
-        for name in OWNERSHIP_SECTIONS:
-            section = self._section_index.get(name)
-            if section is None or not section.ok:
-                return False
-        return True
-
-    @property
-    def status(self) -> str:
-        if all(s.ok for s in self.sections):
-            return STATUS_OK
-        return STATUS_DEGRADED
-
-
-# ── redaction ────────────────────────────────────────────────────────────────
-
-
-def redact_path(path: str | None) -> str | None:
-    """Collapse an absolute path against ``$HOME`` for browser display."""
-    if not path:
-        return path
-    text = str(path)
-    home = os.path.expanduser("~")
-    if home and home != "/" and text.startswith(home):
-        return "~" + text[len(home) :]
-    return text
-
-
-def redact_url(value: str | None) -> str | None:
-    """Strip userinfo and query string from a URL."""
-    if not value:
-        return value
-    text = str(value)
-    try:
-        parsed = urlparse(text)
-    except ValueError:
-        return _REDACTED
-    if not parsed.scheme or not parsed.netloc:
-        return text
-    netloc = parsed.hostname or ""
-    if parsed.port:
-        netloc = f"{netloc}:{parsed.port}"
-    rebuilt = f"{parsed.scheme}://{netloc}{parsed.path}"
-    return rebuilt.rstrip("/") or rebuilt
-
-
-def scrub(value: Any, *, key: str | None = None) -> Any:
-    """Recursively drop credential-shaped values and redact paths/URLs.
-
-    Never raises: an unexpected object degrades to its ``repr`` rather than
-    propagating out of a read-only view.
-    """
-    if key and _CREDENTIAL_KEY_RE.search(key):
-        return _REDACTED
-    if isinstance(value, dict):
-        return {str(k): scrub(v, key=str(k)) for k, v in value.items()}
-    if isinstance(value, (list, tuple)):
-        return [scrub(v, key=key) for v in value]
-    if isinstance(value, str):
-        if value.startswith(("http://", "https://")):
-            return redact_url(value)
-        if value.startswith("/") or value.startswith("~"):
-            return redact_path(value)
-        return value
-    if isinstance(value, (int, float, bool)) or value is None:
-        return value
-    return repr(value)
-
-
-# ── control-plane database (read-only) ───────────────────────────────────────
-
-
-def _open_readonly(db_path: str) -> sqlite3.Connection:
-    """Open the control-plane DB without creating or migrating anything."""
-    conn = sqlite3.connect(f"file:{db_path}?mode=ro", uri=True, timeout=5)
-    conn.row_factory = sqlite3.Row
-    return conn
-
-
-def _table_names(conn: sqlite3.Connection) -> set[str]:
-    rows = conn.execute(
-        "SELECT name FROM sqlite_master WHERE type = 'table'"
-    ).fetchall()
-    return {str(row[0]) for row in rows}
-
-
-def _load_cp_db_sections(
-    *,
-    db_path: str | None = None,
-    limit: int = 200,
-) -> tuple[InventorySection, InventorySection]:
-    """Return the ``sessions`` and ``leases`` sections from the #613 DB."""
-    path = (db_path or control_plane_db.default_db_path()).strip()
-
-    def _both_unavailable(reason: str) -> tuple[InventorySection, InventorySection]:
-        return (
-            InventorySection(
-                name="sessions",
-                authority=AUTHORITY_CONTROL_PLANE_DB,
-                status=STATUS_UNAVAILABLE,
-                reason=reason,
-            ),
-            InventorySection(
-                name="leases",
-                authority=AUTHORITY_CONTROL_PLANE_DB,
-                status=STATUS_UNAVAILABLE,
-                reason=reason,
-            ),
-        )
-
-    if not path:
-        return _both_unavailable("control-plane database path is not configured")
-    if not os.path.exists(path):
-        return _both_unavailable(
-            f"control-plane database not present at {redact_path(path)}; "
-            "no session or lease authority available"
-        )
-
-    started = time.perf_counter()
-    try:
-        conn = _open_readonly(path)
-    except sqlite3.Error as exc:
-        return _both_unavailable(f"control-plane database could not be opened: {exc}")
-
-    try:
-        tables = _table_names(conn)
-        if "sessions" not in tables or "leases" not in tables:
-            missing = sorted({"sessions", "leases"} - tables)
-            return _both_unavailable(
-                "control-plane database is missing required tables: "
-                + ", ".join(missing)
-            )
-
-        session_rows = [
-            dict(row)
-            for row in conn.execute(
-                "SELECT session_id, role, profile, namespace, pid, started_at,"
-                " last_heartbeat_at, status FROM sessions"
-                " ORDER BY last_heartbeat_at DESC LIMIT ?",
-                (max(1, int(limit)),),
-            ).fetchall()
-        ]
-
-        has_work_items = "work_items" in tables
-        if has_work_items:
-            lease_sql = (
-                "SELECT l.lease_id, l.session_id, l.role, l.phase, l.status,"
-                " l.expires_at, w.remote, w.org, w.repo, w.kind AS work_kind,"
-                " w.number AS work_number, w.state AS work_state,"
-                " s.pid AS session_pid, s.profile AS session_profile,"
-                " s.namespace AS session_namespace, s.status AS session_status"
-                " FROM leases l"
-                " JOIN work_items w ON w.work_item_id = l.work_item_id"
-                " LEFT JOIN sessions s ON s.session_id = l.session_id"
-                " ORDER BY l.expires_at DESC LIMIT ?"
-            )
-        else:
-            lease_sql = (
-                "SELECT l.lease_id, l.session_id, l.role, l.phase, l.status,"
-                " l.expires_at FROM leases l"
-                " ORDER BY l.expires_at DESC LIMIT ?"
-            )
-        lease_rows = [
-            dict(row)
-            for row in conn.execute(lease_sql, (max(1, int(limit)),)).fetchall()
-        ]
-    except sqlite3.Error as exc:
-        return _both_unavailable(f"control-plane database read failed: {exc}")
-    finally:
-        conn.close()
-
-    elapsed = (time.perf_counter() - started) * 1000.0
-    now = datetime.now(timezone.utc)
-
-    sessions = tuple(
-        scrub(
-            {
-                "session_id": row.get("session_id"),
-                "role": row.get("role"),
-                "profile": row.get("profile"),
-                "namespace": row.get("namespace"),
-                "pid": row.get("pid"),
-                "pid_alive": issue_lock_store.is_process_alive(row.get("pid")),
-                "started_at": row.get("started_at"),
-                "last_heartbeat_at": row.get("last_heartbeat_at"),
-                "status": row.get("status"),
-            }
-        )
-        for row in session_rows
-    )
-
-    leases = tuple(
-        scrub(
-            {
-                "lease_id": row.get("lease_id"),
-                "session_id": row.get("session_id"),
-                "role": row.get("role"),
-                "phase": row.get("phase"),
-                "status": row.get("status"),
-                "expires_at": row.get("expires_at"),
-                "expired": _is_expired(row.get("expires_at"), now=now),
-                "remote": row.get("remote"),
-                "org": row.get("org"),
-                "repo": row.get("repo"),
-                "work_kind": row.get("work_kind"),
-                "work_number": row.get("work_number"),
-                "work_state": row.get("work_state"),
-                "session_pid": row.get("session_pid"),
-                "session_profile": row.get("session_profile"),
-                "session_namespace": row.get("session_namespace"),
-                "session_status": row.get("session_status"),
-            }
-        )
-        for row in lease_rows
-    )
-
-    degraded_reason = (
-        None
-        if has_work_items
-        else "work_items table absent; lease rows carry no work linkage"
-    )
-    lease_status = STATUS_OK if has_work_items else STATUS_DEGRADED
-
-    return (
-        InventorySection(
-            name="sessions",
-            authority=AUTHORITY_CONTROL_PLANE_DB,
-            status=STATUS_OK,
-            items=sessions,
-            scan_ms=round(elapsed, 3),
-        ),
-        InventorySection(
-            name="leases",
-            authority=AUTHORITY_CONTROL_PLANE_DB,
-            status=lease_status,
-            items=leases,
-            reason=degraded_reason,
-            scan_ms=round(elapsed, 3),
-        ),
-    )
-
-
-def _is_expired(expires_at: str | None, *, now: datetime) -> bool | None:
-    if not expires_at:
-        return None
-    text = str(expires_at).strip().replace("Z", "+00:00")
-    try:
-        parsed = datetime.fromisoformat(text)
-    except ValueError:
-        return None
-    if parsed.tzinfo is None:
-        parsed = parsed.replace(tzinfo=timezone.utc)
-    return parsed <= now
-
-
-# ── durable issue locks (filesystem) ─────────────────────────────────────────
-
-
-def _load_locks_section(*, lock_dir: str | None = None) -> InventorySection:
-    started = time.perf_counter()
-    try:
-        paths = issue_lock_store.iter_lock_files(lock_dir)
-    except OSError as exc:
-        return InventorySection(
-            name="locks",
-            authority=AUTHORITY_FILESYSTEM,
-            status=STATUS_UNAVAILABLE,
-            reason=f"issue lock directory could not be listed: {exc}",
-        )
-
-    items: list[dict[str, Any]] = []
-    unreadable = 0
-    for path in paths:
-        try:
-            record = issue_lock_store.read_lock_file(path)
-        except (OSError, ValueError):
-            unreadable += 1
-            continue
-        if not record:
-            unreadable += 1
-            continue
-        try:
-            freshness = issue_lock_store.assess_lock_freshness(record)
-        except Exception:  # noqa: BLE001 — a read-only view never raises
-            freshness = {"status": "unknown", "live": False, "stale": False}
-        claimant = record.get("claimant") or (
-            (record.get("work_lease") or {}).get("claimant") or {}
-        )
-        items.append(
-            scrub(
-                {
-                    "issue_number": record.get("issue_number"),
-                    "branch_name": record.get("branch_name"),
-                    "remote": record.get("remote"),
-                    "org": record.get("org"),
-                    "repo": record.get("repo"),
-                    "worktree_path": record.get("worktree_path"),
-                    "pid": record.get("session_pid") or record.get("pid"),
-                    "pid_alive": issue_lock_store.is_process_alive(
-                        record.get("session_pid") or record.get("pid")
-                    ),
-                    "claimant_username": (claimant or {}).get("username"),
-                    "claimant_profile": (claimant or {}).get("profile"),
-                    "lock_generation": record.get("lock_generation"),
-                    "freshness_status": freshness.get("status"),
-                    "live": bool(freshness.get("live")),
-                    "stale": bool(freshness.get("stale")),
-                    "freshness_reason": freshness.get("reason"),
-                    "lock_path": record.get("lock_file_path") or path,
-                }
-            )
-        )
-
-    elapsed = (time.perf_counter() - started) * 1000.0
-    reason = (
-        f"{unreadable} lock file(s) were unreadable and are not represented"
-        if unreadable
-        else None
-    )
-    return InventorySection(
-        name="locks",
-        authority=AUTHORITY_FILESYSTEM,
-        status=STATUS_DEGRADED if unreadable else STATUS_OK,
-        items=tuple(items),
-        reason=reason,
-        scan_ms=round(elapsed, 3),
-    )
-
-
-# ── worktrees (filesystem, via the #432 scanner) ─────────────────────────────
-
-
-def _load_worktrees_section(
-    *, load_hygiene: Callable[[], Any] | None = None
-) -> InventorySection:
-    started = time.perf_counter()
-    try:
-        loader = load_hygiene
-        if loader is None:
-            from webui.worktree_scanner import load_hygiene_snapshot
-
-            loader = load_hygiene_snapshot
-        snapshot = loader()
-    except Exception as exc:  # noqa: BLE001 — fail soft, never fail the request
-        return InventorySection(
-            name="worktrees",
-            authority=AUTHORITY_FILESYSTEM,
-            status=STATUS_UNAVAILABLE,
-            reason=f"worktree scan failed: {exc}",
-        )
-
-    items = tuple(
-        scrub(
-            {
-                "rel_path": entry.rel_path,
-                "folder_name": entry.folder_name,
-                "classification": entry.classification,
-                "branch": entry.branch,
-                "head_sha": entry.head_sha,
-                "dirty_tracked": entry.dirty_tracked,
-                "dirty_untracked": entry.dirty_untracked,
-                "detached": entry.detached,
-                "registered_worktree": entry.registered_worktree,
-                "notes": entry.notes,
-            }
-        )
-        for entry in snapshot.entries
-    )
-    scan_error = getattr(snapshot, "scan_error", None)
-    elapsed = (time.perf_counter() - started) * 1000.0
-    return InventorySection(
-        name="worktrees",
-        authority=AUTHORITY_FILESYSTEM,
-        status=STATUS_DEGRADED if scan_error else STATUS_OK,
-        items=items,
-        reason=scan_error,
-        scan_ms=round(elapsed, 3),
-    )
-
-
-# ── namespaces / capability summary ──────────────────────────────────────────
-
-
-def _load_namespaces_section() -> InventorySection:
-    started = time.perf_counter()
-    try:
-        from gitea_auth import get_profile
-
-        profile = get_profile() or {}
-    except Exception as exc:  # noqa: BLE001
-        return InventorySection(
-            name="namespaces",
-            authority=AUTHORITY_FILESYSTEM,
-            status=STATUS_UNAVAILABLE,
-            reason=f"active profile could not be resolved: {exc}",
-        )
-
-    allowed = list(profile.get("allowed_operations") or [])
-    forbidden = list(profile.get("forbidden_operations") or [])
-    profile_name = str(profile.get("profile_name") or "")
-
-    namespace = None
-    try:
-        import role_namespace_gate
-
-        namespace = role_namespace_gate.infer_mcp_namespace(profile_name)
-    except Exception:  # noqa: BLE001 — namespace inference is advisory
-        namespace = None
-
-    item = scrub(
-        {
-            "profile_name": profile_name,
-            "role": profile.get("role"),
-            "mcp_namespace": namespace,
-            "allowed_operations": sorted(allowed),
-            "forbidden_operations": sorted(forbidden),
-            "capability_summary": {
-                "can_author": "gitea.pr.create" in allowed,
-                "can_review": "gitea.pr.approve" in allowed,
-                "can_merge": "gitea.pr.merge" in allowed,
-                "can_close_pr": "gitea.pr.close" in allowed,
-            },
-            "active": True,
-        }
-    )
-    elapsed = (time.perf_counter() - started) * 1000.0
-    return InventorySection(
-        name="namespaces",
-        authority=AUTHORITY_FILESYSTEM,
-        status=STATUS_OK,
-        items=(item,),
-        reason=(
-            "only the profile serving this web process is observable; other "
-            "namespaces are not enumerable from here"
-        ),
-        scan_ms=round(elapsed, 3),
-    )
-
-
-# ── correlation and collision detection ──────────────────────────────────────
-
-
-def _issue_from_branch(branch: str | None) -> int | None:
-    match = re.search(r"issue-(\d+)", str(branch or ""), re.IGNORECASE)
-    return int(match.group(1)) if match else None
-
-
-def correlate(
-    *,
-    leases: InventorySection,
-    locks: InventorySection,
-    worktrees: InventorySection,
-    sessions: InventorySection,
-) -> tuple[tuple[dict[str, Any], ...], tuple[CollisionSignal, ...]]:
-    """Join lease owner ↔ lock ↔ worktree ↔ namespace where evidence allows.
-
-    Correlation rows are emitted from whatever sections did load. Collision
-    signals are only emitted from sections that are ``ok``: a conflict inferred
-    from a partially-read source would be a false accusation.
-    """
-    correlations: list[dict[str, Any]] = []
-    collisions: list[CollisionSignal] = []
-
-    worktree_by_branch: dict[str, dict[str, Any]] = {}
-    for entry in worktrees.items:
-        branch = (entry.get("branch") or "").strip()
-        if branch:
-            worktree_by_branch.setdefault(branch, entry)
-
-    session_by_id = {
-        str(s.get("session_id")): s for s in sessions.items if s.get("session_id")
-    }
-
-    # Lock-centred rows: a durable lock names an issue, a branch, and a worktree.
-    for lock in locks.items:
-        branch = (lock.get("branch_name") or "").strip()
-        worktree = worktree_by_branch.get(branch)
-        matching_leases = [
-            lease
-            for lease in leases.items
-            if lease.get("work_kind") == "issue"
-            and lease.get("work_number") == lock.get("issue_number")
-        ]
-        correlations.append(
-            {
-                "issue_number": lock.get("issue_number"),
-                "branch": branch or None,
-                "lock_live": bool(lock.get("live")),
-                "lock_claimant": lock.get("claimant_profile"),
-                "lock_pid": lock.get("pid"),
-                "lock_pid_alive": lock.get("pid_alive"),
-                "worktree_rel_path": (worktree or {}).get("rel_path"),
-                "worktree_classification": (worktree or {}).get("classification"),
-                "worktree_registered": (worktree or {}).get("registered_worktree"),
-                "lease_ids": [
-                    lease.get("lease_id")
-                    for lease in matching_leases
-                    if lease.get("lease_id")
-                ],
-                "lease_sessions": [
-                    lease.get("session_id")
-                    for lease in matching_leases
-                    if lease.get("session_id")
-                ],
-            }
-        )
-
-        if locks.ok and worktrees.ok:
-            # A claim whose lease window is still open but has no registered
-            # worktree is an anomaly regardless of whether its pid is alive; a
-            # fully time-expired lease is on its way out and is not flagged.
-            if (
-                lock.get("freshness_status") != "expired"
-                and branch
-                and worktree is None
-            ):
-                collisions.append(
-                    CollisionSignal(
-                        kind="lock-without-worktree",
-                        severity="warning",
-                        issue_number=lock.get("issue_number"),
-                        branch=branch,
-                        worktree_path=lock.get("worktree_path"),
-                        message=(
-                            f"Live lock on issue #{lock.get('issue_number')} names "
-                            f"branch {branch!r} but no registered worktree carries "
-                            "that branch (#404)"
-                        ),
-                    )
-                )
-
-        if locks.ok:
-            # A lock whose recorded pid is gone is held by nobody: a clean #753
-            # dead-session recovery candidate. Subclassify by the lease window,
-            # because the two cases need different operator urgency. When the
-            # window is still open the lock would read as live to a naive
-            # timestamp check even though the owner is dead — the more dangerous
-            # case — so it is flagged distinctly from a fully time-expired lease.
-            if lock.get("pid_alive") is False and lock.get("stale"):
-                if lock.get("freshness_status") == "expired":
-                    collisions.append(
-                        CollisionSignal(
-                            kind="stale-lock-dead-owner",
-                            severity="warning",
-                            issue_number=lock.get("issue_number"),
-                            branch=branch or None,
-                            message=(
-                                f"Lock on issue #{lock.get('issue_number')} is stale "
-                                f"and its recorded pid {lock.get('pid')} is not running "
-                                "(#753 dead-session recovery candidate)"
-                            ),
-                        )
-                    )
-                else:
-                    collisions.append(
-                        CollisionSignal(
-                            kind="live-lock-dead-owner",
-                            severity="warning",
-                            issue_number=lock.get("issue_number"),
-                            branch=branch or None,
-                            message=(
-                                f"Lock on issue #{lock.get('issue_number')} has an "
-                                "unexpired lease but its recorded pid "
-                                f"{lock.get('pid')} is not running; it would read as "
-                                "live to a timestamp check (#753 dead-session recovery "
-                                "candidate)"
-                            ),
-                        )
-                    )
-            # The #635 trap: the lease has expired but the recorded pid is a
-            # still-running daemon, so neither dead-pid reclaim nor exact-owner
-            # renewal applies. This is the collision an operator must see.
-            elif (
-                lock.get("freshness_status") == "expired"
-                and lock.get("pid_alive") is True
-            ):
-                collisions.append(
-                    CollisionSignal(
-                        kind="expired-lock-live-owner",
-                        severity="error",
-                        issue_number=lock.get("issue_number"),
-                        branch=branch or None,
-                        message=(
-                            f"Lock on issue #{lock.get('issue_number')} has an expired "
-                            f"lease but its recorded pid {lock.get('pid')} is still "
-                            "running (daemon-pid deadlock; needs an operator decision, "
-                            "#635/#760)"
-                        ),
-                    )
-                )
-
-    # Two live locks on one branch, or two active leases on one work item.
-    if locks.ok:
-        by_branch: dict[str, list[dict[str, Any]]] = {}
-        for lock in locks.items:
-            if not lock.get("live"):
-                continue
-            branch = (lock.get("branch_name") or "").strip()
-            if branch:
-                by_branch.setdefault(branch, []).append(lock)
-        for branch, entries in sorted(by_branch.items()):
-            if len(entries) > 1:
-                collisions.append(
-                    CollisionSignal(
-                        kind="duplicate-live-lock",
-                        severity="error",
-                        branch=branch,
-                        message=(
-                            f"{len(entries)} live locks name branch {branch!r}: "
-                            "issues "
-                            + ", ".join(
-                                f"#{e.get('issue_number')}" for e in entries
-                            )
-                        ),
-                    )
-                )
-
-    if leases.ok:
-        by_work: dict[tuple[str, int], list[dict[str, Any]]] = {}
-        for lease in leases.items:
-            if str(lease.get("status") or "").lower() != "active":
-                continue
-            kind = str(lease.get("work_kind") or "").strip().lower()
-            number = lease.get("work_number")
-            if not kind or number is None:
-                continue
-            by_work.setdefault((kind, int(number)), []).append(lease)
-        for (kind, number), entries in sorted(by_work.items()):
-            sessions_held = {
-                str(e.get("session_id")) for e in entries if e.get("session_id")
-            }
-            if len(sessions_held) > 1:
-                collisions.append(
-                    CollisionSignal(
-                        kind="concurrent-active-lease",
-                        severity="error",
-                        issue_number=number if kind == "issue" else None,
-                        session_ids=tuple(sorted(sessions_held)),
-                        message=(
-                            f"{len(sessions_held)} sessions hold an active lease on "
-                            f"{kind} #{number}"
-                        ),
-                    )
-                )
-            for entry in entries:
-                if entry.get("expired") is True:
-                    collisions.append(
-                        CollisionSignal(
-                            kind="active-lease-past-expiry",
-                            severity="warning",
-                            issue_number=number if kind == "issue" else None,
-                            session_ids=(
-                                (str(entry.get("session_id")),)
-                                if entry.get("session_id")
-                                else ()
-                            ),
-                            message=(
-                                f"Lease {entry.get('lease_id')} on {kind} #{number} "
-                                "is still marked active past its expiry"
-                            ),
-                        )
-                    )
-
-    # A lease whose owning session is gone is an orphan, not free work.
-    if leases.ok and sessions.ok:
-        for lease in leases.items:
-            if str(lease.get("status") or "").lower() != "active":
-                continue
-            session_id = str(lease.get("session_id") or "")
-            if session_id and session_id not in session_by_id:
-                collisions.append(
-                    CollisionSignal(
-                        kind="orphan-lease",
-                        severity="error",
-                        session_ids=(session_id,),
-                        message=(
-                            f"Active lease {lease.get('lease_id')} names session "
-                            f"{session_id}, which has no session record"
-                        ),
-                    )
-                )
-
-    return tuple(correlations), tuple(collisions)
-
-
-# ── snapshot assembly ────────────────────────────────────────────────────────
-
-
-def load_inventory_snapshot(
-    *,
-    db_path: str | None = None,
-    lock_dir: str | None = None,
-    load_hygiene: Callable[[], Any] | None = None,
-    include: tuple[str, ...] | None = None,
-) -> InventorySnapshot:
-    """Build the unified inventory snapshot.
-
-    Every section is loaded independently and fails soft. *include* restricts
-    the sections that are scanned; omitted sections are simply absent rather
-    than reported as empty, so a resource-split request cannot be mistaken for
-    a whole-inventory answer.
-    """
-    started = time.perf_counter()
-    wanted = tuple(include) if include else SECTION_NAMES
-
-    sections: list[InventorySection] = []
-    sessions_section: InventorySection | None = None
-    leases_section: InventorySection | None = None
-
-    if "sessions" in wanted or "leases" in wanted:
-        sessions_section, leases_section = _load_cp_db_sections(db_path=db_path)
-        if "sessions" in wanted:
-            sections.append(sessions_section)
-        if "leases" in wanted:
-            sections.append(leases_section)
-
-    locks_section = (
-        _load_locks_section(lock_dir=lock_dir)
-        if "locks" in wanted
-        else _empty_section("locks", AUTHORITY_FILESYSTEM)
-    )
-    if "locks" in wanted:
-        sections.append(locks_section)
-
-    worktrees_section = (
-        _load_worktrees_section(load_hygiene=load_hygiene)
-        if "worktrees" in wanted
-        else _empty_section("worktrees", AUTHORITY_FILESYSTEM)
-    )
-    if "worktrees" in wanted:
-        sections.append(worktrees_section)
-
-    if "namespaces" in wanted:
-        sections.append(_load_namespaces_section())
-
-    correlations, collisions = correlate(
-        leases=leases_section or _empty_section("leases", AUTHORITY_CONTROL_PLANE_DB),
-        locks=locks_section,
-        worktrees=worktrees_section,
-        sessions=sessions_section
-        or _empty_section("sessions", AUTHORITY_CONTROL_PLANE_DB),
-    )
-
-    elapsed = (time.perf_counter() - started) * 1000.0
-    index = {section.name: section for section in sections}
-    return InventorySnapshot(
-        generated_at=datetime.now(timezone.utc).isoformat(),
-        sections=tuple(sections),
-        collisions=collisions,
-        correlations=correlations,
-        scan_ms=round(elapsed, 3),
-        _section_index=index,
-    )
-
-
-def _empty_section(name: str, authority: str) -> InventorySection:
-    """A section that was not requested — never a claim that it is empty."""
-    return InventorySection(
-        name=name,
-        authority=authority,
-        status=STATUS_UNAVAILABLE,
-        reason="section not requested in this scan",
-    )
-
-
-def snapshot_to_dict(snapshot: InventorySnapshot) -> dict[str, Any]:
-    """Serialize the snapshot for the versioned API."""
-    return {
-        "api_version": snapshot.api_version,
-        "schema_version": snapshot.schema_version,
-        "generated_at": snapshot.generated_at,
-        "status": snapshot.status,
-        "scan_ms": snapshot.scan_ms,
-        "ownership_authority_complete": snapshot.ownership_authority_complete,
-        "ownership_note": (
-            "Every ownership source read cleanly; an item absent from leases "
-            "and locks is genuinely unclaimed."
-            if snapshot.ownership_authority_complete
-            else "One or more ownership sources are degraded; nothing in this "
-            "snapshot may be treated as unowned. Collisions are reported only "
-            "from sections that read cleanly."
-        ),
-        "degraded_sections": list(snapshot.degraded_sections),
-        "field_authority": {
-            "sessions": AUTHORITY_CONTROL_PLANE_DB,
-            "leases": AUTHORITY_CONTROL_PLANE_DB,
-            "locks": AUTHORITY_FILESYSTEM,
-            "worktrees": AUTHORITY_FILESYSTEM,
-            "namespaces": AUTHORITY_FILESYSTEM,
-        },
-        "sections": {section.name: section.to_dict() for section in snapshot.sections},
-        "correlations": [dict(row) for row in snapshot.correlations],
-        "collisions": [signal.to_dict() for signal in snapshot.collisions],
-    }

webui/layout.py

@@ -2,28 +2,66 @@
 
 from __future__ import annotations
 
-NAV_ITEMS = (
+import os
-    ("/", "Home"),
+
-    ("/queue", "Queue"),
+from webui.nav import NAV_GROUPS
-    ("/projects", "Projects"),
-    ("/prompts", "Prompts"),
-    ("/runtime", "Runtime"),
-    ("/audit", "Audit"),
-    ("/worktrees", "Worktrees"),
-    ("/leases", "Leases"),
-    ("/actions", "Actions"),
-)
 
 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(
+    nav_links = _render_nav()
-        f'<a href="{href}">{label}</a>' for href, label in NAV_ITEMS
+    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())

webui/project_registry.py

@@ -1,16 +1,4 @@
-"""Load and validate the web UI project registry (#427, evolved for #635).
+"""Load and validate the web UI project registry (#427)."""
-
-Phase 1 of the console architecture ADR keeps this loader read-only. It owns
-the versioned project registry contract served at ``/api/v1/projects``:
-
-* the on-disk file carries a ``version`` (schema version 1 or 2);
-* version 1 files stay loadable and are normalized with explicit defaults, so
-  an operator registry written for #427 keeps working;
-* every validation failure raises :class:`RegistryError`, which carries an
-  actionable ``remediation`` string instead of leaking a traceback;
-* serialization never emits credentials — credential-shaped keys are rejected
-  at load time, before any DTO is built.
-"""
 
 from __future__ import annotations
 
@@ -20,59 +8,7 @@ from dataclasses import dataclass
 from pathlib import Path
 from typing import Any
 
-from webui.registry_safety import is_forbidden_key
+from webui.registry_safety import reject_credential_keys as _reject_credential_keys
-
-#: Version of the JSON contract served under ``/api/v1/...``.
-REGISTRY_API_VERSION = "v1"
-
-#: Schema version written by this repository's packaged registry.
-CURRENT_SCHEMA_VERSION = 2
-
-#: Schema versions this loader accepts. Version 1 is normalized on load.
-SUPPORTED_SCHEMA_VERSIONS = (1, 2)
-
-#: Lifecycle state of a registered project.
-PROJECT_STATUSES = ("active", "onboarding", "paused", "archived")
-_DEFAULT_PROJECT_STATUS = "active"
-
-#: Completion state of a single onboarding step.
-ONBOARDING_STATES = ("complete", "pending", "blocked", "not_applicable")
-_DEFAULT_ONBOARDING_STATE = "pending"
-
-#: Redacted, last-seen health of a project's control plane.
-HEALTH_STATUSES = ("healthy", "degraded", "unreachable", "unknown")
-
-class RegistryError(ValueError):
-    """A registry file could not be loaded or failed validation.
-
-    Carries an operator-facing ``remediation`` so routes can fail closed with
-    an actionable message rather than a stack trace.
-    """
-
-    def __init__(
-        self,
-        message: str,
-        *,
-        remediation: str,
-        source_path: Path | None = None,
-        field_path: str | None = None,
-    ) -> None:
-        super().__init__(message)
-        self.message = message
-        self.remediation = remediation
-        self.source_path = source_path
-        self.field_path = field_path
-
-    def to_dict(self) -> dict[str, Any]:
-        """Serialize for a fail-closed JSON error response."""
-        return {
-            "error": "registry_invalid",
-            "detail": self.message,
-            "remediation": self.remediation,
-            "field_path": self.field_path,
-            "source_path": str(self.source_path) if self.source_path else None,
-        }
-
 
 _REQUIRED_PROJECT_FIELDS = (
     "id",
@@ -93,30 +29,6 @@ class OnboardingStep:
     id: str
     title: str
     description: str
-    state: str = _DEFAULT_ONBOARDING_STATE
-    required: bool = True
-
-
-@dataclass(frozen=True)
-class OnboardingSummary:
-    """Aggregate onboarding progress for a single project."""
-
-    total: int
-    complete: int
-    pending: int
-    blocked: int
-    not_applicable: int
-    required_outstanding: int
-    onboarding_complete: bool
-
-
-@dataclass(frozen=True)
-class ProjectHealth:
-    """Redacted last-seen health. Never carries endpoints or credentials."""
-
-    status: str
-    checked_at: str | None
-    detail: str | None
 
 
 @dataclass(frozen=True)
@@ -131,13 +43,6 @@ class ProjectRecord:
     workflow_paths: dict[str, str]
     schema_paths: dict[str, str]
     onboarding_checklist: tuple[OnboardingStep, ...]
-    status: str = _DEFAULT_PROJECT_STATUS
-    remote_name: str | None = None
-    last_seen_health: ProjectHealth | None = None
-
-    @property
-    def repo_full_name(self) -> str:
-        return f"{self.gitea_owner}/{self.repo_name}"
 
 
 @dataclass(frozen=True)
@@ -146,15 +51,6 @@ class ProjectRegistry:
     projects: tuple[ProjectRecord, ...]
     source_path: Path
 
-    @property
-    def schema_version(self) -> int:
-        """Alias of :attr:`version` — the schema version read from disk."""
-        return self.version
-
-    @property
-    def api_version(self) -> str:
-        return REGISTRY_API_VERSION
-
 
 def default_registry_path() -> Path:
     override = os.environ.get("WEBUI_PROJECT_REGISTRY", "").strip()
@@ -163,221 +59,40 @@ def default_registry_path() -> Path:
     return (Path(__file__).resolve().parent / "data" / "projects.registry.json").resolve()
 
 
-def _reject_credential_keys(obj: Any, *, path: str = "", source: Path | None = None) -> None:
+def _parse_onboarding(raw: list[dict[str, Any]] | None) -> tuple[OnboardingStep, ...]:
-    """Recursive credential-key guard that reports an actionable ``field_path``.
+    if not raw:
-
-    Key *shape* is decided by :func:`webui.registry_safety.is_forbidden_key`, the
-    single source of truth shared with the worker registry (#798).
-    """
-    if isinstance(obj, dict):
-        for key, value in obj.items():
-            key_path = f"{path}.{key}" if path else key
-            if is_forbidden_key(key):
-                raise RegistryError(
-                    f"registry must not store credentials ({key_path})",
-                    remediation=(
-                        f"Remove the credential-shaped key '{key_path}' from the registry. "
-                        "Tokens live in the keychain and are resolved server-side by "
-                        "gitea_auth; the registry is redacted metadata only."
-                    ),
-                    source_path=source,
-                    field_path=key_path,
-                )
-            _reject_credential_keys(value, path=key_path, source=source)
-    elif isinstance(obj, list):
-        for index, item in enumerate(obj):
-            _reject_credential_keys(item, path=f"{path}[{index}]", source=source)
-
-
-def _require_enum(
-    value: Any,
-    *,
-    allowed: tuple[str, ...],
-    field_path: str,
-    source: Path | None,
-) -> str:
-    text = str(value)
-    if text not in allowed:
-        raise RegistryError(
-            f"{field_path} must be one of {', '.join(allowed)} (got {text!r})",
-            remediation=(
-                f"Set {field_path} to one of: {', '.join(allowed)}. "
-                "Unknown values fail closed so the console never renders an "
-                "unverified state."
-            ),
-            source_path=source,
-            field_path=field_path,
-        )
-    return text
-
-
-def _parse_onboarding(
-    raw: Any,
-    *,
-    project_path: str,
-    source: Path | None,
-) -> tuple[OnboardingStep, ...]:
-    if raw is None:
         return ()
-    if not isinstance(raw, list):
-        raise RegistryError(
-            f"{project_path}.onboarding_checklist must be an array",
-            remediation=(
-                f"Rewrite {project_path}.onboarding_checklist as a JSON array of "
-                "steps with id, title, description, and optional state."
-            ),
-            source_path=source,
-            field_path=f"{project_path}.onboarding_checklist",
-        )
     steps: list[OnboardingStep] = []
-    for index, item in enumerate(raw):
+    for item in raw:
-        step_path = f"{project_path}.onboarding_checklist[{index}]"
-        if not isinstance(item, dict):
-            raise RegistryError(
-                f"{step_path} must be an object",
-                remediation=f"Rewrite {step_path} as an object with id, title, description.",
-                source_path=source,
-                field_path=step_path,
-            )
-        missing = [field for field in ("id", "title", "description") if field not in item]
-        if missing:
-            raise RegistryError(
-                f"{step_path} missing required fields: {', '.join(missing)}",
-                remediation=(
-                    f"Add {', '.join(missing)} to {step_path}. Every onboarding step "
-                    "must be self-describing for an operator who has no chat history."
-                ),
-                source_path=source,
-                field_path=step_path,
-            )
-        state = _require_enum(
-            item.get("state", _DEFAULT_ONBOARDING_STATE),
-            allowed=ONBOARDING_STATES,
-            field_path=f"{step_path}.state",
-            source=source,
-        )
         steps.append(
             OnboardingStep(
                 id=str(item["id"]),
                 title=str(item["title"]),
                 description=str(item["description"]),
-                state=state,
-                required=bool(item.get("required", True)),
             )
         )
     return tuple(steps)
 
 
-def _parse_health(
+def _parse_project(raw: dict[str, Any]) -> ProjectRecord:
-    raw: Any,
-    *,
-    project_path: str,
-    source: Path | None,
-) -> ProjectHealth | None:
-    if raw is None:
-        return None
-    if not isinstance(raw, dict):
-        raise RegistryError(
-            f"{project_path}.last_seen_health must be an object when present",
-            remediation=(
-                f"Rewrite {project_path}.last_seen_health as an object with status "
-                f"(one of {', '.join(HEALTH_STATUSES)}), optional checked_at and detail, "
-                "or remove it. Never store endpoints or credentials here."
-            ),
-            source_path=source,
-            field_path=f"{project_path}.last_seen_health",
-        )
-    status = _require_enum(
-        raw.get("status", "unknown"),
-        allowed=HEALTH_STATUSES,
-        field_path=f"{project_path}.last_seen_health.status",
-        source=source,
-    )
-    checked_at = raw.get("checked_at")
-    detail = raw.get("detail")
-    return ProjectHealth(
-        status=status,
-        checked_at=str(checked_at) if checked_at is not None else None,
-        detail=str(detail) if detail is not None else None,
-    )
-
-
-def _parse_project(raw: Any, *, index: int, source: Path | None) -> ProjectRecord:
-    project_path = f"projects[{index}]"
-    if not isinstance(raw, dict):
-        raise RegistryError(
-            f"{project_path} must be an object",
-            remediation=f"Rewrite {project_path} as a JSON object describing one project.",
-            source_path=source,
-            field_path=project_path,
-        )
-
     missing = [field for field in _REQUIRED_PROJECT_FIELDS if field not in raw]
     if missing:
-        raise RegistryError(
+        raise ValueError(f"project missing required fields: {', '.join(missing)}")
-            f"{project_path} missing required fields: {', '.join(missing)}",
-            remediation=(
-                f"Add {', '.join(missing)} to {project_path}. See "
-                "docs/webui-project-registry-api.md for the field-by-field contract."
-            ),
-            source_path=source,
-            field_path=project_path,
-        )
 
     profiles = raw["profiles"]
     if not isinstance(profiles, dict):
-        raise RegistryError(
+        raise ValueError("profiles must be an object")
-            f"{project_path}.profiles must be an object",
-            remediation=(
-                f"Rewrite {project_path}.profiles as an object mapping "
-                f"{', '.join(_REQUIRED_PROFILE_ROLES)} to MCP profile names."
-            ),
-            source_path=source,
-            field_path=f"{project_path}.profiles",
-        )
     for role in _REQUIRED_PROFILE_ROLES:
         if role not in profiles or not profiles[role]:
-            raise RegistryError(
+            raise ValueError(f"profiles.{role} is required")
-                f"{project_path}.profiles.{role} is required",
-                remediation=(
-                    f"Set {project_path}.profiles.{role} to the configured MCP profile "
-                    "name for that role. Role separation is a workflow-safety invariant."
-                ),
-                source_path=source,
-                field_path=f"{project_path}.profiles.{role}",
-            )
 
     workflow_paths = raw["workflow_paths"]
     if not isinstance(workflow_paths, dict) or not workflow_paths:
-        raise RegistryError(
+        raise ValueError("workflow_paths must be a non-empty object")
-            f"{project_path}.workflow_paths must be a non-empty object",
-            remediation=(
-                f"Add at least a 'skill' entry to {project_path}.workflow_paths pointing "
-                "at the project's canonical workflow skill."
-            ),
-            source_path=source,
-            field_path=f"{project_path}.workflow_paths",
-        )
 
     schema_paths = raw.get("schema_paths") or {}
     if not isinstance(schema_paths, dict):
-        raise RegistryError(
+        raise ValueError("schema_paths must be an object when present")
-            f"{project_path}.schema_paths must be an object when present",
-            remediation=(
-                f"Rewrite {project_path}.schema_paths as an object of label to repo path, "
-                "or remove it."
-            ),
-            source_path=source,
-            field_path=f"{project_path}.schema_paths",
-        )
-
-    status = _require_enum(
-        raw.get("status", _DEFAULT_PROJECT_STATUS),
-        allowed=PROJECT_STATUSES,
-        field_path=f"{project_path}.status",
-        source=source,
-    )
-    remote_name = raw.get("remote_name")
 
     return ProjectRecord(
         id=str(raw["id"]),
@@ -389,211 +104,61 @@ def _parse_project(raw: Any, *, index: int, source: Path | None) -> ProjectRecor
         profiles={role: str(profiles[role]) for role in _REQUIRED_PROFILE_ROLES},
         workflow_paths={key: str(value) for key, value in workflow_paths.items()},
         schema_paths={key: str(value) for key, value in schema_paths.items()},
-        onboarding_checklist=_parse_onboarding(
+        onboarding_checklist=_parse_onboarding(raw.get("onboarding_checklist")),
-            raw.get("onboarding_checklist"),
-            project_path=project_path,
-            source=source,
-        ),
-        status=status,
-        remote_name=str(remote_name) if remote_name else None,
-        last_seen_health=_parse_health(
-            raw.get("last_seen_health"),
-            project_path=project_path,
-            source=source,
-        ),
     )
 
 
 def load_registry(path: Path | None = None) -> ProjectRegistry:
-    """Load the versioned project registry from disk.
+    """Load the versioned project registry from disk."""
-
-    Raises:
-        RegistryError: whenever the file is unreadable, is not valid JSON, or
-            fails schema validation. The error carries an operator remediation.
-    """
     source = (path or default_registry_path()).resolve()
-    try:
+    raw_text = source.read_text(encoding="utf-8")
-        raw_text = source.read_text(encoding="utf-8")
+    payload = json.loads(raw_text)
-    except OSError as exc:
-        raise RegistryError(
-            f"registry file could not be read: {exc.strerror or exc}",
-            remediation=(
-                f"Create a readable registry at {source}, or point "
-                "WEBUI_PROJECT_REGISTRY at an existing file."
-            ),
-            source_path=source,
-        ) from exc
-
-    try:
-        payload = json.loads(raw_text)
-    except json.JSONDecodeError as exc:
-        raise RegistryError(
-            f"registry is not valid JSON: {exc.msg} (line {exc.lineno}, column {exc.colno})",
-            remediation=(
-                f"Fix the JSON syntax in {source} at line {exc.lineno}, column {exc.colno}."
-            ),
-            source_path=source,
-        ) from exc
-
     if not isinstance(payload, dict):
-        raise RegistryError(
+        raise ValueError("registry root must be an object")
-            "registry root must be an object",
-            remediation=(
-                "Wrap the registry in a JSON object with 'version' and 'projects' keys."
-            ),
-            source_path=source,
-        )
 
     version = payload.get("version")
-    if version not in SUPPORTED_SCHEMA_VERSIONS:
+    if version != 1:
-        supported = ", ".join(str(item) for item in SUPPORTED_SCHEMA_VERSIONS)
+        raise ValueError(f"unsupported registry version: {version!r}")
-        raise RegistryError(
-            f"unsupported registry version: {version!r}",
-            remediation=(
-                f"Set 'version' to one of {supported} (current schema is "
-                f"{CURRENT_SCHEMA_VERSION}). Migration notes live in "
-                "docs/webui-project-registry-api.md."
-            ),
-            source_path=source,
-            field_path="version",
-        )
 
-    _reject_credential_keys(payload, source=source)
+    _reject_credential_keys(payload)
 
     projects_raw = payload.get("projects")
     if not isinstance(projects_raw, list) or not projects_raw:
-        raise RegistryError(
+        raise ValueError("projects must be a non-empty array")
-            "projects must be a non-empty array",
-            remediation=(
-                "Add at least one project object to 'projects'. An empty console "
-                "registry fails closed rather than rendering a blank inventory."
-            ),
-            source_path=source,
-            field_path="projects",
-        )
 
-    projects = tuple(
+    projects = tuple(_parse_project(item) for item in projects_raw)
-        _parse_project(item, index=index, source=source)
+    return ProjectRegistry(version=version, projects=projects, source_path=source)
-        for index, item in enumerate(projects_raw)
-    )
-    return ProjectRegistry(version=int(version), projects=projects, source_path=source)
-
-
-def onboarding_summary(project: ProjectRecord) -> OnboardingSummary:
-    """Aggregate a project's onboarding checklist state."""
-    steps = project.onboarding_checklist
-    counts = {state: 0 for state in ONBOARDING_STATES}
-    for step in steps:
-        counts[step.state] += 1
-    required_outstanding = sum(
-        1
-        for step in steps
-        if step.required and step.state in ("pending", "blocked")
-    )
-    return OnboardingSummary(
-        total=len(steps),
-        complete=counts["complete"],
-        pending=counts["pending"],
-        blocked=counts["blocked"],
-        not_applicable=counts["not_applicable"],
-        required_outstanding=required_outstanding,
-        onboarding_complete=required_outstanding == 0,
-    )
 
 
 def project_to_dict(project: ProjectRecord) -> dict[str, Any]:
-    """Serialize a project for JSON API responses and HTML views.
+    """Serialize a project for JSON API responses."""
-
-    The HTML views render from this same DTO, so the console and the API can
-    never disagree about a project's status or onboarding progress.
-    """
-    summary = onboarding_summary(project)
-    health = project.last_seen_health
     return {
         "id": project.id,
         "repo_name": project.repo_name,
         "gitea_owner": project.gitea_owner,
-        "repo_full_name": project.repo_full_name,
         "remote_host": project.remote_host,
-        "remote_name": project.remote_name,
         "default_branch": project.default_branch,
         "local_checkout_path": project.local_checkout_path,
-        "status": project.status,
         "profiles": dict(project.profiles),
         "workflow_paths": dict(project.workflow_paths),
         "schema_paths": dict(project.schema_paths),
         "onboarding_checklist": [
-            {
+            {"id": step.id, "title": step.title, "description": step.description}
-                "id": step.id,
-                "title": step.title,
-                "description": step.description,
-                "state": step.state,
-                "required": step.required,
-            }
             for step in project.onboarding_checklist
         ],
-        "onboarding_summary": {
-            "total": summary.total,
-            "complete": summary.complete,
-            "pending": summary.pending,
-            "blocked": summary.blocked,
-            "not_applicable": summary.not_applicable,
-            "required_outstanding": summary.required_outstanding,
-            "onboarding_complete": summary.onboarding_complete,
-        },
-        "last_seen_health": (
-            None
-            if health is None
-            else {
-                "status": health.status,
-                "checked_at": health.checked_at,
-                "detail": health.detail,
-            }
-        ),
     }
 
 
 def registry_to_dict(registry: ProjectRegistry) -> dict[str, Any]:
-    """Serialize the whole registry, including API provenance (#632 section 6)."""
     return {
-        "api_version": registry.api_version,
-        "schema_version": registry.schema_version,
-        # Retained for the unversioned MVP alias consumers (#427).
         "version": registry.version,
         "source_path": str(registry.source_path),
-        "source": {
-            "kind": "file",
-            "path": str(registry.source_path),
-            "inventory_complete": True,
-        },
-        "project_count": len(registry.projects),
         "projects": [project_to_dict(project) for project in registry.projects],
     }
 
 
-def project_detail_to_dict(
-    registry: ProjectRegistry,
-    project: ProjectRecord,
-) -> dict[str, Any]:
-    """Serialize a single project for ``/api/v1/projects/{project_id}``."""
-    return {
-        "api_version": registry.api_version,
-        "schema_version": registry.schema_version,
-        "source": {
-            "kind": "file",
-            "path": str(registry.source_path),
-            "inventory_complete": True,
-        },
-        "project": project_to_dict(project),
-    }
-
-
 def find_project(registry: ProjectRegistry, project_id: str) -> ProjectRecord | None:
     for project in registry.projects:
         if project.id == project_id:
             return project
-    return None
+    return None
-
-
-def known_project_ids(registry: ProjectRegistry) -> list[str]:
-    return [project.id for project in registry.projects]

webui/project_views.py

@@ -1,65 +1,34 @@
-"""HTML views for project registry pages (#427, evolved for #635).
+"""HTML views for project registry pages (#427)."""
-
-Every view renders from :func:`webui.project_registry.project_to_dict`, the
-same DTO the ``/api/v1/projects`` JSON responses use, so the HTML console and
-the API can never disagree about status or onboarding progress.
-"""
 
 from __future__ import annotations
 
 import html
-from typing import Any
 
 from webui.layout import render_page
-from webui.project_registry import (
+from webui.project_registry import ProjectRecord, ProjectRegistry
-    ProjectRecord,
-    ProjectRegistry,
-    RegistryError,
-    project_to_dict,
-)
-
-_STATE_LABELS = {
-    "complete": "Complete",
-    "pending": "Pending",
-    "blocked": "Blocked",
-    "not_applicable": "Not applicable",
-}
 
 
 def _escape(text: str) -> str:
     return html.escape(text, quote=True)
 
 
-def _progress_label(summary: dict[str, Any]) -> str:
-    total = summary["total"]
-    if not total:
-        return "no steps"
-    label = f"{summary['complete']}/{total} complete"
-    if summary["blocked"]:
-        label += f", {summary['blocked']} blocked"
-    return label
-
-
 def render_projects_list(registry: ProjectRegistry) -> str:
     rows = []
     for project in registry.projects:
-        dto = project_to_dict(project)
         rows.append(
             "<tr>"
-            f"<td><a href=\"/projects/{_escape(dto['id'])}\">{_escape(dto['repo_name'])}</a></td>"
+            f"<td><a href=\"/projects/{_escape(project.id)}\">{_escape(project.repo_name)}</a></td>"
-            f"<td>{_escape(dto['gitea_owner'])}</td>"
+            f"<td>{_escape(project.gitea_owner)}</td>"
-            f"<td>{_escape(dto['remote_host'])}</td>"
+            f"<td>{_escape(project.remote_host)}</td>"
-            f"<td>{_escape(dto['default_branch'])}</td>"
+            f"<td>{_escape(project.default_branch)}</td>"
-            f"<td><code>{_escape(dto['status'])}</code></td>"
+            f"<td><code>{_escape(project.profiles['author'])}</code></td>"
-            f"<td>{_escape(_progress_label(dto['onboarding_summary']))}</td>"
-            f"<td><code>{_escape(dto['profiles']['author'])}</code></td>"
             "</tr>"
         )
     table = (
         "<table class=\"registry\">"
         "<thead><tr>"
         "<th>Repository</th><th>Owner</th><th>Remote</th>"
-        "<th>Branch</th><th>Status</th><th>Onboarding</th><th>Author profile</th>"
+        "<th>Branch</th><th>Author profile</th>"
         "</tr></thead>"
         f"<tbody>{''.join(rows)}</tbody></table>"
     )
@@ -67,39 +36,32 @@ def render_projects_list(registry: ProjectRegistry) -> str:
         "<h2>Projects</h2>"
         "<p>Configured repositories managed by the MCP Control Plane.</p>"
         f"<p class=\"meta\">Registry: <code>{_escape(str(registry.source_path))}</code> "
-        f"(schema version {registry.schema_version}, "
+        f"(version {registry.version})</p>"
-        f"API {_escape(registry.api_version)})</p>"
         f"{table}"
-        "<p><a href=\"/api/v1/projects\">JSON API</a> "
+        "<p><a href=\"/api/projects\">JSON API</a></p>"
-        "(<a href=\"/api/projects\">unversioned alias</a>)</p>"
     )
     return render_page(title="Projects", body_html=body)
 
 
 def render_project_detail(project: ProjectRecord) -> str:
-    dto = project_to_dict(project)
     profile_rows = "".join(
         f"<tr><th>{_escape(role)}</th><td><code>{_escape(name)}</code></td></tr>"
-        for role, name in dto["profiles"].items()
+        for role, name in project.profiles.items()
     )
     workflow_rows = "".join(
         f"<tr><th>{_escape(key)}</th><td><code>{_escape(path)}</code></td></tr>"
-        for key, path in dto["workflow_paths"].items()
+        for key, path in project.workflow_paths.items()
     )
     schema_rows = "".join(
         f"<tr><th>{_escape(key)}</th><td><code>{_escape(path)}</code></td></tr>"
-        for key, path in dto["schema_paths"].items()
+        for key, path in project.schema_paths.items()
     )
     checklist_items = []
-    for index, step in enumerate(dto["onboarding_checklist"], start=1):
+    for index, step in enumerate(project.onboarding_checklist, start=1):
-        state_label = _STATE_LABELS.get(step["state"], step["state"])
-        requirement = "required" if step["required"] else "optional"
         checklist_items.append(
-            f"<li class=\"step-{_escape(step['state'])}\">"
+            "<li>"
-            f"<strong>{index}. {_escape(step['title'])}</strong>"
+            f"<strong>{index}. {_escape(step.title)}</strong>"
-            f" <span class=\"badge\">{_escape(state_label)}</span>"
+            f"<p>{_escape(step.description)}</p>"
-            f" <span class=\"meta\">({_escape(requirement)})</span>"
-            f"<p>{_escape(step['description'])}</p>"
             "</li>"
         )
     checklist_html = (
@@ -107,38 +69,16 @@ def render_project_detail(project: ProjectRecord) -> str:
         if checklist_items
         else "<p>No onboarding steps defined.</p>"
     )
-    summary = dto["onboarding_summary"]
-    summary_html = (
-        "<p class=\"meta\">Onboarding: "
-        f"{_escape(_progress_label(summary))}; required outstanding "
-        f"{summary['required_outstanding']}.</p>"
-    )
-    health = dto["last_seen_health"]
-    health_html = (
-        "<p class=\"meta\">No health probe recorded (Phase 1 is read-only).</p>"
-        if health is None
-        else (
-            "<table class=\"detail\">"
-            f"<tr><th>Status</th><td><code>{_escape(health['status'])}</code></td></tr>"
-            f"<tr><th>Checked at</th><td>{_escape(str(health['checked_at'] or 'unknown'))}</td></tr>"
-            f"<tr><th>Detail</th><td>{_escape(str(health['detail'] or ''))}</td></tr>"
-            "</table>"
-        )
-    )
-    remote_name = dto["remote_name"] or "unset"
     body = (
-        f"<h2>{_escape(dto['repo_name'])}</h2>"
+        f"<h2>{_escape(project.repo_name)}</h2>"
         "<p><a href=\"/projects\">← All projects</a></p>"
         "<h3>Identity</h3>"
         "<table class=\"detail\">"
-        f"<tr><th>Registry id</th><td><code>{_escape(dto['id'])}</code></td></tr>"
+        f"<tr><th>Registry id</th><td><code>{_escape(project.id)}</code></td></tr>"
-        f"<tr><th>Status</th><td><code>{_escape(dto['status'])}</code></td></tr>"
+        f"<tr><th>Gitea owner</th><td>{_escape(project.gitea_owner)}</td></tr>"
-        f"<tr><th>Gitea owner</th><td>{_escape(dto['gitea_owner'])}</td></tr>"
+        f"<tr><th>Remote host</th><td>{_escape(project.remote_host)}</td></tr>"
-        f"<tr><th>Repository</th><td><code>{_escape(dto['repo_full_name'])}</code></td></tr>"
+        f"<tr><th>Default branch</th><td><code>{_escape(project.default_branch)}</code></td></tr>"
-        f"<tr><th>Remote name</th><td><code>{_escape(remote_name)}</code></td></tr>"
+        f"<tr><th>Local checkout</th><td><code>{_escape(project.local_checkout_path)}</code></td></tr>"
-        f"<tr><th>Remote host</th><td>{_escape(dto['remote_host'])}</td></tr>"
-        f"<tr><th>Default branch</th><td><code>{_escape(dto['default_branch'])}</code></td></tr>"
-        f"<tr><th>Local checkout</th><td><code>{_escape(dto['local_checkout_path'])}</code></td></tr>"
         "</table>"
         "<h3>Profiles</h3>"
         f"<table class=\"detail\">{profile_rows}</table>"
@@ -146,30 +86,8 @@ def render_project_detail(project: ProjectRecord) -> str:
         f"<table class=\"detail\">{workflow_rows}</table>"
         "<h3>Schema paths</h3>"
         f"<table class=\"detail\">{schema_rows}</table>"
-        "<h3>Last seen health</h3>"
-        f"{health_html}"
         "<h3>Onboarding checklist</h3>"
-        "<p class=\"meta\">Read-only — complete these steps outside the UI.</p>"
+        "<p class=\"meta\">Read-only MVP — complete these steps outside the UI.</p>"
-        f"{summary_html}"
         f"{checklist_html}"
-        f"<p><a href=\"/api/v1/projects/{_escape(dto['id'])}\">JSON detail</a></p>"
     )
-    return render_page(title=dto["repo_name"], body_html=body)
+    return render_page(title=project.repo_name, body_html=body)
-
-
-def render_registry_error(error: RegistryError) -> str:
-    """Render a fail-closed page for an invalid registry."""
-    source = str(error.source_path) if error.source_path else "unknown"
-    field = error.field_path or "n/a"
-    body = (
-        "<h2>Project registry unavailable</h2>"
-        "<p>The registry failed validation, so the console refuses to render a "
-        "partial inventory.</p>"
-        "<table class=\"detail\">"
-        f"<tr><th>Detail</th><td>{_escape(error.message)}</td></tr>"
-        f"<tr><th>Field</th><td><code>{_escape(field)}</code></td></tr>"
-        f"<tr><th>Source</th><td><code>{_escape(source)}</code></td></tr>"
-        f"<tr><th>Remediation</th><td>{_escape(error.remediation)}</td></tr>"
-        "</table>"
-    )
-    return render_page(title="Project registry unavailable", body_html=body)