fix: hide empty-state copy on queue credential fail-closed (Closes #458)

When fetch_error is set, render only the unavailable banner and an
explicit not-fetched inventory note. Suppress queue tables, pagination
proof, and "No open items." copy that implied a successful empty fetch.

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

docs/webui-local-dev.md

@@ -36,6 +36,8 @@ Optional environment variables:
 |------|-------------|
 | `/` | Home / operator overview |
 | `/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/{id}` | Project detail + onboarding checklist |
 | `/api/projects` | JSON registry export |
@@ -69,8 +71,19 @@ Prompts are generated at load time from canonical workflow files under
 copy/paste starters; canonical workflow files remain the only full policy
 source.
 
+## Live queue dashboard (#429)
+
+`/queue` loads open PRs and issues for the default registry project (seed:
+**Gitea-Tools** on `https://gitea.prgs.cc`) using existing `gitea_auth` read
+credentials. The UI surfaces pagination proof (returned count, pages fetched,
+`has_more`, `inventory_complete`) and classification badges (`claimed`,
+`blocked`, `in-review`, `duplicate`) when evidence exists.
+
+If credentials are missing or the fetch fails, the page shows an explicit error
+instead of an empty queue (fail closed).
+
 ## Tests
 
 ```bash
-pytest tests/test_webui_skeleton.py tests/test_webui_project_registry.py tests/test_webui_prompt_library.py -q
+pytest tests/test_webui_skeleton.py tests/test_webui_project_registry.py tests/test_webui_prompt_library.py tests/test_webui_queue_dashboard.py -q
 ```

gitea_mcp_server.py

@@ -440,10 +440,48 @@ def verify_preflight_purity(remote: str | None = None, worktree_path: str | None
             "Pre-flight order violation: Task capability (gitea_resolve_task_capability) has not been resolved (fail closed)"
         )
 
-    if worktree_path:
+    workspace = author_mutation_worktree.resolve_mutation_workspace(
-        dirty_files = sorted(_parse_porcelain_entries(_get_workspace_porcelain(worktree_path)))
+        worktree_path,
+        PROJECT_ROOT,
+        active_worktree_env=os.environ.get(ACTIVE_WORKTREE_ENV),
+        author_worktree_env=os.environ.get(AUTHOR_WORKTREE_ENV),
+    )
+    real_workspace = os.path.realpath(workspace)
+    real_root = os.path.realpath(PROJECT_ROOT)
+
+    if real_workspace != real_root:
+        if not _preflight_in_test_mode():
+            if not os.path.exists(real_workspace):
+                raise RuntimeError(
+                    f"Branches-only mutation guard (#274): worktree path '{workspace}' does not exist (fail closed)"
+                )
+            if not os.path.isdir(real_workspace):
+                raise RuntimeError(
+                    f"Branches-only mutation guard (#274): worktree path '{workspace}' is not a directory (fail closed)"
+                )
+            try:
+                res = subprocess.run(
+                    ["git", "-C", real_workspace, "rev-parse", "--git-common-dir"],
+                    capture_output=True,
+                    text=True,
+                    check=True,
+                )
+                common_dir = os.path.realpath(res.stdout.strip())
+                expected_dir = os.path.realpath(os.path.join(real_root, ".git"))
+                if common_dir != expected_dir:
+                    raise RuntimeError(
+                        f"Branches-only mutation guard (#274): worktree '{workspace}' does not belong to the target repository '{PROJECT_ROOT}' (fail closed)"
+                    )
+            except Exception as e:
+                if isinstance(e, RuntimeError):
+                    raise e
+                raise RuntimeError(
+                    f"Branches-only mutation guard (#274): worktree '{workspace}' is not a valid git repository (fail closed)"
+                )
+
+        dirty_files = sorted(_parse_porcelain_entries(_get_workspace_porcelain(workspace)))
         if dirty_files:
-            details = _preflight_workspace_details(worktree_path, dirty_files)
+            details = _preflight_workspace_details(workspace, dirty_files)
             raise RuntimeError(
                 "Pre-flight order violation: Active task workspace has tracked "
                 "file edits before mutation (fail closed). "
@@ -972,6 +1010,7 @@ def gitea_create_issue(
     repo: str | None = None,
     allow_duplicate_override: bool = False,
     split_from_issue: int | None = None,
+    worktree_path: str | None = None,
 ) -> dict:
     """Create a new issue on a Gitea repository.
 
@@ -984,6 +1023,7 @@ def gitea_create_issue(
         repo: Override the repository name.
         allow_duplicate_override: Operator-approved split after duplicate found.
         split_from_issue: Existing duplicate issue number when overriding.
+        worktree_path: Optional path to verify branches-only guard.
 
     Returns:
         dict with 'number' of the created issue ('url' only with the reveal opt-in).
@@ -1010,7 +1050,7 @@ def gitea_create_issue(
     )
     if blocked:
         return blocked
-    verify_preflight_purity(remote)
+    verify_preflight_purity(remote, worktree_path=worktree_path)
     base = repo_api_url(h, o, r)
     open_issues = api_get_all(f"{base}/issues?state=open&type=issues", auth)
     closed_issues = api_get_all(

tests/test_create_issue_workspace_guard.py

@@ -0,0 +1,145 @@
+import os
+import sys
+import unittest
+from pathlib import Path
+from unittest.mock import patch, MagicMock
+
+# Ensure we import from the repo root
+sys.path.insert(0, str(Path(__file__).resolve().parent.parent))
+
+import gitea_mcp_server as srv
+
+FAKE_AUTH = {"Authorization": "token test-token"}
+# Stable control checkout (parent of branches/), not the MCP server worktree root.
+CONTROL_CHECKOUT_ROOT = str(Path(__file__).resolve().parents[3])
+PROJECT_ROOT = srv.PROJECT_ROOT
+
+
+class TestCreateIssueWorkspaceGuard(unittest.TestCase):
+
+    def setUp(self):
+        # Reset preflight flags
+        srv._preflight_whoami_called = True
+        srv._preflight_capability_called = True
+        srv._preflight_resolved_role = "author"
+        srv._preflight_whoami_violation = False
+        srv._preflight_capability_violation = False
+        
+        # Disable early return in verify_preflight_purity for testing
+        self._orig_in_test = srv._preflight_in_test_mode
+        srv._preflight_in_test_mode = lambda: False
+
+    def tearDown(self):
+        srv._preflight_in_test_mode = self._orig_in_test
+
+    @patch("gitea_mcp_server._auth", return_value=FAKE_AUTH)
+    @patch("gitea_mcp_server._profile_permission_block", return_value=None)
+    @patch("gitea_mcp_server._namespace_mutation_block", return_value=None)
+    @patch("gitea_mcp_server.role_session_router.check_author_mutation_after_reviewer_stop", return_value=(True, []))
+    @patch("gitea_mcp_server.api_request")
+    @patch("gitea_mcp_server.api_get_all", return_value=[])
+    @patch("gitea_mcp_server.issue_lock_worktree.read_worktree_git_state", return_value={"current_branch": "feat/issue-1"})
+    def test_create_issue_stable_checkout_rejected(self, _git, _get_all, mock_api, _role, _ns, _prof, _auth):
+        # Without worktree_path/env hints, workspace resolves to PROJECT_ROOT. When that
+        # path is the stable control checkout (not under branches/), mutation must fail.
+        with patch.object(srv, "PROJECT_ROOT", CONTROL_CHECKOUT_ROOT):
+            with self.assertRaises(RuntimeError) as ctx:
+                srv.gitea_create_issue(title="Test issue", body="body text")
+        self.assertIn("stable control checkout", str(ctx.exception))
+
+    @patch("gitea_mcp_server._auth", return_value=FAKE_AUTH)
+    @patch("gitea_mcp_server._profile_permission_block", return_value=None)
+    @patch("gitea_mcp_server._namespace_mutation_block", return_value=None)
+    @patch("gitea_mcp_server.role_session_router.check_author_mutation_after_reviewer_stop", return_value=(True, []))
+    @patch("gitea_mcp_server.api_request")
+    @patch("gitea_mcp_server.api_get_all", return_value=[])
+    @patch("gitea_mcp_server.issue_lock_worktree.read_worktree_git_state", return_value={"current_branch": "feat/issue-1"})
+    @patch("os.path.exists", return_value=True)
+    @patch("os.path.isdir", return_value=True)
+    @patch("subprocess.run")
+    def test_create_issue_valid_worktree_succeeds(self, mock_run, mock_isdir, mock_exists, _git, _get_all, mock_api, _role, _ns, _prof, _auth):
+        # Mock subprocess.run for git --git-common-dir to return PROJECT_ROOT/.git
+        mock_res = MagicMock()
+        mock_res.stdout = f"{CONTROL_CHECKOUT_ROOT}/.git\n"
+        mock_run.return_value = mock_res
+
+        mock_api.return_value = {"number": 42, "html_url": "https://gitea.example.com/issues/42"}
+
+        # Provide a valid branches path under the control checkout root
+        valid_path = os.path.join(CONTROL_CHECKOUT_ROOT, "branches", "feat-issue-1")
+        
+        with patch.object(srv, "PROJECT_ROOT", CONTROL_CHECKOUT_ROOT):
+            with patch("gitea_mcp_server._get_workspace_porcelain", return_value=""):
+                res = srv.gitea_create_issue(
+                    title="Test issue", body="body", worktree_path=valid_path
+                )
+            
+        self.assertEqual(res["number"], 42)
+        mock_api.assert_called_once()
+
+    @patch("gitea_mcp_server._auth", return_value=FAKE_AUTH)
+    @patch("gitea_mcp_server._profile_permission_block", return_value=None)
+    @patch("gitea_mcp_server._namespace_mutation_block", return_value=None)
+    @patch("gitea_mcp_server.role_session_router.check_author_mutation_after_reviewer_stop", return_value=(True, []))
+    @patch("gitea_mcp_server.api_request")
+    @patch("gitea_mcp_server.api_get_all", return_value=[])
+    @patch("gitea_mcp_server.issue_lock_worktree.read_worktree_git_state", return_value={"current_branch": "feat/issue-1"})
+    def test_create_issue_missing_worktree_fails_closed(self, _git, _get_all, mock_api, _role, _ns, _prof, _auth):
+        # Path under branches/ but doesn't exist
+        missing_path = os.path.join(
+            CONTROL_CHECKOUT_ROOT, "branches", "nonexistent-worktree-path-999"
+        )
+        
+        with patch.object(srv, "PROJECT_ROOT", CONTROL_CHECKOUT_ROOT):
+            with self.assertRaises(RuntimeError) as ctx:
+                srv.gitea_create_issue(
+                    title="Test issue", body="body", worktree_path=missing_path
+                )
+        self.assertIn("does not exist (fail closed)", str(ctx.exception))
+
+    @patch("gitea_mcp_server._auth", return_value=FAKE_AUTH)
+    @patch("gitea_mcp_server._profile_permission_block", return_value=None)
+    @patch("gitea_mcp_server._namespace_mutation_block", return_value=None)
+    @patch("gitea_mcp_server.role_session_router.check_author_mutation_after_reviewer_stop", return_value=(True, []))
+    @patch("gitea_mcp_server.api_request")
+    @patch("gitea_mcp_server.api_get_all", return_value=[])
+    @patch("gitea_mcp_server.issue_lock_worktree.read_worktree_git_state", return_value={"current_branch": "feat/issue-1"})
+    @patch("os.path.exists", return_value=True)
+    @patch("os.path.isdir", return_value=True)
+    @patch("subprocess.run")
+    def test_create_issue_wrong_repo_fails_closed(self, mock_run, mock_isdir, mock_exists, _git, _get_all, mock_api, _role, _ns, _prof, _auth):
+        # Mock subprocess.run for git --git-common-dir to return a different path
+        mock_res = MagicMock()
+        mock_res.stdout = "/Users/jasonwalker/Development/some-other-repo/.git\n"
+        mock_run.return_value = mock_res
+
+        wrong_repo_path = os.path.join(CONTROL_CHECKOUT_ROOT, "branches", "feat-issue-1")
+
+        with patch.object(srv, "PROJECT_ROOT", CONTROL_CHECKOUT_ROOT):
+            with self.assertRaises(RuntimeError) as ctx:
+                srv.gitea_create_issue(
+                    title="Test issue", body="body", worktree_path=wrong_repo_path
+                )
+        self.assertIn("does not belong to the target repository", str(ctx.exception))
+
+    @patch("gitea_mcp_server._auth", return_value=FAKE_AUTH)
+    @patch("gitea_mcp_server._profile_permission_block", return_value=None)
+    @patch("gitea_mcp_server._namespace_mutation_block", return_value=None)
+    def test_create_issue_fails_without_whoami_preflight(self, _ns, _prof, _auth):
+        srv._preflight_whoami_called = False
+        with self.assertRaises(RuntimeError) as ctx:
+            srv.gitea_create_issue(title="Test issue", body="body")
+        self.assertIn("Identity (gitea_whoami) has not been verified", str(ctx.exception))
+
+    @patch("gitea_mcp_server._auth", return_value=FAKE_AUTH)
+    @patch("gitea_mcp_server._profile_permission_block", return_value=None)
+    @patch("gitea_mcp_server._namespace_mutation_block", return_value=None)
+    def test_create_issue_fails_without_capability_preflight(self, _ns, _prof, _auth):
+        srv._preflight_capability_called = False
+        with self.assertRaises(RuntimeError) as ctx:
+            srv.gitea_create_issue(title="Test issue", body="body")
+        self.assertIn("Task capability (gitea_resolve_task_capability) has not been resolved", str(ctx.exception))
+
+
+if __name__ == "__main__":
+    unittest.main()

tests/test_webui_queue_dashboard.py

@@ -0,0 +1,245 @@
+"""Tests for web UI live queue dashboard (#429)."""
+import sys
+import unittest
+from datetime import datetime, timedelta, timezone
+from pathlib import Path
+from unittest import mock
+
+sys.path.insert(0, str(Path(__file__).resolve().parent.parent))
+
+from starlette.testclient import TestClient
+
+from webui.app import create_app
+from webui.queue_loader import (
+    PaginationMeta,
+    _classify_issue,
+    _classify_pr,
+    _extract_linked_issue,
+    load_queue_snapshot,
+    snapshot_to_dict,
+)
+
+_RECENT = datetime.now(timezone.utc).isoformat()
+_STALE = (datetime.now(timezone.utc) - timedelta(days=30)).isoformat()
+
+_SAMPLE_PRS = [
+    {
+        "number": 100,
+        "title": "feat: queue dashboard (Closes #429)",
+        "body": "",
+        "mergeable": True,
+        "updated_at": _RECENT,
+        "head": {"ref": "feat/issue-429", "sha": "abc123def456"},
+        "base": {"ref": "master"},
+    },
+    {
+        "number": 99,
+        "title": "fix: conflict",
+        "body": "Closes #50",
+        "mergeable": False,
+        "updated_at": _STALE,
+        "head": {"ref": "fix/issue-50", "sha": "deadbeef0001"},
+        "base": {"ref": "master"},
+    },
+    {
+        "number": 98,
+        "title": "feat: duplicate A (Closes #60)",
+        "body": "",
+        "mergeable": True,
+        "updated_at": _RECENT,
+        "head": {"ref": "feat/issue-60-a", "sha": "111111111111"},
+        "base": {"ref": "master"},
+    },
+    {
+        "number": 97,
+        "title": "feat: duplicate B (Closes #60)",
+        "body": "",
+        "mergeable": True,
+        "updated_at": _RECENT,
+        "head": {"ref": "feat/issue-60-b", "sha": "222222222222"},
+        "base": {"ref": "master"},
+    },
+]
+
+_SAMPLE_ISSUES = [
+    {
+        "number": 429,
+        "title": "Web UI queue dashboard",
+        "state": "open",
+        "labels": [{"name": "status:in-progress"}],
+        "assignee": None,
+        "updated_at": _RECENT,
+    },
+    {
+        "number": 60,
+        "title": "Duplicate PR target",
+        "state": "open",
+        "labels": [],
+        "assignee": None,
+        "updated_at": _RECENT,
+    },
+    {
+        "number": 50,
+        "title": "Blocked PR target",
+        "state": "open",
+        "labels": [],
+        "assignee": None,
+        "updated_at": _STALE,
+    },
+]
+
+
+def _mock_pagination(count: int) -> PaginationMeta:
+    return PaginationMeta(
+        page=1,
+        per_page=50,
+        returned_count=count,
+        has_more=False,
+        is_final_page=True,
+        inventory_complete=True,
+        pages_fetched=1,
+    )
+
+
+def _mock_fetch_prs(*_args, **_kwargs):
+    return list(_SAMPLE_PRS), _mock_pagination(len(_SAMPLE_PRS))
+
+
+def _mock_fetch_issues(*_args, **_kwargs):
+    return list(_SAMPLE_ISSUES), _mock_pagination(len(_SAMPLE_ISSUES))
+
+
+class TestQueueClassification(unittest.TestCase):
+    def test_extract_linked_issue_from_title(self):
+        self.assertEqual(
+            _extract_linked_issue("feat: X (Closes #429)", ""),
+            429,
+        )
+
+    def test_classify_pr_blocked_and_stale(self):
+        pr = _SAMPLE_PRS[1]
+        badges = _classify_pr(
+            pr,
+            issue_claimed=set(),
+            issue_to_prs={50: [99]},
+        )
+        self.assertIn("blocked", badges)
+        self.assertIn("stale", badges)
+
+    def test_classify_pr_duplicate(self):
+        badges = _classify_pr(
+            _SAMPLE_PRS[2],
+            issue_claimed=set(),
+            issue_to_prs={60: [98, 97]},
+        )
+        self.assertIn("duplicate", badges)
+        self.assertIn("in-review", badges)
+
+    def test_classify_issue_claimed_and_duplicate(self):
+        claimed = _classify_issue(_SAMPLE_ISSUES[0], linked_prs=[])
+        self.assertIn("claimed", claimed)
+        duplicate = _classify_issue(_SAMPLE_ISSUES[1], linked_prs=[98, 97])
+        self.assertIn("duplicate", duplicate)
+
+
+class TestQueueLoader(unittest.TestCase):
+    def test_snapshot_with_mock_fetch(self):
+        snapshot = load_queue_snapshot(
+            fetch_prs=_mock_fetch_prs,
+            fetch_issues=_mock_fetch_issues,
+        )
+        self.assertEqual(snapshot.project_id, "gitea-tools")
+        self.assertIsNone(snapshot.fetch_error)
+        self.assertEqual(len(snapshot.prs), 4)
+        self.assertEqual(len(snapshot.issues), 3)
+        self.assertTrue(snapshot.pr_pagination.inventory_complete)
+        self.assertTrue(snapshot.issue_pagination.inventory_complete)
+
+        pr100 = next(p for p in snapshot.prs if p.number == 100)
+        self.assertEqual(pr100.extra["linked_issue"], "429")
+        self.assertIn("claimed", pr100.badges)
+
+    def test_snapshot_dict_export(self):
+        snapshot = load_queue_snapshot(
+            fetch_prs=_mock_fetch_prs,
+            fetch_issues=_mock_fetch_issues,
+        )
+        data = snapshot_to_dict(snapshot)
+        self.assertEqual(data["project_id"], "gitea-tools")
+        self.assertIsNone(data["fetch_error"])
+        self.assertEqual(len(data["prs"]), 4)
+        self.assertTrue(data["pagination"]["prs"]["inventory_complete"])
+
+
+class TestQueueRoutes(unittest.TestCase):
+    def setUp(self):
+        self.client = TestClient(create_app())
+        self._patch = mock.patch(
+            "webui.app.load_queue_snapshot",
+            return_value=load_queue_snapshot(
+                fetch_prs=_mock_fetch_prs,
+                fetch_issues=_mock_fetch_issues,
+            ),
+        )
+        self._patch.start()
+
+    def tearDown(self):
+        self._patch.stop()
+
+    def test_queue_page_renders_tables_and_pagination(self):
+        response = self.client.get("/queue")
+        self.assertEqual(response.status_code, 200)
+        self.assertIn("Live queue", response.text)
+        self.assertIn("Open pull requests", response.text)
+        self.assertIn("Open issues", response.text)
+        self.assertIn("pages_fetched", response.text)
+        self.assertIn("Gitea-Tools", response.text)
+        self.assertNotIn("child issue", response.text.lower())
+
+    def test_api_queue_json(self):
+        response = self.client.get("/api/queue")
+        self.assertEqual(response.status_code, 200)
+        data = response.json()
+        self.assertEqual(data["repo"], "Scaled-Tech-Consulting/Gitea-Tools")
+        self.assertEqual(data["pagination"]["issues"]["returned_count"], 3)
+
+    def test_queue_fail_closed_without_credentials(self):
+        with mock.patch("webui.queue_loader.get_auth_header", return_value=None):
+            snapshot = load_queue_snapshot()
+        self.assertIsNotNone(snapshot.fetch_error)
+        self.assertEqual(len(snapshot.prs), 0)
+
+    def test_queue_fail_closed_hides_empty_state_copy(self):
+        with mock.patch("webui.queue_loader.get_auth_header", return_value=None):
+            snapshot = load_queue_snapshot()
+        with mock.patch("webui.app.load_queue_snapshot", return_value=snapshot):
+            response = self.client.get("/queue")
+        self.assertEqual(response.status_code, 200)
+        self.assertIn("Queue unavailable", response.text)
+        self.assertIn("not fetched", response.text.lower())
+        self.assertNotIn("No open items.", response.text)
+        self.assertNotIn("Open pull requests", response.text)
+        self.assertNotIn("pages_fetched", response.text)
+
+    def test_queue_successful_empty_inventory_shows_empty_state(self):
+        def _empty_prs(*_args, **_kwargs):
+            return [], _mock_pagination(0)
+
+        def _empty_issues(*_args, **_kwargs):
+            return [], _mock_pagination(0)
+
+        snapshot = load_queue_snapshot(
+            fetch_prs=_empty_prs,
+            fetch_issues=_empty_issues,
+        )
+        with mock.patch("webui.app.load_queue_snapshot", return_value=snapshot):
+            response = self.client.get("/queue")
+        self.assertEqual(response.status_code, 200)
+        self.assertIsNone(snapshot.fetch_error)
+        self.assertEqual(response.text.count("No open items."), 2)
+        self.assertIn("pages_fetched", response.text)
+        self.assertIn("pagination (complete)", response.text)
+
+
+if __name__ == "__main__":
+    unittest.main()

tests/test_webui_skeleton.py

@@ -56,11 +56,16 @@ class TestWebuiSkeleton(unittest.TestCase):
         self.assertEqual(response.status_code, 405)
         self.assertEqual(response.json()["error"], "read-only-mvp")
 
+    def test_queue_route_renders(self):
+        response = self.client.get("/queue")
+        self.assertEqual(response.status_code, 200)
+        self.assertIn("Live queue", response.text)
+
     def test_nav_links_on_all_pages(self):
-        for path in ("/", "/projects", "/prompts", "/runtime", "/audit"):
+        for path in ("/", "/queue", "/projects", "/prompts", "/runtime", "/audit"):
             with self.subTest(path=path):
                 text = self.client.get(path).text
-                for href in ("/projects", "/prompts", "/runtime", "/audit"):
+                for href in ("/queue", "/projects", "/prompts", "/runtime", "/audit"):
                     self.assertIn(f'href="{href}"', text)
 
 

webui/app.py

@@ -14,6 +14,8 @@ from webui.project_registry import find_project, load_registry, registry_to_dict
 from webui.project_views import render_project_detail, render_projects_list
 from webui.prompt_library import find_prompt, library_to_dict
 from webui.prompt_views import render_prompt_detail, render_prompts_page
+from webui.queue_loader import load_queue_snapshot, snapshot_to_dict
+from webui.queue_views import render_queue_page
 
 _READ_ONLY_METHODS = frozenset({"GET", "HEAD", "OPTIONS"})
 
@@ -32,6 +34,7 @@ async def home(_request: Request) -> HTMLResponse:
         "<h2>Operator console</h2>"
         "<p>Local entry point for MCP Control Plane operational views.</p>"
         "<ul>"
+        "<li><strong>Queue</strong> — live PR and issue dashboard (#429)</li>"
         "<li><strong>Projects</strong> — registry and onboarding (#427)</li>"
         "<li><strong>Prompts</strong> — canonical workflow prompt library (#428)</li>"
         "<li><strong>Runtime</strong> — MCP health and stale-runtime detection (#430)</li>"
@@ -52,6 +55,15 @@ async def health(_request: Request) -> JSONResponse:
     })
 
 
+async def queue(_request: Request) -> HTMLResponse:
+    snapshot = load_queue_snapshot()
+    return HTMLResponse(render_page(title="Queue", body_html=render_queue_page(snapshot)))
+
+
+async def api_queue(_request: Request) -> JSONResponse:
+    return JSONResponse(snapshot_to_dict(load_queue_snapshot()))
+
+
 async def projects(_request: Request) -> HTMLResponse:
     registry = load_registry()
     return HTMLResponse(render_projects_list(registry))
@@ -151,6 +163,8 @@ def create_app() -> Starlette:
         routes=[
             Route("/", home, methods=["GET"]),
             Route("/health", health, methods=["GET"]),
+            Route("/queue", queue, methods=["GET"]),
+            Route("/api/queue", api_queue, methods=["GET"]),
             Route("/projects", projects, methods=["GET"]),
             Route("/projects/{project_id}", project_detail, methods=["GET"]),
             Route("/api/projects", api_projects, methods=["GET"]),

webui/layout.py

@@ -4,6 +4,7 @@ from __future__ import annotations
 
 NAV_ITEMS = (
     ("/", "Home"),
+    ("/queue", "Queue"),
     ("/projects", "Projects"),
     ("/prompts", "Prompts"),
     ("/runtime", "Runtime"),
@@ -145,6 +146,20 @@ def render_page(*, title: str, body_html: str, extra_head: str = "") -> str:
     }}
     .copy-btn:hover {{ filter: brightness(1.08); }}
     .muted {{ color: var(--muted); }}
+    .badges {{ display: inline-flex; flex-wrap: wrap; gap: 0.35rem; margin-left: 0.5rem; }}
+    .badge {{
+      font-size: 0.72rem;
+      padding: 0.1rem 0.45rem;
+      border-radius: 999px;
+      border: 1px solid var(--border);
+      color: var(--muted);
+      text-transform: lowercase;
+    }}
+    .badge-claimed {{ color: #8fd19e; border-color: #3d6b4a; }}
+    .badge-blocked {{ color: #f0a8a8; border-color: #7a3b3b; }}
+    .badge-in-review {{ color: #9ec8f0; border-color: #3d5f7a; }}
+    .badge-duplicate {{ color: #e0c27a; border-color: #6b5730; }}
+    .badge-stale {{ color: #c9b8e8; border-color: #5a4a78; }}
   </style>
   {extra_head}
 </head>

webui/queue_loader.py

@@ -0,0 +1,385 @@
+"""Load live Gitea PR/issue queue state for the web UI dashboard (#429)."""
+
+from __future__ import annotations
+
+import re
+from dataclasses import dataclass
+from datetime import datetime, timezone
+from typing import Any, Callable
+from urllib.parse import urlparse
+
+from gitea_auth import api_fetch_page, get_auth_header, repo_api_url
+
+from webui.project_registry import ProjectRecord, load_registry
+
+_CLOSES_RE = re.compile(r"(?:closes|fixes|resolves)\s+#(\d+)", re.I)
+_ISSUE_REF_RE = re.compile(r"#(\d+)")
+_STALE_DAYS = 14
+
+
+@dataclass(frozen=True)
+class PaginationMeta:
+    page: int
+    per_page: int
+    returned_count: int
+    has_more: bool
+    is_final_page: bool
+    inventory_complete: bool
+    pages_fetched: int
+
+
+@dataclass(frozen=True)
+class QueueItem:
+    number: int
+    title: str
+    badges: tuple[str, ...]
+    extra: dict[str, str]
+
+
+@dataclass(frozen=True)
+class QueueSnapshot:
+    project_id: str
+    repo_label: str
+    prs: tuple[QueueItem, ...]
+    issues: tuple[QueueItem, ...]
+    pr_pagination: PaginationMeta | None
+    issue_pagination: PaginationMeta | None
+    fetch_error: str | None = None
+
+
+def _host_from_url(remote_host: str) -> str:
+    parsed = urlparse(remote_host.strip())
+    return parsed.netloc or remote_host.strip().rstrip("/")
+
+
+def _extract_linked_issue(title: str | None, body: str | None = None) -> int | None:
+    for text in (title, body):
+        if not text:
+            continue
+        match = _CLOSES_RE.search(text)
+        if match:
+            return int(match.group(1))
+    if body:
+        refs = _ISSUE_REF_RE.findall(body)
+        if refs:
+            return int(refs[0])
+    return None
+
+
+def _is_stale(updated_at: str | None) -> bool:
+    if not updated_at:
+        return False
+    try:
+        normalized = updated_at.replace("Z", "+00:00")
+        parsed = datetime.fromisoformat(normalized)
+        if parsed.tzinfo is None:
+            parsed = parsed.replace(tzinfo=timezone.utc)
+        age = datetime.now(timezone.utc) - parsed.astimezone(timezone.utc)
+        return age.days >= _STALE_DAYS
+    except ValueError:
+        return False
+
+
+def _classify_pr(
+    pr: dict,
+    *,
+    issue_claimed: set[int],
+    issue_to_prs: dict[int, list[int]],
+) -> tuple[str, ...]:
+    badges: list[str] = []
+    mergeable = pr.get("mergeable")
+    if mergeable is False:
+        badges.append("blocked")
+    linked = _extract_linked_issue(pr.get("title"), pr.get("body"))
+    if linked is not None:
+        if linked in issue_claimed:
+            badges.append("claimed")
+        if len(issue_to_prs.get(linked, [])) > 1:
+            badges.append("duplicate")
+    if _is_stale(pr.get("updated_at")):
+        badges.append("stale")
+    if mergeable is True and "blocked" not in badges:
+        badges.append("in-review")
+    if not badges:
+        badges.append("open")
+    return tuple(dict.fromkeys(badges))
+
+
+def _classify_issue(
+    issue: dict,
+    *,
+    linked_prs: list[int],
+) -> tuple[str, ...]:
+    badges: list[str] = []
+    labels = [lb.get("name", "") for lb in issue.get("labels", [])]
+    if "status:in-progress" in labels:
+        badges.append("claimed")
+    if len(linked_prs) > 1:
+        badges.append("duplicate")
+    elif linked_prs and "claimed" not in badges:
+        badges.append("in-review")
+    if _is_stale(issue.get("updated_at")):
+        badges.append("stale")
+    if not badges:
+        badges.append("open")
+    return tuple(dict.fromkeys(badges))
+
+
+def _format_pr_item(pr: dict, badges: tuple[str, ...]) -> QueueItem:
+    head = pr.get("head") or {}
+    base = pr.get("base") or {}
+    mergeable = pr.get("mergeable")
+    merge_label = (
+        "mergeable" if mergeable is True else "conflicted" if mergeable is False else "unknown"
+    )
+    linked = _extract_linked_issue(pr.get("title"), pr.get("body"))
+    return QueueItem(
+        number=int(pr["number"]),
+        title=str(pr.get("title") or ""),
+        badges=badges,
+        extra={
+            "branch": f"{head.get('ref', '?')} → {base.get('ref', '?')}",
+            "head_sha": str(head.get("sha") or "")[:12],
+            "mergeable": merge_label,
+            "linked_issue": str(linked) if linked is not None else "",
+        },
+    )
+
+
+def _format_issue_item(issue: dict, badges: tuple[str, ...]) -> QueueItem:
+    labels = ", ".join(lb.get("name", "") for lb in issue.get("labels", []))
+    assignee = (issue.get("assignee") or {}).get("login", "")
+    return QueueItem(
+        number=int(issue["number"]),
+        title=str(issue.get("title") or ""),
+        badges=badges,
+        extra={
+            "labels": labels or "—",
+            "assignee": assignee or "unassigned",
+            "state": str(issue.get("state") or ""),
+        },
+    )
+
+
+def _pagination_from_pages(
+    *,
+    per_page: int,
+    pages_fetched: int,
+    returned_count: int,
+    is_final_page: bool,
+) -> PaginationMeta:
+    return PaginationMeta(
+        page=1,
+        per_page=per_page,
+        returned_count=returned_count,
+        has_more=not is_final_page,
+        is_final_page=is_final_page,
+        inventory_complete=is_final_page,
+        pages_fetched=pages_fetched,
+    )
+
+
+def _fetch_prs(
+    host: str,
+    org: str,
+    repo: str,
+    auth: str,
+    *,
+    per_page: int = 50,
+) -> tuple[list[dict], PaginationMeta]:
+    url = f"{repo_api_url(host, org, repo)}/pulls?state=open"
+    all_raw: list[dict] = []
+    pages_fetched = 0
+    is_final = False
+    page = 1
+    while pages_fetched < 20:
+        raw_page, meta = api_fetch_page(url, auth, page=page, limit=per_page)
+        pages_fetched += 1
+        all_raw.extend(raw_page)
+        is_final = bool(meta["is_final_page"])
+        if is_final:
+            break
+        page += 1
+    pagination = _pagination_from_pages(
+        per_page=per_page,
+        pages_fetched=pages_fetched,
+        returned_count=len(all_raw),
+        is_final_page=is_final,
+    )
+    return all_raw, pagination
+
+
+def _fetch_issues(
+    host: str,
+    org: str,
+    repo: str,
+    auth: str,
+    *,
+    per_page: int = 50,
+) -> tuple[list[dict], PaginationMeta]:
+    url = f"{repo_api_url(host, org, repo)}/issues?state=open&type=issues"
+    all_raw: list[dict] = []
+    page = 1
+    pages_fetched = 0
+    is_final = False
+    while pages_fetched < 20:
+        raw_page, meta = api_fetch_page(url, auth, page=page, limit=per_page)
+        pages_fetched += 1
+        all_raw.extend(raw_page)
+        is_final = bool(meta["is_final_page"])
+        if is_final:
+            break
+        page += 1
+    pagination = _pagination_from_pages(
+        per_page=per_page,
+        pages_fetched=pages_fetched,
+        returned_count=len(all_raw),
+        is_final_page=is_final,
+    )
+    return all_raw, pagination
+
+
+def load_queue_snapshot(
+    project_id: str | None = None,
+    *,
+    fetch_prs: Callable[..., tuple[list[dict], PaginationMeta]] | None = None,
+    fetch_issues: Callable[..., tuple[list[dict], PaginationMeta]] | None = None,
+) -> QueueSnapshot:
+    """Load open PR and issue queues for a registry project (default: first entry)."""
+    registry = load_registry()
+    project: ProjectRecord | None = None
+    if project_id:
+        for entry in registry.projects:
+            if entry.id == project_id:
+                project = entry
+                break
+    else:
+        project = registry.projects[0] if registry.projects else None
+
+    if project is None:
+        return QueueSnapshot(
+            project_id=project_id or "",
+            repo_label="",
+            prs=(),
+            issues=(),
+            pr_pagination=None,
+            issue_pagination=None,
+            fetch_error="project not found in registry",
+        )
+
+    host = _host_from_url(project.remote_host)
+    pr_fetch = fetch_prs or _fetch_prs
+    issue_fetch = fetch_issues or _fetch_issues
+    using_live_fetch = fetch_prs is None or fetch_issues is None
+    auth = get_auth_header(host) if using_live_fetch else "test-auth"
+    if using_live_fetch and not auth:
+        return QueueSnapshot(
+            project_id=project.id,
+            repo_label=f"{project.gitea_owner}/{project.repo_name}",
+            prs=(),
+            issues=(),
+            pr_pagination=None,
+            issue_pagination=None,
+            fetch_error=(
+                f"Gitea credentials unavailable for {host}; "
+                "queue cannot be loaded (fail closed — not showing empty queue)"
+            ),
+        )
+
+    try:
+        raw_prs, pr_pagination = pr_fetch(
+            host, project.gitea_owner, project.repo_name, auth
+        )
+        raw_issues, issue_pagination = issue_fetch(
+            host, project.gitea_owner, project.repo_name, auth
+        )
+    except Exception as exc:  # noqa: BLE001 — surface operator-visible fetch errors
+        return QueueSnapshot(
+            project_id=project.id,
+            repo_label=f"{project.gitea_owner}/{project.repo_name}",
+            prs=(),
+            issues=(),
+            pr_pagination=None,
+            issue_pagination=None,
+            fetch_error=f"Gitea fetch failed: {exc}",
+        )
+
+    issue_to_prs: dict[int, list[int]] = {}
+    for pr in raw_prs:
+        linked = _extract_linked_issue(pr.get("title"), pr.get("body"))
+        if linked is not None:
+            issue_to_prs.setdefault(linked, []).append(int(pr["number"]))
+
+    issue_claimed = {
+        int(i["number"])
+        for i in raw_issues
+        if any(lb.get("name") == "status:in-progress" for lb in i.get("labels", []))
+    }
+
+    pr_items = tuple(
+        _format_pr_item(
+            pr,
+            _classify_pr(
+                pr,
+                issue_claimed=issue_claimed,
+                issue_to_prs=issue_to_prs,
+            ),
+        )
+        for pr in sorted(raw_prs, key=lambda p: int(p["number"]), reverse=True)
+    )
+    issue_items = tuple(
+        _format_issue_item(
+            issue,
+            _classify_issue(
+                issue,
+                linked_prs=issue_to_prs.get(int(issue["number"]), []),
+            ),
+        )
+        for issue in sorted(raw_issues, key=lambda i: int(i["number"]), reverse=True)
+    )
+
+    return QueueSnapshot(
+        project_id=project.id,
+        repo_label=f"{project.gitea_owner}/{project.repo_name}",
+        prs=pr_items,
+        issues=issue_items,
+        pr_pagination=pr_pagination,
+        issue_pagination=issue_pagination,
+    )
+
+
+def snapshot_to_dict(snapshot: QueueSnapshot) -> dict[str, Any]:
+    """JSON-serializable export for /api/queue."""
+
+    def _page(meta: PaginationMeta | None) -> dict[str, Any] | None:
+        if meta is None:
+            return None
+        return {
+            "page": meta.page,
+            "per_page": meta.per_page,
+            "returned_count": meta.returned_count,
+            "has_more": meta.has_more,
+            "is_final_page": meta.is_final_page,
+            "inventory_complete": meta.inventory_complete,
+            "pages_fetched": meta.pages_fetched,
+        }
+
+    def _item(item: QueueItem) -> dict[str, Any]:
+        return {
+            "number": item.number,
+            "title": item.title,
+            "badges": list(item.badges),
+            **item.extra,
+        }
+
+    return {
+        "project_id": snapshot.project_id,
+        "repo": snapshot.repo_label,
+        "fetch_error": snapshot.fetch_error,
+        "prs": [_item(p) for p in snapshot.prs],
+        "issues": [_item(i) for i in snapshot.issues],
+        "pagination": {
+            "prs": _page(snapshot.pr_pagination),
+            "issues": _page(snapshot.issue_pagination),
+        },
+    }

webui/queue_views.py

@@ -0,0 +1,121 @@
+"""HTML views for the live Gitea queue dashboard (#429)."""
+
+from __future__ import annotations
+
+import html
+
+from webui.queue_loader import PaginationMeta, QueueItem, QueueSnapshot
+
+
+def _badge_html(badges: tuple[str, ...]) -> str:
+    if not badges:
+        return ""
+    chips = "".join(
+        f'<span class="badge badge-{html.escape(b)}">{html.escape(b)}</span>'
+        for b in badges
+    )
+    return f'<span class="badges">{chips}</span>'
+
+
+def _pagination_html(label: str, meta: PaginationMeta | None) -> str:
+    if meta is None:
+        return (
+            f"<p class='muted'><strong>{html.escape(label)} pagination:</strong> "
+            "unavailable</p>"
+        )
+    status = "complete" if meta.inventory_complete else "partial"
+    more = "yes" if meta.has_more else "no"
+    return (
+        f"<p class='meta'><strong>{html.escape(label)} pagination ({status}):</strong> "
+        f"returned {meta.returned_count} · per_page {meta.per_page} · "
+        f"pages_fetched {meta.pages_fetched} · has_more {more} · "
+        f"final_page {'yes' if meta.is_final_page else 'no'}</p>"
+    )
+
+
+def _queue_table(
+    *,
+    title: str,
+    items: tuple[QueueItem, ...],
+    columns: tuple[tuple[str, str], ...],
+) -> str:
+    if not items:
+        return f"<h3>{html.escape(title)}</h3><p class='muted'>No open items.</p>"
+
+    headers = "".join(f"<th>{html.escape(label)}</th>" for _, label in columns)
+    rows = []
+    for item in items:
+        cells = []
+        for key, _ in columns:
+            if key == "number":
+                cells.append(f"<td>#{item.number}</td>")
+            elif key == "title":
+                cells.append(
+                    f"<td>{html.escape(item.title)}{_badge_html(item.badges)}</td>"
+                )
+            else:
+                cells.append(f"<td>{html.escape(item.extra.get(key, ''))}</td>")
+        rows.append("<tr>" + "".join(cells) + "</tr>")
+
+    body = (
+        f"<h3>{html.escape(title)}</h3>"
+        f"<table class='registry'><thead><tr>{headers}</tr></thead>"
+        f"<tbody>{''.join(rows)}</tbody></table>"
+    )
+    return body
+
+
+def render_queue_page(snapshot: QueueSnapshot) -> str:
+    error_block = ""
+    if snapshot.fetch_error:
+        error_block = (
+            f'<div class="stub"><p><strong>Queue unavailable:</strong> '
+            f"{html.escape(snapshot.fetch_error)}</p></div>"
+        )
+        inventory_block = (
+            "<p class='muted'><strong>Inventory:</strong> not fetched "
+            "(fail closed — queue tables omitted).</p>"
+        )
+        return (
+            "<h2>Live queue</h2>"
+            f"<p class='meta'>Repository: <code>{html.escape(snapshot.repo_label)}</code> "
+            f"· project <code>{html.escape(snapshot.project_id)}</code></p>"
+            f"{error_block}"
+            f"{inventory_block}"
+            "<p class='muted'>Read-only MVP — claims, reviews, and merges stay in Gitea/MCP tools.</p>"
+        )
+
+    pr_section = _queue_table(
+        title="Open pull requests",
+        items=snapshot.prs,
+        columns=(
+            ("number", "#"),
+            ("title", "Title"),
+            ("branch", "Branch"),
+            ("head_sha", "Head"),
+            ("mergeable", "Mergeable"),
+            ("linked_issue", "Linked issue"),
+        ),
+    )
+    issue_section = _queue_table(
+        title="Open issues",
+        items=snapshot.issues,
+        columns=(
+            ("number", "#"),
+            ("title", "Title"),
+            ("labels", "Labels"),
+            ("assignee", "Assignee"),
+            ("state", "State"),
+        ),
+    )
+
+    return (
+        "<h2>Live queue</h2>"
+        f"<p class='meta'>Repository: <code>{html.escape(snapshot.repo_label)}</code> "
+        f"· project <code>{html.escape(snapshot.project_id)}</code></p>"
+        f"{_pagination_html('PR', snapshot.pr_pagination)}"
+        f"{pr_section}"
+        f"{_pagination_html('Issue', snapshot.issue_pagination)}"
+        f"{issue_section}"
+        "<p class='muted'>Read-only MVP — claims, reviews, and merges stay in Gitea/MCP tools.</p>"
+    )