feat(webui): unified session/lease/lock/worktree inventory API (Closes #636 )

Co-Authored-By: Claude Opus 4.8 (1M context) <[email protected]>
Merge pull request 'feat(webui): versioned project registry API (Closes #635 )' (#819 ) from feat/issue-635-project-registry-api into master
2026-07-22 21:13:54 -04:00 · 2026-07-22 20:04:26 -05:00 · 2026-07-22 18:34:31 -05:00 · 2026-07-22 17:27:49 -05:00 · 2026-07-22 16:12:07 -04:00
11 changed files with 3354 additions and 96 deletions
@@ -43,6 +43,10 @@ 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 |
@@ -51,9 +55,11 @@ that govern when a write path may open (#632, epic #631).
 | `/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 (#427) |
+| `/projects` | Project registry list with status and onboarding progress (#427, #635) |
 | `/projects/{id}` | Project detail + onboarding checklist |
-| `/api/projects` | JSON registry export |
+| `/api/v1/projects` | Versioned JSON registry export (#635) |
 | `/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) |
@@ -206,6 +212,53 @@ 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
@@ -0,0 +1,213 @@
 # 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.
@@ -9173,11 +9173,27 @@ def gitea_publish_unpublished_issue_branch(
    if blocked:
        return blocked
-    verify_preflight_purity(remote, task=task, org=org, repo=repo)
+    # #815: resolve the caller's worktree *before* preflight and forward it, so
    # 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)
@@ -0,0 +1,622 @@
 """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()
@@ -0,0 +1,468 @@
 """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()
@@ -1,4 +1,4 @@
-"""Tests for web UI project registry (#427)."""
+"""Tests for web UI project registry (#427) and its API evolution (#635)."""
 import json
 import sys
 import tempfile
@@ -11,57 +11,246 @@ 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"
-class TestProjectRegistryLoader(unittest.TestCase):
+def _valid_project(**overrides):
    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, 1)
+        self.assertEqual(registry.version, CURRENT_SCHEMA_VERSION)
        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_registry_rejects_credential_keys(self):
+    def test_default_registry_onboarding_summary_is_complete(self):
-        payload = {
+        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(
-                    "id": "bad",
+                    onboarding_checklist=[
-                    "repo_name": "Bad",
+                        {"id": "step", "title": "Step", "description": "Do it"}
-                    "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",
                }
            ],
-        }
+        })
        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({
            "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": "a", "title": "A", "description": "d", "state": "almost"}
                    ]
                )
            ],
        })
        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:
-            json.dump(payload, handle)
+            handle.write("{not json")
            path = Path(handle.name)
-        try:
+        self._temp_paths.append(path)
-            with self.assertRaises(ValueError):
+        with self.assertRaises(RegistryError) as ctx:
-                load_registry(path)
+            load_registry(path)
-        finally:
+        self.assertIn("not valid JSON", ctx.exception.message)
-            path.unlink(missing_ok=True)
+        self.assertIn("line", ctx.exception.remediation)
    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()
@@ -81,31 +270,147 @@ 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_json(self):
+    def test_api_projects_alias_stays_compatible(self):
        response = self.client.get("/api/projects")
        self.assertEqual(response.status_code, 200)
        data = response.json()
-        self.assertEqual(data["version"], 1)
+        # #427 consumers keep these keys.
        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()
-        encoded = json.dumps(project_to_dict(registry.projects[0]))
+        dto = 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()
@@ -11,8 +11,20 @@ from starlette.routing import Route
 from webui.deployment_boundary import deployment_snapshot
 from webui.layout import render_page
-from webui.project_registry import find_project, load_registry, registry_to_dict
+from webui.project_registry import (
-from webui.project_views import render_project_detail, render_projects_list
+    ProjectRegistry,
    RegistryError,
    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
@@ -29,6 +41,11 @@ 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"})
@@ -81,14 +98,26 @@ 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 = load_registry()
+    registry, error = _load_project_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 = load_registry()
+    registry, error = _load_project_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(
@@ -106,10 +135,47 @@ async def project_detail(request: Request) -> HTMLResponse:
 async def api_projects(_request: Request) -> JSONResponse:
-    registry = load_registry()
+    """Unversioned MVP alias, retained through Phase 1 (#632 section 6)."""
    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())
@@ -241,6 +307,30 @@ 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":
@@ -268,6 +358,12 @@ 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"]),
@@ -291,6 +387,12 @@ 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(
                "/api/v1/inventory/{section}",
                api_inventory_section,
                methods=["GET"],
            ),
        ],
        exception_handlers={405: method_not_allowed},
    )
@@ -1,13 +1,15 @@
 {
-  "version": 1,
+  "version": 2,
  "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",
@@ -26,24 +28,32 @@
        {
          "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
        }
      ]
    }
  ]
-}
+}
@@ -0,0 +1,952 @@
 """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],
    }
@@ -1,4 +1,16 @@
-"""Load and validate the web UI project registry (#427)."""
+"""Load and validate the web UI project registry (#427, evolved for #635).
 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
@@ -8,7 +20,59 @@ from dataclasses import dataclass
 from pathlib import Path
 from typing import Any
-from webui.registry_safety import reject_credential_keys as _reject_credential_keys
+from webui.registry_safety import is_forbidden_key
 #: 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",
@@ -29,6 +93,30 @@ 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)
@@ -43,6 +131,13 @@ 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)
@@ -51,6 +146,15 @@ 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()
@@ -59,40 +163,221 @@ def default_registry_path() -> Path:
    return (Path(__file__).resolve().parent / "data" / "projects.registry.json").resolve()
-def _parse_onboarding(raw: list[dict[str, Any]] | None) -> tuple[OnboardingStep, ...]:
+def _reject_credential_keys(obj: Any, *, path: str = "", source: Path | None = None) -> None:
-    if not raw:
+    """Recursive credential-key guard that reports an actionable ``field_path``.
    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 item in raw:
+    for index, item in enumerate(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_project(raw: dict[str, Any]) -> ProjectRecord:
+def _parse_health(
    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 ValueError(f"project missing required fields: {', '.join(missing)}")
+        raise RegistryError(
            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 ValueError("profiles must be an object")
+        raise RegistryError(
            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 ValueError(f"profiles.{role} is required")
+            raise RegistryError(
                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 ValueError("workflow_paths must be a non-empty object")
+        raise RegistryError(
            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 ValueError("schema_paths must be an object when present")
+        raise RegistryError(
            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"]),
@@ -104,61 +389,211 @@ def _parse_project(raw: dict[str, Any]) -> ProjectRecord:
        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(raw.get("onboarding_checklist")),
+        onboarding_checklist=_parse_onboarding(
            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()
-    raw_text = source.read_text(encoding="utf-8")
+    try:
-    payload = json.loads(raw_text)
+        raw_text = source.read_text(encoding="utf-8")
    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 ValueError("registry root must be an object")
+        raise RegistryError(
            "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 != 1:
+    if version not in SUPPORTED_SCHEMA_VERSIONS:
-        raise ValueError(f"unsupported registry version: {version!r}")
+        supported = ", ".join(str(item) for item in SUPPORTED_SCHEMA_VERSIONS)
        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)
+    _reject_credential_keys(payload, source=source)
    projects_raw = payload.get("projects")
    if not isinstance(projects_raw, list) or not projects_raw:
-        raise ValueError("projects must be a non-empty array")
+        raise RegistryError(
            "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(_parse_project(item) for item in projects_raw)
+    projects = tuple(
-    return ProjectRegistry(version=version, projects=projects, source_path=source)
+        _parse_project(item, index=index, source=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."""
+    """Serialize a project for JSON API responses and HTML views.
    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]
@@ -1,34 +1,65 @@
-"""HTML views for project registry pages (#427)."""
+"""HTML views for project registry pages (#427, evolved for #635).
 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 ProjectRecord, ProjectRegistry
+from webui.project_registry import (
    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(project.id)}\">{_escape(project.repo_name)}</a></td>"
+            f"<td><a href=\"/projects/{_escape(dto['id'])}\">{_escape(dto['repo_name'])}</a></td>"
-            f"<td>{_escape(project.gitea_owner)}</td>"
+            f"<td>{_escape(dto['gitea_owner'])}</td>"
-            f"<td>{_escape(project.remote_host)}</td>"
+            f"<td>{_escape(dto['remote_host'])}</td>"
-            f"<td>{_escape(project.default_branch)}</td>"
+            f"<td>{_escape(dto['default_branch'])}</td>"
-            f"<td><code>{_escape(project.profiles['author'])}</code></td>"
+            f"<td><code>{_escape(dto['status'])}</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>Author profile</th>"
+        "<th>Branch</th><th>Status</th><th>Onboarding</th><th>Author profile</th>"
        "</tr></thead>"
        f"<tbody>{''.join(rows)}</tbody></table>"
    )
@@ -36,32 +67,39 @@ 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"(version {registry.version})</p>"
+        f"(schema version {registry.schema_version}, "
        f"API {_escape(registry.api_version)})</p>"
        f"{table}"
-        "<p><a href=\"/api/projects\">JSON API</a></p>"
+        "<p><a href=\"/api/v1/projects\">JSON API</a> "
        "(<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 project.profiles.items()
+        for role, name in dto["profiles"].items()
    )
    workflow_rows = "".join(
        f"<tr><th>{_escape(key)}</th><td><code>{_escape(path)}</code></td></tr>"
-        for key, path in project.workflow_paths.items()
+        for key, path in dto["workflow_paths"].items()
    )
    schema_rows = "".join(
        f"<tr><th>{_escape(key)}</th><td><code>{_escape(path)}</code></td></tr>"
-        for key, path in project.schema_paths.items()
+        for key, path in dto["schema_paths"].items()
    )
    checklist_items = []
-    for index, step in enumerate(project.onboarding_checklist, start=1):
+    for index, step in enumerate(dto["onboarding_checklist"], start=1):
        state_label = _STATE_LABELS.get(step["state"], step["state"])
        requirement = "required" if step["required"] else "optional"
        checklist_items.append(
-            "<li>"
+            f"<li class=\"step-{_escape(step['state'])}\">"
-            f"<strong>{index}. {_escape(step.title)}</strong>"
+            f"<strong>{index}. {_escape(step['title'])}</strong>"
-            f"<p>{_escape(step.description)}</p>"
+            f" <span class=\"badge\">{_escape(state_label)}</span>"
            f" <span class=\"meta\">({_escape(requirement)})</span>"
            f"<p>{_escape(step['description'])}</p>"
            "</li>"
        )
    checklist_html = (
@@ -69,16 +107,38 @@ 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(project.repo_name)}</h2>"
+        f"<h2>{_escape(dto['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(project.id)}</code></td></tr>"
+        f"<tr><th>Registry id</th><td><code>{_escape(dto['id'])}</code></td></tr>"
-        f"<tr><th>Gitea owner</th><td>{_escape(project.gitea_owner)}</td></tr>"
+        f"<tr><th>Status</th><td><code>{_escape(dto['status'])}</code></td></tr>"
-        f"<tr><th>Remote host</th><td>{_escape(project.remote_host)}</td></tr>"
+        f"<tr><th>Gitea owner</th><td>{_escape(dto['gitea_owner'])}</td></tr>"
-        f"<tr><th>Default branch</th><td><code>{_escape(project.default_branch)}</code></td></tr>"
+        f"<tr><th>Repository</th><td><code>{_escape(dto['repo_full_name'])}</code></td></tr>"
-        f"<tr><th>Local checkout</th><td><code>{_escape(project.local_checkout_path)}</code></td></tr>"
+        f"<tr><th>Remote name</th><td><code>{_escape(remote_name)}</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>"
@@ -86,8 +146,30 @@ 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 MVP — complete these steps outside the UI.</p>"
+        "<p class=\"meta\">Read-only — 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=project.repo_name, body_html=body)
+    return render_page(title=dto["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)
Author	SHA1	Message	Date
sysadminandClaude Opus 4.8	b7a63a5579	feat(webui): unified session/lease/lock/worktree inventory API (Closes #636 ) Co-Authored-By: Claude Opus 4.8 (1M context) <[email protected]>	2026-07-22 21:13:54 -04:00
sysadmin	53c2c92782	Merge pull request 'feat(webui): versioned project registry API (Closes #635 )' (#819 ) from feat/issue-635-project-registry-api into master	2026-07-22 20:04:26 -05:00
sysadmin	14c9c4d702	Merge pull request 'fix(mcp): forward worktree_path into publication preflight (Closes #815 )' (#817 ) from fix/issue-815-preflight-worktree-forwarding into master	2026-07-22 18:34:31 -05:00
jcwalker3andClaude Opus 4.8	95a5eb254f	fix(mcp): forward worktree_path into publication preflight (Closes #815 ) gitea_publish_unpublished_issue_branch accepted a required worktree_path but resolved it only after verify_preflight_purity had already run, so the #618 branches-only guard and every other workspace-resolution layer behind preflight received None and fell back to the MCP process root. A daemon rooted at the stable control checkout therefore refused a valid registered issue worktree the caller had explicitly supplied, before the publication assessor could use it — the sole verify_preflight_purity call site that accepted a worktree argument and dropped it. Resolve the workspace once, before preflight, and forward it. A blank or absent path forwards None and keeps the ordinary #618 fail-closed fallback, so guard strictness is unchanged for missing, empty, unregistered, foreign, or control-checkout worktrees. Public tool contract, ownership, cleanliness, hash, ancestry, and read-after-write protections from PR #814 are untouched. Adds tests/test_issue_815_preflight_worktree_forwarding.py: a #735-style capture proving the argument reaches verify_preflight_purity, a faithful production reproduction (control-rooted daemon, no session lock, explicit worktree) that clears #618 on the fixed source and is trapped at #618 on the unpatched source, an end-to-end control-rooted publication in the real topology (PROJECT_ROOT is the stable control checkout, the issue worktree is a distinct registered path, production guards forced on), and negative coverage keeping every #618 and assessor refusal intact. The prior #812 suite masked the defect by patching PROJECT_ROOT to equal the issue worktree. Co-Authored-By: Claude Opus 4.8 (1M context) <[email protected]>	2026-07-22 17:27:49 -05:00
sysadminandClaude Opus 4.8	b2f6e9a6dc	feat(webui): versioned project registry API (Closes #635 ) Evolve the MVP project registry (#427) into a versioned, fail-closed project registry API for the console (Phase 1, read-only). - Add schema version 2 with project `status`, per-step onboarding `state`/`required`, optional redacted `last_seen_health`, and `remote_name`. Version 1 files stay loadable and are normalized with explicit defaults. - Serve `/api/v1/projects` and `/api/v1/projects/{project_id}` with API provenance (`api_version`, `schema_version`, `source`). `/api/projects` is retained as an unversioned Phase 1 alias. - Replace bare `ValueError` with `RegistryError`, carrying an operator `remediation` and `field_path`; invalid registries fail closed as a 500 JSON payload or a dedicated HTML error page instead of a traceback. - Reject credential-shaped keys before any DTO is built, reusing `registry_safety.is_forbidden_key` as the single source of truth shared with the worker registry (#798). - Render HTML views from `project_to_dict`, so the console and the JSON API cannot disagree about status or onboarding progress. - Document the contract in docs/webui-project-registry-api.md. Tests: registry load/validate (valid, missing project, schema validation, credential rejection) and API route coverage. Co-Authored-By: Claude Opus 4.8 (1M context) <[email protected]>	2026-07-22 16:12:07 -04:00