Compare commits
| Author | SHA1 | Date | |
|---|---|---|---|
  jcwalker3andClaude Opus 4.8
|
08061b7b8a |
@@ -67,6 +67,11 @@ that govern when a write path may open (#632, epic #631).
|
||||
| `/api/actions/{id}/preview` | Mutation ledger preview (GET, read-only) |
|
||||
| `/leases` | Lease and collision visibility (#433) |
|
||||
| `/api/leases` | JSON lease/collision export |
|
||||
| `/sessions` | Phase 1 shell stub — session inventory (backed by #636) |
|
||||
| `/inventory` | Phase 1 shell stub — unified inventory (backed by #636) |
|
||||
| `/timeline` | Phase 1 shell stub — workflow event timeline |
|
||||
| `/policy` | Phase 1 shell stub — capability/role policy placeholder |
|
||||
| `/insights` | Phase 1 shell stub — operational insights placeholder |
|
||||
|
||||
Most routes are GET-only. POST/PUT/PATCH/DELETE return `405` with
|
||||
`read-only-mvp`, except `/audit` and `/api/audit` which accept POST for
|
||||
@@ -147,6 +152,26 @@ health, workflow/schema SHA-256 hashes, and stale-runtime warnings when the
|
||||
checkout is behind merged safety-gate changes. Restart guidance links to #420;
|
||||
no tokens or MCP restart actions are exposed.
|
||||
|
||||
## Application shell — Phase 1 (#638)
|
||||
|
||||
The console shell (`webui/layout.py`) renders a grouped navigation driven by a
|
||||
single nav-config module, `webui/nav.py`. Nav groups follow the epic #631
|
||||
Phase 1 information architecture: **Health, Traffic, Runtime/Sessions,
|
||||
Projects, Inventory, Timeline, Policy** (placeholder), and **Insights**
|
||||
(placeholder). Live views and Phase 1 placeholders (`stub`) are declared in one
|
||||
place so the layout and the route table cannot drift.
|
||||
|
||||
The header carries two read-only status badges — an **environment** badge
|
||||
(`local` for loopback binds, `remote` otherwise, derived from `WEBUI_HOST`) and
|
||||
a **mode: read-only** badge — plus a **Docs** link to this document. No
|
||||
privileged action controls are present in the Phase 1 shell.
|
||||
|
||||
Not-yet-implemented surfaces (`/sessions`, `/inventory`, `/timeline`,
|
||||
`/policy`, `/insights`) resolve to graceful read-only stub pages instead of
|
||||
404s; their backing views land in later child issues of #631 (the inventory
|
||||
surfaces are backed by #636). Mutating methods on stub routes still fail closed
|
||||
with `read-only-mvp`.
|
||||
|
||||
## Deployment boundary (#435)
|
||||
|
||||
MVP serves on loopback by default. Binding `0.0.0.0` or `::` is **refused**
|
||||
|
||||
+2
-18
@@ -9173,27 +9173,11 @@ def gitea_publish_unpublished_issue_branch(
|
||||
if blocked:
|
||||
return blocked
|
||||
|
||||
# #815: resolve the caller's worktree *before* preflight and forward it, so
|
||||
# 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,
|
||||
)
|
||||
verify_preflight_purity(remote, task=task, org=org, repo=repo)
|
||||
|
||||
h, o, r = _resolve(remote, host, org, repo)
|
||||
git_remote = (git_remote_name or remote or "").strip()
|
||||
workspace = os.path.realpath(os.path.abspath((worktree_path or "").strip() or "."))
|
||||
|
||||
existing_lock = issue_lock_store.load_issue_lock(
|
||||
remote=remote, org=o, repo=r, issue_number=int(issue_number)
|
||||
|
||||
@@ -1,622 +0,0 @@
|
||||
"""Publication preflight must receive the caller's worktree (#815).
|
||||
|
||||
``gitea_publish_unpublished_issue_branch`` takes a **required** ``worktree_path``
|
||||
but resolved it only *after* ``verify_preflight_purity`` had already run. Every
|
||||
workspace-resolution layer behind that preflight — canonical root, root checkout,
|
||||
create-issue bootstrap, the #618 branches-only guard, issue scope, and anti-stomp
|
||||
— therefore received ``None`` and fell back to the MCP process root. A daemon
|
||||
rooted at the stable control checkout refused a valid registered issue worktree
|
||||
that the caller had explicitly supplied, before the publication assessor ever ran.
|
||||
|
||||
The #812 suite could not see this. Its fixture sets ``self.worktree =
|
||||
os.path.realpath(self.repo)`` and patches ``PROJECT_ROOT`` to that same path, so
|
||||
the fallback resolved to the very worktree the argument named. The production
|
||||
topology — control checkout on a stable branch, issue worktree somewhere else —
|
||||
was never constructed, and preflight additionally no-ops under pytest unless
|
||||
production guards are forced on.
|
||||
|
||||
These tests build that topology honestly:
|
||||
|
||||
* ``PROJECT_ROOT`` is a control checkout sitting on ``master``;
|
||||
* the registered issue worktree is a genuinely separate path under ``branches/``;
|
||||
* ``GITEA_TEST_FORCE_PRODUCTION_GUARDS`` is set so the #618 guard really runs;
|
||||
* no patch makes the issue worktree appear to be ``PROJECT_ROOT``.
|
||||
|
||||
Nothing here reads, writes, or references the protected worktree named in #812
|
||||
AC17 and #815 AC9. Every fixture is built from scratch against a local bare
|
||||
remote, so publication and read-after-write verification genuinely execute.
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import os
|
||||
import subprocess
|
||||
import sys
|
||||
import tempfile
|
||||
import unittest
|
||||
from datetime import datetime, timedelta, timezone
|
||||
from unittest.mock import patch
|
||||
|
||||
sys.path.insert(0, os.path.dirname(os.path.dirname(os.path.abspath(__file__))))
|
||||
sys.path.insert(0, os.path.dirname(os.path.abspath(__file__)))
|
||||
|
||||
import issue_lock_provenance # noqa: E402
|
||||
import issue_lock_store # noqa: E402
|
||||
import mcp_server # noqa: E402
|
||||
from mutation_profile_fixture import shared_mutation_env # noqa: E402
|
||||
|
||||
ISSUE = 9815
|
||||
BRANCH = f"feat/issue-{ISSUE}-forwarding-fixture"
|
||||
WORKTREE_DIRNAME = BRANCH.replace("/", "-")
|
||||
IDENTITY = "example-user"
|
||||
PROFILE = "test-author-prgs"
|
||||
ORG = "Scaled-Tech-Consulting"
|
||||
REPO = "Gitea-Tools"
|
||||
GIT_REMOTE = "prgs"
|
||||
|
||||
AUTHOR_PROFILE = {
|
||||
"profile_name": "prgs-author",
|
||||
"role": "author",
|
||||
"allowed_operations": [
|
||||
"gitea.read", "gitea.issue.create", "gitea.issue.comment",
|
||||
"gitea.pr.create", "gitea.repo.commit", "gitea.branch.push",
|
||||
],
|
||||
"forbidden_operations": [],
|
||||
"audit_label": "prgs-author",
|
||||
}
|
||||
|
||||
|
||||
def _ts(hours: int) -> str:
|
||||
return (
|
||||
(datetime.now(timezone.utc) + timedelta(hours=hours))
|
||||
.isoformat()
|
||||
.replace("+00:00", "Z")
|
||||
)
|
||||
|
||||
|
||||
class TestPreflightReceivesTheWorktree(unittest.TestCase):
|
||||
"""AC1 — the supplied path reaches ``verify_preflight_purity`` itself.
|
||||
|
||||
Follows the #735 capture pattern: replace preflight with a recorder that
|
||||
raises, so the argument can be proven forwarded without performing the
|
||||
mutation. This is the direct unit-level statement of the defect.
|
||||
"""
|
||||
|
||||
def _capture_preflight(self, **kwargs):
|
||||
captured: dict = {}
|
||||
|
||||
def _capture(*a, **kw):
|
||||
captured.update(kw)
|
||||
captured["_args"] = a
|
||||
raise RuntimeError("capture-only")
|
||||
|
||||
with patch.object(
|
||||
mcp_server, "verify_preflight_purity", side_effect=_capture
|
||||
), patch.object(
|
||||
mcp_server, "get_profile", return_value=AUTHOR_PROFILE
|
||||
), patch.object(
|
||||
mcp_server, "_resolve",
|
||||
return_value=("gitea.prgs.cc", ORG, REPO),
|
||||
), patch.object(
|
||||
mcp_server, "_auth", return_value="token fake",
|
||||
), patch.object(
|
||||
mcp_server.role_session_router,
|
||||
"check_author_mutation_after_reviewer_stop",
|
||||
return_value=(True, []),
|
||||
), patch.object(
|
||||
mcp_server, "_namespace_mutation_block", return_value=None
|
||||
), patch.object(
|
||||
mcp_server, "_profile_permission_block", return_value=None
|
||||
):
|
||||
try:
|
||||
mcp_server.gitea_publish_unpublished_issue_branch(**kwargs)
|
||||
except RuntimeError as exc:
|
||||
if "capture-only" not in str(exc) and not captured:
|
||||
raise
|
||||
self.assertTrue(
|
||||
captured,
|
||||
"gitea_publish_unpublished_issue_branch never called "
|
||||
"verify_preflight_purity",
|
||||
)
|
||||
return captured
|
||||
|
||||
def _base_kwargs(self, **overrides):
|
||||
kwargs = {
|
||||
"issue_number": ISSUE,
|
||||
"branch_name": BRANCH,
|
||||
"worktree_path": "/tmp/issue-815-explicit-worktree",
|
||||
"expected_head": "a" * 40,
|
||||
"remote": "prgs",
|
||||
"org": ORG,
|
||||
"repo": REPO,
|
||||
"git_remote_name": GIT_REMOTE,
|
||||
}
|
||||
kwargs.update(overrides)
|
||||
return kwargs
|
||||
|
||||
def test_explicit_worktree_path_reaches_preflight(self):
|
||||
captured = self._capture_preflight(**self._base_kwargs())
|
||||
self.assertEqual(
|
||||
captured.get("worktree_path"),
|
||||
os.path.realpath(os.path.abspath("/tmp/issue-815-explicit-worktree")),
|
||||
"the authoritative worktree_path must be forwarded into preflight",
|
||||
)
|
||||
|
||||
def test_forwarded_path_is_the_one_publication_uses(self):
|
||||
"""AC4 — preflight and publication must judge the same resolved path."""
|
||||
raw = "/tmp/issue-815-explicit-worktree/./"
|
||||
captured = self._capture_preflight(**self._base_kwargs(worktree_path=raw))
|
||||
expected = os.path.realpath(os.path.abspath(raw.strip()))
|
||||
self.assertEqual(captured.get("worktree_path"), expected)
|
||||
|
||||
def test_blank_worktree_path_forwards_none(self):
|
||||
"""AC5/AC8 — nothing usable supplied keeps the fail-closed fallback."""
|
||||
for blank in ("", " "):
|
||||
with self.subTest(blank=repr(blank)):
|
||||
captured = self._capture_preflight(
|
||||
**self._base_kwargs(worktree_path=blank)
|
||||
)
|
||||
self.assertIsNone(
|
||||
captured.get("worktree_path"),
|
||||
"a blank worktree must not resolve to the process cwd",
|
||||
)
|
||||
|
||||
def test_org_repo_and_task_forwarding_are_not_regressed(self):
|
||||
"""AC6 — #735's org/repo forwarding and the task name still hold."""
|
||||
captured = self._capture_preflight(**self._base_kwargs())
|
||||
self.assertEqual(captured.get("org"), ORG)
|
||||
self.assertEqual(captured.get("repo"), REPO)
|
||||
self.assertEqual(captured.get("task"), "publish_unpublished_branch")
|
||||
|
||||
|
||||
class _ProductionTopologyBase(unittest.TestCase):
|
||||
"""Control checkout on master + a distinct registered issue worktree.
|
||||
|
||||
This is the shape the production daemon runs in and the shape the #812
|
||||
fixture never built. ``PROJECT_ROOT`` is the control checkout; the issue
|
||||
worktree is a real registered worktree at a different path; production
|
||||
guards are forced on so the #618 branches-only guard genuinely evaluates.
|
||||
"""
|
||||
|
||||
def setUp(self):
|
||||
self.lock_dir = tempfile.TemporaryDirectory()
|
||||
self.addCleanup(self.lock_dir.cleanup)
|
||||
self.origin = tempfile.mkdtemp(prefix="issue815-origin-")
|
||||
self.control = tempfile.mkdtemp(prefix="issue815-control-")
|
||||
for path in (self.origin, self.control):
|
||||
self.addCleanup(
|
||||
lambda p=path: subprocess.run(["rm", "-rf", p], check=False)
|
||||
)
|
||||
self._init_repos()
|
||||
self.remotes = patch.dict(
|
||||
mcp_server.REMOTES,
|
||||
{"prgs": {"host": "gitea.prgs.cc", "org": ORG, "repo": REPO}},
|
||||
)
|
||||
self.remotes.start()
|
||||
self.addCleanup(patch.stopall)
|
||||
mcp_server._IDENTITY_CACHE.clear()
|
||||
|
||||
def _git(self, *args, cwd=None):
|
||||
return subprocess.run(
|
||||
["git", "-C", cwd or self.control, *args],
|
||||
capture_output=True, text=True, check=True,
|
||||
)
|
||||
|
||||
def _init_repos(self):
|
||||
subprocess.run(
|
||||
["git", "init", "-q", "--bare", "-b", "master", self.origin], check=True
|
||||
)
|
||||
self._git("init", "-q", "-b", "master")
|
||||
self._git("config", "user.email", "[email protected]")
|
||||
self._git("config", "user.name", "Test")
|
||||
self._git("remote", "add", GIT_REMOTE, self.origin)
|
||||
|
||||
with open(os.path.join(self.control, "seed.txt"), "w") as fh:
|
||||
fh.write("seed\n")
|
||||
# The real repository gitignores branches/, so a registered worktree
|
||||
# living there does not dirty the stable control checkout. Mirror that,
|
||||
# or the #615 dirty-runtime block fires on the worktree we just created.
|
||||
with open(os.path.join(self.control, ".gitignore"), "w") as fh:
|
||||
fh.write("branches/\n")
|
||||
self._git("add", "seed.txt", ".gitignore")
|
||||
self._git("commit", "-q", "-m", "seed")
|
||||
self.base_sha = self._git("rev-parse", "HEAD").stdout.strip()
|
||||
self._git("push", "-q", GIT_REMOTE, "master")
|
||||
|
||||
# The control checkout STAYS on master. This is the whole point: the
|
||||
# daemon's process root is the stable control checkout, never the
|
||||
# worktree the publication targets.
|
||||
self.worktree = os.path.realpath(
|
||||
os.path.join(self.control, "branches", WORKTREE_DIRNAME)
|
||||
)
|
||||
self._git("worktree", "add", "-q", "-b", BRANCH, self.worktree, "master")
|
||||
|
||||
with open(os.path.join(self.worktree, "work.txt"), "w") as fh:
|
||||
fh.write("unpublished implementation\n")
|
||||
self._git("add", "work.txt", cwd=self.worktree)
|
||||
self._git("commit", "-q", "-m", "unpublished implementation", cwd=self.worktree)
|
||||
self.head_sha = self._git("rev-parse", "HEAD", cwd=self.worktree).stdout.strip()
|
||||
|
||||
self.control_branch = self._git(
|
||||
"rev-parse", "--abbrev-ref", "HEAD"
|
||||
).stdout.strip()
|
||||
|
||||
# ── durable lock naming the caller and the issue worktree ────────────
|
||||
def lock_path(self):
|
||||
return issue_lock_store.lock_file_path(
|
||||
remote="prgs", org=ORG, repo=REPO, issue_number=ISSUE,
|
||||
lock_dir=self.lock_dir.name,
|
||||
)
|
||||
|
||||
def write_lock(self, *, bind_session=True, **overrides):
|
||||
path = self.lock_path()
|
||||
claimant = overrides.pop(
|
||||
"claimant", {"username": IDENTITY, "profile": PROFILE}
|
||||
)
|
||||
pid = overrides.pop("session_pid", os.getpid())
|
||||
lease = {
|
||||
"operation_type": issue_lock_store.AUTHOR_ISSUE_WORK_LEASE,
|
||||
"issue_number": ISSUE,
|
||||
"pr_number": None,
|
||||
"branch": overrides.get("branch_name", BRANCH),
|
||||
"worktree_path": overrides.get("worktree_path", self.worktree),
|
||||
"claimant": claimant,
|
||||
"created_at": _ts(-2),
|
||||
"last_heartbeat_at": _ts(-2),
|
||||
"expires_at": _ts(-1),
|
||||
}
|
||||
lease.update(overrides.pop("work_lease", {}))
|
||||
data = {
|
||||
"issue_number": ISSUE,
|
||||
"branch_name": BRANCH,
|
||||
"remote": "prgs",
|
||||
"org": ORG,
|
||||
"repo": REPO,
|
||||
"worktree_path": self.worktree,
|
||||
"session_pid": pid,
|
||||
"pid": pid,
|
||||
"lock_generation": 1,
|
||||
"work_lease": lease,
|
||||
"lock_provenance": issue_lock_provenance.build_sanctioned_lock_provenance(
|
||||
tool="gitea_lock_issue", claimant=claimant
|
||||
),
|
||||
}
|
||||
data.update(overrides)
|
||||
data["lock_file_path"] = path
|
||||
issue_lock_store.save_lock_file(path, data)
|
||||
# Bind the session pointer so the #683 issue-scope guard resolves an
|
||||
# owning issue for this author session. In real production the publish
|
||||
# task does not require a session lock — require_author_lock is keyed on
|
||||
# the test-only production_guards_forced() flag, which this suite must
|
||||
# set to make preflight run at all — so this pointer is fixture
|
||||
# scaffolding to clear a guard production would not apply here, never a
|
||||
# softening of the worktree-forwarding behaviour under test. The
|
||||
# preflight-negative cases below leave it unbound precisely so the #618
|
||||
# guard is reached with no session fallback to rescue a bad worktree.
|
||||
if bind_session:
|
||||
pointer = {
|
||||
"pid": os.getpid(),
|
||||
"lock_file_path": path,
|
||||
"issue_number": ISSUE,
|
||||
"branch_name": data["branch_name"],
|
||||
"remote": "prgs",
|
||||
"org": ORG,
|
||||
"repo": REPO,
|
||||
}
|
||||
issue_lock_store.save_lock_file(
|
||||
issue_lock_store.session_pointer_path(self.lock_dir.name), pointer
|
||||
)
|
||||
return path
|
||||
|
||||
def _tool_env(self):
|
||||
env = shared_mutation_env(
|
||||
PROFILE, include_example_repo=True,
|
||||
GITEA_ISSUE_LOCK_DIR=self.lock_dir.name,
|
||||
)
|
||||
env["GITEA_ISSUE_LOCK_DIR"] = self.lock_dir.name
|
||||
# The defect only exists where preflight actually runs. Under pytest the
|
||||
# production root/branches/scope guards are skipped unless forced on, so
|
||||
# force them: this test exists to exercise the #618 guard, not to bypass
|
||||
# it. Parity is pinned to the server's own startup head so repointing
|
||||
# PROJECT_ROOT does not read as a stale daemon.
|
||||
env["GITEA_TEST_FORCE_PRODUCTION_GUARDS"] = "1"
|
||||
# Production is a promoted stable-control runtime. The pytest process
|
||||
# itself runs from a branches/ worktree, which the #615 runtime-mode
|
||||
# gate correctly classifies as dev-test; declaring the sanctioned mode
|
||||
# models the production daemon rather than defeating the gate. Without
|
||||
# this, forcing production guards on would trip the *runtime-mode* block
|
||||
# for a reason unrelated to the #815 worktree-forwarding defect.
|
||||
env["GITEA_MCP_RUNTIME_MODE"] = "stable-control"
|
||||
startup_head = mcp_server._STARTUP_PARITY.get("startup_head") or ""
|
||||
env["GITEA_TEST_CURRENT_HEAD"] = startup_head
|
||||
env["GITEA_TEST_LIVE_REMOTE_HEAD"] = startup_head
|
||||
return env
|
||||
|
||||
def run_publish(self, *, open_prs=None, expected_head=None, **kwargs):
|
||||
"""Drive the public tool with PROJECT_ROOT pinned to the CONTROL checkout."""
|
||||
env = self._tool_env()
|
||||
with patch(
|
||||
"mcp_server._list_open_pulls", return_value=list(open_prs or [])
|
||||
), patch(
|
||||
"mcp_server._auth", return_value="token x"
|
||||
), patch(
|
||||
"mcp_server.get_auth_header", return_value="token x"
|
||||
), patch(
|
||||
"mcp_server._work_lease_claimant",
|
||||
return_value={"username": IDENTITY, "profile": PROFILE},
|
||||
), patch.object(
|
||||
# NOTE: the control checkout — deliberately NOT self.worktree.
|
||||
mcp_server, "PROJECT_ROOT", self.control
|
||||
), patch.dict(os.environ, env, clear=True):
|
||||
os.environ["GITEA_ISSUE_LOCK_DIR"] = self.lock_dir.name
|
||||
return mcp_server.gitea_publish_unpublished_issue_branch(
|
||||
issue_number=kwargs.pop("issue_number", ISSUE),
|
||||
branch_name=kwargs.pop("branch_name", BRANCH),
|
||||
worktree_path=kwargs.pop("worktree_path", self.worktree),
|
||||
expected_head=expected_head or self.head_sha,
|
||||
remote="prgs",
|
||||
git_remote_name=kwargs.pop("git_remote_name", GIT_REMOTE),
|
||||
**kwargs,
|
||||
)
|
||||
|
||||
def remote_head(self, branch=BRANCH):
|
||||
res = subprocess.run(
|
||||
["git", "-C", self.origin, "rev-parse", "--verify", "--quiet", branch],
|
||||
capture_output=True, text=True, check=False,
|
||||
)
|
||||
return (res.stdout or "").strip() or None
|
||||
|
||||
|
||||
class TestForwardingClearsThe618Guard(_ProductionTopologyBase):
|
||||
"""AC2 — the faithful production reproduction, and the sharpest fix proof.
|
||||
|
||||
The production recovery worker had **no** session issue lock — acquiring one
|
||||
was the very thing the deadlock prevented — so preflight had nothing but the
|
||||
explicit ``worktree_path`` argument to resolve the workspace from. This class
|
||||
reproduces exactly that: no session pointer is bound, so there is no
|
||||
author-lock fallback to rescue a dropped argument.
|
||||
|
||||
With the argument forwarded (fixed source) the #618 branches-only guard
|
||||
accepts the registered issue worktree and the call advances to the next
|
||||
guard. With the argument dropped (the buggy source this issue reports)
|
||||
preflight falls back to ``PROJECT_ROOT`` — the stable control checkout — and
|
||||
the #618 guard traps the call there. The two outcomes are told apart by the
|
||||
guard that fired, on its own error text.
|
||||
|
||||
This test therefore *fails* against the unpatched source (the call is trapped
|
||||
at #618 instead of clearing it), which is what makes it a regression rather
|
||||
than a smoke test.
|
||||
"""
|
||||
|
||||
_CONTROL_CHECKOUT_MARKERS = ("stable control checkout", "#618")
|
||||
|
||||
def test_explicit_worktree_clears_618_without_a_session_lock(self):
|
||||
# No write_lock(): the session is deliberately unbound, as in production.
|
||||
with self.assertRaises(RuntimeError) as ctx:
|
||||
self.run_publish()
|
||||
message = str(ctx.exception)
|
||||
# The workspace guard is satisfied — the failure is the *later* scope
|
||||
# guard (no owning issue), never the control-checkout refusal. If the
|
||||
# argument were dropped, this call would be trapped at #618 instead.
|
||||
for marker in self._CONTROL_CHECKOUT_MARKERS:
|
||||
self.assertNotIn(
|
||||
marker, message,
|
||||
f"the explicit worktree must clear #618; got a control-checkout "
|
||||
f"refusal instead: {message}",
|
||||
)
|
||||
self.assertIn(
|
||||
"owning issue", message,
|
||||
f"expected the downstream scope guard to fire, got: {message}",
|
||||
)
|
||||
self.assertIsNone(self.remote_head())
|
||||
|
||||
def test_dropped_argument_would_be_trapped_at_618(self):
|
||||
# Simulate the buggy call shape directly: no session lock, and preflight
|
||||
# given no worktree, exactly as the unpatched source left it. This pins
|
||||
# the control-checkout refusal that the fix eliminates, so the pair of
|
||||
# tests brackets the defect from both sides regardless of which source
|
||||
# version is loaded.
|
||||
env = self._tool_env()
|
||||
with patch(
|
||||
"mcp_server._list_open_pulls", return_value=[]
|
||||
), patch(
|
||||
"mcp_server._auth", return_value="token x"
|
||||
), patch(
|
||||
"mcp_server.get_auth_header", return_value="token x"
|
||||
), patch(
|
||||
"mcp_server._work_lease_claimant",
|
||||
return_value={"username": IDENTITY, "profile": PROFILE},
|
||||
), patch.object(
|
||||
mcp_server, "PROJECT_ROOT", self.control
|
||||
), patch.dict(os.environ, env, clear=True):
|
||||
os.environ["GITEA_ISSUE_LOCK_DIR"] = self.lock_dir.name
|
||||
with self.assertRaises(RuntimeError) as ctx:
|
||||
# Drive verify_preflight_purity the way the buggy body did:
|
||||
# no worktree_path forwarded at all.
|
||||
mcp_server.verify_preflight_purity(
|
||||
"prgs",
|
||||
task="publish_unpublished_branch",
|
||||
org=ORG,
|
||||
repo=REPO,
|
||||
)
|
||||
message = str(ctx.exception)
|
||||
self.assertTrue(
|
||||
any(m in message for m in self._CONTROL_CHECKOUT_MARKERS),
|
||||
f"a dropped worktree must trap at the control checkout: {message}",
|
||||
)
|
||||
self.assertIsNone(self.remote_head())
|
||||
|
||||
|
||||
class TestProductionTopologyPublishes(_ProductionTopologyBase):
|
||||
"""AC2/AC4/AC7 — the explicit registered worktree is what preflight validates."""
|
||||
|
||||
def test_fixture_is_genuinely_the_production_topology(self):
|
||||
"""Guard the guard: if this drifts, the regression stops meaning anything."""
|
||||
self.assertNotEqual(
|
||||
os.path.realpath(self.control), self.worktree,
|
||||
"the issue worktree must not be PROJECT_ROOT",
|
||||
)
|
||||
self.assertEqual(
|
||||
self.control_branch, "master",
|
||||
"the control checkout must sit on a stable branch",
|
||||
)
|
||||
self.assertTrue(
|
||||
os.path.realpath(self.worktree).startswith(
|
||||
os.path.realpath(os.path.join(self.control, "branches")) + os.sep
|
||||
),
|
||||
"the issue worktree must live under branches/",
|
||||
)
|
||||
listed = subprocess.run(
|
||||
["git", "-C", self.control, "worktree", "list"],
|
||||
capture_output=True, text=True, check=True,
|
||||
).stdout
|
||||
self.assertIn(
|
||||
self.worktree, listed, "the issue worktree must be genuinely registered"
|
||||
)
|
||||
|
||||
def test_publishes_from_a_control_rooted_daemon(self):
|
||||
"""The exact production failure: this refused with #618 before the fix."""
|
||||
self.write_lock()
|
||||
self.assertIsNone(self.remote_head(), "fixture must start unpublished")
|
||||
res = self.run_publish()
|
||||
self.assertTrue(res.get("success"), res)
|
||||
self.assertTrue(res.get("performed"), res)
|
||||
self.assertEqual(self.remote_head(), self.head_sha)
|
||||
|
||||
def test_dry_run_uses_the_explicit_worktree(self):
|
||||
"""AC4 — dry-run reaches the same decision without publishing."""
|
||||
self.write_lock()
|
||||
res = self.run_publish(dry_run=True)
|
||||
self.assertTrue(res.get("success"), res)
|
||||
self.assertFalse(res.get("performed"), res)
|
||||
self.assertTrue(res.get("would_publish"), res)
|
||||
self.assertIsNone(self.remote_head(), "dry-run must not publish")
|
||||
|
||||
def test_dry_run_and_apply_agree_on_the_same_worktree(self):
|
||||
"""AC4 — both paths resolve the same workspace, so both succeed."""
|
||||
self.write_lock()
|
||||
dry = self.run_publish(dry_run=True)
|
||||
self.assertTrue(dry.get("would_publish"), dry)
|
||||
applied = self.run_publish()
|
||||
self.assertTrue(applied.get("performed"), applied)
|
||||
self.assertEqual(self.remote_head(), self.head_sha)
|
||||
|
||||
def test_read_after_write_verification_still_runs(self):
|
||||
"""AC6 — PR #814's post-publication verification is unchanged."""
|
||||
self.write_lock()
|
||||
res = self.run_publish()
|
||||
self.assertTrue(res.get("verified"), res)
|
||||
self.assertEqual(res.get("remote_head_sha"), self.head_sha)
|
||||
|
||||
|
||||
class TestProductionTopologyFailsClosed(_ProductionTopologyBase):
|
||||
"""AC3/AC5/AC8 — the fix does not weaken any refusal.
|
||||
|
||||
A refusal reaches the caller by one of two mechanisms, and this class holds
|
||||
them apart deliberately. A bad *workspace* is caught by the #618 preflight
|
||||
guard, which raises before the assessor is built. A bad *content/ownership*
|
||||
fact passes preflight (the worktree itself is fine) and is then refused by
|
||||
the publication assessor, which returns ``success: False``. Both are
|
||||
fail-closed; asserting the wrong mechanism would hide a regression.
|
||||
"""
|
||||
|
||||
# ── #618 preflight refusals: no session lock, so nothing rescues a bad
|
||||
# workspace and the guard fires exactly as it does in production ──────
|
||||
def _assert_preflight_raises(self, **kwargs):
|
||||
with self.assertRaises(RuntimeError) as ctx:
|
||||
self.run_publish(**kwargs)
|
||||
self.assertIsNone(
|
||||
self.remote_head(), "a blocked publication must not reach the remote"
|
||||
)
|
||||
return str(ctx.exception)
|
||||
|
||||
def test_blank_worktree_path_fails_closed_via_618(self):
|
||||
"""AC5 — a blank path forwards None, so preflight sees the control root."""
|
||||
for blank in ("", " "):
|
||||
with self.subTest(blank=repr(blank)):
|
||||
message = self._assert_preflight_raises(worktree_path=blank)
|
||||
self.assertIn("618", message)
|
||||
|
||||
def test_control_checkout_as_worktree_fails_closed_via_618(self):
|
||||
"""AC5 — naming the stable control checkout explicitly is still refused."""
|
||||
message = self._assert_preflight_raises(worktree_path=self.control)
|
||||
self.assertIn("618", message)
|
||||
|
||||
def test_unregistered_directory_fails_closed(self):
|
||||
"""AC3 — a plain directory under branches/ is not a registered worktree."""
|
||||
bogus = os.path.join(self.control, "branches", "not-a-worktree")
|
||||
os.makedirs(bogus, exist_ok=True)
|
||||
self._assert_preflight_raises(worktree_path=bogus)
|
||||
|
||||
def test_missing_worktree_path_fails_closed(self):
|
||||
"""AC3 — a path that does not exist is refused, not silently replaced."""
|
||||
missing = os.path.join(self.control, "branches", "absent-worktree")
|
||||
self._assert_preflight_raises(worktree_path=missing)
|
||||
|
||||
# ── assessor refusals: preflight passes on a valid worktree, then the
|
||||
# publication assessor refuses on content/ownership evidence ──────────
|
||||
def _assert_assessor_refuses(self, **kwargs):
|
||||
res = self.run_publish(**kwargs)
|
||||
self.assertFalse(res.get("success"), res)
|
||||
self.assertFalse(res.get("performed"), res)
|
||||
self.assertIsNone(self.remote_head())
|
||||
return res
|
||||
|
||||
def test_changed_local_head_still_refuses(self):
|
||||
"""AC6 — the declared expected_head remains authoritative."""
|
||||
self.write_lock()
|
||||
self._assert_assessor_refuses(expected_head="b" * 40)
|
||||
|
||||
def test_foreign_claimant_still_refuses(self):
|
||||
"""AC6 — ownership still comes from the durable lock record."""
|
||||
self.write_lock(claimant={"username": "someone-else", "profile": PROFILE})
|
||||
self._assert_assessor_refuses()
|
||||
|
||||
def test_dirty_worktree_still_refuses(self):
|
||||
"""AC6 — cleanliness enforcement survives the forwarding change."""
|
||||
self.write_lock()
|
||||
with open(os.path.join(self.worktree, "work.txt"), "a") as fh:
|
||||
fh.write("uncommitted drift\n")
|
||||
self._assert_assessor_refuses()
|
||||
|
||||
def test_competing_open_pr_still_refuses(self):
|
||||
"""AC6 — a rival claim on another branch still blocks."""
|
||||
self.write_lock()
|
||||
self._assert_assessor_refuses(
|
||||
open_prs=[{"number": 4242, "head": {"ref": f"fix/issue-{ISSUE}-rival"}}]
|
||||
)
|
||||
|
||||
def test_issue_lock_record_is_not_mutated_by_a_refusal(self):
|
||||
"""AC6 — record separation (#812 AC23) is unaffected by this change."""
|
||||
path = self.write_lock()
|
||||
with open(path, "rb") as fh:
|
||||
before = fh.read()
|
||||
self._assert_assessor_refuses(expected_head="c" * 40)
|
||||
with open(path, "rb") as fh:
|
||||
self.assertEqual(before, fh.read())
|
||||
|
||||
|
||||
class TestProtectedFixtureNotReferenced(unittest.TestCase):
|
||||
"""AC9 — this regression never names the protected #635 fixture.
|
||||
|
||||
The forbidden tokens are reconstructed from fragments so this assertion
|
||||
file does not itself contain them and produce a false positive.
|
||||
"""
|
||||
|
||||
def test_no_reference_to_the_protected_worktree(self):
|
||||
forbidden = [
|
||||
"issue-635-" + "project-registry-api",
|
||||
"b2f6e9a6dc40e9651ef8" + "76f322dd0a68bddebfd8",
|
||||
]
|
||||
here = os.path.abspath(__file__)
|
||||
with open(here, "r", encoding="utf-8") as fh:
|
||||
text = fh.read()
|
||||
for token in forbidden:
|
||||
self.assertNotIn(
|
||||
token, text,
|
||||
f"the protected #635 fixture must not be referenced: {token}",
|
||||
)
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
unittest.main()
|
||||
@@ -0,0 +1,135 @@
|
||||
"""Tests for the Phase 1 operator console application shell (#638)."""
|
||||
import sys
|
||||
import unittest
|
||||
from pathlib import Path
|
||||
|
||||
sys.path.insert(0, str(Path(__file__).resolve().parent.parent))
|
||||
|
||||
from starlette.routing import Route
|
||||
from starlette.testclient import TestClient
|
||||
|
||||
from webui import layout
|
||||
from webui.app import create_app
|
||||
from webui.nav import NAV_GROUPS, STUB_PAGES, nav_hrefs
|
||||
|
||||
|
||||
class TestShellNav(unittest.TestCase):
|
||||
def setUp(self):
|
||||
self.client = TestClient(create_app())
|
||||
|
||||
def test_nav_group_labels_present(self):
|
||||
text = self.client.get("/").text
|
||||
for group in NAV_GROUPS:
|
||||
with self.subTest(group=group.label):
|
||||
self.assertIn(f">{group.label}<", text)
|
||||
|
||||
def test_phase1_group_labels_cover_expected_ia(self):
|
||||
labels = {group.label for group in NAV_GROUPS}
|
||||
for expected in (
|
||||
"Health",
|
||||
"Traffic",
|
||||
"Runtime/Sessions",
|
||||
"Projects",
|
||||
"Inventory",
|
||||
"Timeline",
|
||||
"Policy",
|
||||
"Insights",
|
||||
):
|
||||
with self.subTest(label=expected):
|
||||
self.assertIn(expected, labels)
|
||||
|
||||
def test_every_nav_href_resolves_to_a_get_route(self):
|
||||
app = create_app()
|
||||
get_paths = {
|
||||
route.path
|
||||
for route in app.routes
|
||||
if isinstance(route, Route) and "GET" in route.methods
|
||||
}
|
||||
for href in nav_hrefs():
|
||||
with self.subTest(href=href):
|
||||
self.assertIn(href, get_paths, f"nav href {href} has no GET route")
|
||||
|
||||
def test_legacy_hrefs_still_navigable(self):
|
||||
text = self.client.get("/").text
|
||||
for href in ("/queue", "/projects", "/prompts", "/runtime",
|
||||
"/audit", "/worktrees", "/leases", "/actions"):
|
||||
with self.subTest(href=href):
|
||||
self.assertIn(f'href="{href}"', text)
|
||||
|
||||
|
||||
class TestShellBadges(unittest.TestCase):
|
||||
def setUp(self):
|
||||
self.client = TestClient(create_app())
|
||||
|
||||
def test_mode_badge_present(self):
|
||||
self.assertIn("mode: read-only", self.client.get("/").text)
|
||||
|
||||
def test_environment_badge_present(self):
|
||||
self.assertIn("env:", self.client.get("/").text)
|
||||
|
||||
def test_default_environment_is_local(self):
|
||||
self.assertEqual(layout.environment_label(), "local")
|
||||
|
||||
def test_remote_bind_reports_remote_environment(self):
|
||||
import os
|
||||
|
||||
prior = os.environ.get("WEBUI_HOST")
|
||||
os.environ["WEBUI_HOST"] = "10.0.0.5"
|
||||
try:
|
||||
self.assertEqual(layout.environment_label(), "remote")
|
||||
finally:
|
||||
if prior is None:
|
||||
os.environ.pop("WEBUI_HOST", None)
|
||||
else:
|
||||
os.environ["WEBUI_HOST"] = prior
|
||||
|
||||
def test_docs_link_present(self):
|
||||
text = self.client.get("/").text
|
||||
self.assertIn(layout.DOCS_URL, text)
|
||||
self.assertIn(">Docs<", text)
|
||||
|
||||
|
||||
class TestShellStubs(unittest.TestCase):
|
||||
def setUp(self):
|
||||
self.client = TestClient(create_app())
|
||||
|
||||
def test_stub_routes_render_200(self):
|
||||
for path, (title, _desc) in STUB_PAGES.items():
|
||||
with self.subTest(path=path):
|
||||
response = self.client.get(path)
|
||||
self.assertEqual(response.status_code, 200, path)
|
||||
self.assertIn(title, response.text)
|
||||
self.assertIn("placeholder", response.text)
|
||||
|
||||
def test_stub_routes_are_read_only(self):
|
||||
for path in STUB_PAGES:
|
||||
with self.subTest(path=path):
|
||||
response = self.client.post(path)
|
||||
self.assertEqual(response.status_code, 405)
|
||||
self.assertEqual(response.json()["error"], "read-only-mvp")
|
||||
|
||||
def test_stub_pages_carry_nav_and_badges(self):
|
||||
response = self.client.get("/inventory")
|
||||
self.assertIn("mode: read-only", response.text)
|
||||
self.assertIn('href="/queue"', response.text)
|
||||
|
||||
|
||||
class TestShellHome(unittest.TestCase):
|
||||
def setUp(self):
|
||||
self.client = TestClient(create_app())
|
||||
|
||||
def test_home_summarizes_console(self):
|
||||
text = self.client.get("/").text
|
||||
self.assertIn("Operator console", text)
|
||||
self.assertIn("Phase 1", text)
|
||||
|
||||
def test_home_links_legacy_pages(self):
|
||||
text = self.client.get("/").text
|
||||
self.assertIn("MVP legacy pages", text)
|
||||
for href in ("/queue", "/audit", "/leases"):
|
||||
with self.subTest(href=href):
|
||||
self.assertIn(f'href="{href}"', text)
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
unittest.main()
|
||||
+54
-11
@@ -11,6 +11,7 @@ from starlette.routing import Route
|
||||
|
||||
from webui.deployment_boundary import deployment_snapshot
|
||||
from webui.layout import render_page
|
||||
from webui.nav import NAV_GROUPS, STUB_PAGES
|
||||
from webui.project_registry import find_project, load_registry, registry_to_dict
|
||||
from webui.project_views import render_project_detail, render_projects_list
|
||||
from webui.prompt_library import find_prompt, library_to_dict
|
||||
@@ -43,24 +44,62 @@ def _stub_page(title: str, description: str) -> HTMLResponse:
|
||||
return HTMLResponse(render_page(title=title, body_html=body))
|
||||
|
||||
|
||||
_LEGACY_PAGES = (
|
||||
("/queue", "Queue", "live PR and issue dashboard (#429)"),
|
||||
("/projects", "Projects", "registry and onboarding (#427)"),
|
||||
("/prompts", "Prompts", "canonical workflow prompt library (#428)"),
|
||||
("/runtime", "Runtime", "MCP health and stale-runtime detection (#430)"),
|
||||
("/audit", "Audit", "final-report paste and validator preview (#431)"),
|
||||
("/worktrees", "Worktrees", "branch hygiene dashboard (#432)"),
|
||||
("/leases", "Leases", "collision and lease visibility (#433)"),
|
||||
("/actions", "Actions", "gated write-action framework (#434)"),
|
||||
)
|
||||
|
||||
|
||||
def _render_home_nav_groups() -> str:
|
||||
groups = []
|
||||
for group in NAV_GROUPS:
|
||||
items = "".join(
|
||||
f'<li><a href="{item.href}">{item.label}</a>'
|
||||
+ ("" if item.status == "live" else " <span class=\"muted\">(stub)</span>")
|
||||
+ "</li>"
|
||||
for item in group.items
|
||||
)
|
||||
groups.append(f"<h3>{group.label}</h3><ul>{items}</ul>")
|
||||
return "".join(groups)
|
||||
|
||||
|
||||
async def home(_request: Request) -> HTMLResponse:
|
||||
legacy = "".join(
|
||||
f"<li><strong>{label}</strong> — {desc} "
|
||||
f'(<a href="{href}">{href}</a>)</li>'
|
||||
for href, label, desc in _LEGACY_PAGES
|
||||
)
|
||||
body = (
|
||||
"<h2>Operator console</h2>"
|
||||
"<p>Local entry point for MCP Control Plane operational views.</p>"
|
||||
"<ul>"
|
||||
"<li><strong>Queue</strong> — live PR and issue dashboard (#429)</li>"
|
||||
"<li><strong>Projects</strong> — registry and onboarding (#427)</li>"
|
||||
"<li><strong>Prompts</strong> — canonical workflow prompt library (#428)</li>"
|
||||
"<li><strong>Runtime</strong> — MCP health and stale-runtime detection (#430)</li>"
|
||||
"<li><strong>Audit</strong> — final-report paste and validator preview (#431)</li>"
|
||||
"<li><strong>Worktrees</strong> — branch hygiene dashboard (#432)</li>"
|
||||
"<li><strong>Leases</strong> — collision and lease visibility (#433)</li>"
|
||||
"<li><strong>Actions</strong> — gated write-action framework (#434)</li>"
|
||||
"</ul>"
|
||||
"<p>Read-only home for the MCP Control Plane Phase 1 operator console. "
|
||||
"Gitea, MCP capability gates, and canonical workflows remain the source "
|
||||
"of truth; this console never mutates them.</p>"
|
||||
"<h2>Phase 1 surfaces</h2>"
|
||||
+ _render_home_nav_groups()
|
||||
+ "<h2>MVP legacy pages</h2>"
|
||||
"<ul>" + legacy + "</ul>"
|
||||
)
|
||||
return HTMLResponse(render_page(title="Home", body_html=body))
|
||||
|
||||
|
||||
async def phase_stub(request: Request) -> HTMLResponse:
|
||||
"""Graceful read-only placeholder for a not-yet-implemented Phase 1 surface."""
|
||||
title, description = STUB_PAGES[request.url.path]
|
||||
body = (
|
||||
f"<h2>{title}</h2>"
|
||||
f'<div class="stub"><p>{description}</p>'
|
||||
"<p>Phase 1 shell placeholder — no write actions. Tracked under "
|
||||
"epic #631.</p></div>"
|
||||
)
|
||||
return HTMLResponse(render_page(title=title, body_html=body))
|
||||
|
||||
|
||||
async def health(_request: Request) -> JSONResponse:
|
||||
bind_host = _request.app.state.webui_bind_host
|
||||
return JSONResponse({
|
||||
@@ -291,6 +330,10 @@ def create_app(*, bind_host: str | None = None) -> Starlette:
|
||||
methods=["POST"],
|
||||
),
|
||||
Route("/api/leases", api_leases, methods=["GET"]),
|
||||
*[
|
||||
Route(path, phase_stub, methods=["GET"])
|
||||
for path in STUB_PAGES
|
||||
],
|
||||
],
|
||||
exception_handlers={405: method_not_allowed},
|
||||
)
|
||||
|
||||
+95
-17
@@ -2,28 +2,66 @@
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
NAV_ITEMS = (
|
||||
("/", "Home"),
|
||||
("/queue", "Queue"),
|
||||
("/projects", "Projects"),
|
||||
("/prompts", "Prompts"),
|
||||
("/runtime", "Runtime"),
|
||||
("/audit", "Audit"),
|
||||
("/worktrees", "Worktrees"),
|
||||
("/leases", "Leases"),
|
||||
("/actions", "Actions"),
|
||||
)
|
||||
import os
|
||||
|
||||
from webui.nav import NAV_GROUPS
|
||||
|
||||
MVP_NOTICE = (
|
||||
"Read-only MVP — Gitea, MCP tools, and canonical workflows remain the "
|
||||
"source of truth. No mutation endpoints."
|
||||
)
|
||||
|
||||
# Canonical docs entry point surfaced from the shell header (#638).
|
||||
DOCS_URL = (
|
||||
"https://gitea.prgs.cc/Scaled-Tech-Consulting/Gitea-Tools/src/branch/"
|
||||
"master/docs/webui-local-dev.md"
|
||||
)
|
||||
|
||||
_LOCAL_HOSTS = frozenset({"", "127.0.0.1", "localhost", "::1"})
|
||||
|
||||
|
||||
def environment_label() -> str:
|
||||
"""Classify the serving environment as ``local`` or ``remote`` (#638).
|
||||
|
||||
Derived from the same ``WEBUI_HOST`` default the app binds to; loopback
|
||||
hosts are ``local``, anything else is ``remote``. Read-only signal only.
|
||||
"""
|
||||
host = (os.environ.get("WEBUI_HOST", "127.0.0.1") or "").strip().lower()
|
||||
return "local" if host in _LOCAL_HOSTS else "remote"
|
||||
|
||||
|
||||
def _render_nav() -> str:
|
||||
groups_html = []
|
||||
for group in NAV_GROUPS:
|
||||
links = "".join(
|
||||
f'<a href="{item.href}"'
|
||||
+ (' class="nav-stub"' if item.status == "stub" else "")
|
||||
+ f'>{item.label}</a>'
|
||||
for item in group.items
|
||||
)
|
||||
groups_html.append(
|
||||
'<div class="nav-group">'
|
||||
f'<span class="nav-group-label">{group.label}</span>'
|
||||
f'<span class="nav-group-links">{links}</span>'
|
||||
"</div>"
|
||||
)
|
||||
return "".join(groups_html)
|
||||
|
||||
|
||||
def _render_badges() -> str:
|
||||
env = environment_label()
|
||||
return (
|
||||
'<div class="header-badges">'
|
||||
f'<span class="badge env-badge env-{env}">env: {env}</span>'
|
||||
'<span class="badge mode-badge">mode: read-only</span>'
|
||||
f'<a class="badge docs-link" href="{DOCS_URL}">Docs</a>'
|
||||
"</div>"
|
||||
)
|
||||
|
||||
|
||||
def render_page(*, title: str, body_html: str, extra_head: str = "") -> str:
|
||||
nav_links = "".join(
|
||||
f'<a href="{href}">{label}</a>' for href, label in NAV_ITEMS
|
||||
)
|
||||
nav_links = _render_nav()
|
||||
header_badges = _render_badges()
|
||||
return f"""<!DOCTYPE html>
|
||||
<html lang="en">
|
||||
<head>
|
||||
@@ -53,21 +91,58 @@ def render_page(*, title: str, body_html: str, extra_head: str = "") -> str:
|
||||
padding: 0.75rem 1.25rem;
|
||||
}}
|
||||
header h1 {{
|
||||
margin: 0 0 0.5rem;
|
||||
margin: 0;
|
||||
font-size: 1.1rem;
|
||||
font-weight: 600;
|
||||
}}
|
||||
.header-top {{
|
||||
display: flex;
|
||||
flex-wrap: wrap;
|
||||
align-items: center;
|
||||
justify-content: space-between;
|
||||
gap: 0.5rem 1rem;
|
||||
margin-bottom: 0.6rem;
|
||||
}}
|
||||
.header-badges {{ display: inline-flex; flex-wrap: wrap; gap: 0.4rem; }}
|
||||
.env-badge.env-local {{ color: #8fd19e; border-color: #3d6b4a; }}
|
||||
.env-badge.env-remote {{ color: #e0c27a; border-color: #6b5730; }}
|
||||
.mode-badge {{ color: #9ec8f0; border-color: #3d5f7a; }}
|
||||
a.docs-link {{
|
||||
color: var(--accent);
|
||||
border-color: var(--accent);
|
||||
text-decoration: none;
|
||||
text-transform: none;
|
||||
}}
|
||||
a.docs-link:hover {{ filter: brightness(1.12); }}
|
||||
nav {{
|
||||
display: flex;
|
||||
flex-wrap: wrap;
|
||||
gap: 0.75rem 1rem;
|
||||
gap: 0.5rem 1.25rem;
|
||||
}}
|
||||
.nav-group {{
|
||||
display: flex;
|
||||
flex-direction: column;
|
||||
gap: 0.15rem;
|
||||
}}
|
||||
.nav-group-label {{
|
||||
font-size: 0.68rem;
|
||||
text-transform: uppercase;
|
||||
letter-spacing: 0.04em;
|
||||
color: var(--muted);
|
||||
}}
|
||||
.nav-group-links {{ display: inline-flex; flex-wrap: wrap; gap: 0.6rem; }}
|
||||
nav a {{
|
||||
color: var(--accent);
|
||||
text-decoration: none;
|
||||
font-size: 0.9rem;
|
||||
}}
|
||||
nav a:hover {{ text-decoration: underline; }}
|
||||
nav a.nav-stub {{ color: var(--muted); }}
|
||||
nav a.nav-stub::after {{
|
||||
content: " ·stub";
|
||||
font-size: 0.7rem;
|
||||
color: var(--muted);
|
||||
}}
|
||||
main {{
|
||||
max-width: 52rem;
|
||||
margin: 0 auto;
|
||||
@@ -166,7 +241,10 @@ def render_page(*, title: str, body_html: str, extra_head: str = "") -> str:
|
||||
</head>
|
||||
<body>
|
||||
<header>
|
||||
<h1>MCP Control Plane</h1>
|
||||
<div class="header-top">
|
||||
<h1>MCP Control Plane</h1>
|
||||
{header_badges}
|
||||
</div>
|
||||
<nav>{nav_links}</nav>
|
||||
</header>
|
||||
<main>
|
||||
|
||||
+111
@@ -0,0 +1,111 @@
|
||||
"""Navigation IA for the Phase 1 operator console shell (#638).
|
||||
|
||||
Single source of truth for the console navigation so ``webui/layout.py`` and
|
||||
the ``webui/app.py`` route table stay aligned with epic #631. Read-only: every
|
||||
destination is a GET view or a Phase 1 placeholder. No mutation links.
|
||||
|
||||
Nav groups follow the #631 Phase 1 information architecture: Health, Traffic,
|
||||
Runtime/Sessions, Projects, Inventory, Timeline, Policy (placeholder), and
|
||||
Insights (placeholder). Later-phase surfaces are declared as ``stub`` items and
|
||||
backed by ``STUB_PAGES`` so their nav links resolve to a graceful placeholder
|
||||
instead of a 404.
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
from dataclasses import dataclass
|
||||
|
||||
|
||||
@dataclass(frozen=True)
|
||||
class NavItem:
|
||||
"""A single navigation destination.
|
||||
|
||||
``status`` is ``"live"`` for implemented views and ``"stub"`` for Phase 1
|
||||
placeholders whose backing view lands in a later child issue.
|
||||
"""
|
||||
|
||||
href: str
|
||||
label: str
|
||||
status: str = "live"
|
||||
|
||||
|
||||
@dataclass(frozen=True)
|
||||
class NavGroup:
|
||||
label: str
|
||||
items: tuple[NavItem, ...]
|
||||
|
||||
|
||||
NAV_GROUPS: tuple[NavGroup, ...] = (
|
||||
NavGroup("Health", (
|
||||
NavItem("/health", "Liveness"),
|
||||
)),
|
||||
NavGroup("Traffic", (
|
||||
NavItem("/queue", "Queue"),
|
||||
NavItem("/leases", "Leases"),
|
||||
NavItem("/actions", "Actions"),
|
||||
)),
|
||||
NavGroup("Runtime/Sessions", (
|
||||
NavItem("/runtime", "Runtime health"),
|
||||
NavItem("/sessions", "Sessions", "stub"),
|
||||
)),
|
||||
NavGroup("Projects", (
|
||||
NavItem("/projects", "Projects"),
|
||||
)),
|
||||
NavGroup("Inventory", (
|
||||
NavItem("/inventory", "Inventory", "stub"),
|
||||
NavItem("/worktrees", "Worktrees"),
|
||||
)),
|
||||
NavGroup("Timeline", (
|
||||
NavItem("/timeline", "Timeline", "stub"),
|
||||
)),
|
||||
NavGroup("Policy", (
|
||||
NavItem("/policy", "Policy", "stub"),
|
||||
NavItem("/prompts", "Prompts"),
|
||||
)),
|
||||
NavGroup("Insights", (
|
||||
NavItem("/insights", "Insights", "stub"),
|
||||
NavItem("/audit", "Audit"),
|
||||
)),
|
||||
)
|
||||
|
||||
|
||||
# Phase 1 placeholder destinations whose backing views land in later child
|
||||
# issues of epic #631. Each maps a path to (title, description). Routes are
|
||||
# registered so nav links resolve to a graceful, read-only stub page.
|
||||
STUB_PAGES: dict[str, tuple[str, str]] = {
|
||||
"/sessions": (
|
||||
"Sessions",
|
||||
"Active session, capability, and role inventory. Backed by the unified "
|
||||
"inventory API (#636) once it lands.",
|
||||
),
|
||||
"/inventory": (
|
||||
"Inventory",
|
||||
"Unified sessions, leases, locks, namespaces, and worktree inventory. "
|
||||
"Backed by the Phase 1 inventory API (#636).",
|
||||
),
|
||||
"/timeline": (
|
||||
"Timeline",
|
||||
"Workflow event timeline across issues and PRs. A later Phase 1 surface.",
|
||||
),
|
||||
"/policy": (
|
||||
"Policy",
|
||||
"Capability and role policy surface. Placeholder until a later phase.",
|
||||
),
|
||||
"/insights": (
|
||||
"Insights",
|
||||
"Aggregate operational insights and trends. Placeholder until a later "
|
||||
"phase.",
|
||||
),
|
||||
}
|
||||
|
||||
|
||||
def iter_nav_items():
|
||||
"""Yield every ``NavItem`` across all groups in declared order."""
|
||||
for group in NAV_GROUPS:
|
||||
for item in group.items:
|
||||
yield item
|
||||
|
||||
|
||||
def nav_hrefs() -> tuple[str, ...]:
|
||||
"""Return every navigation href in declared order."""
|
||||
return tuple(item.href for item in iter_nav_items())
|
||||
Reference in New Issue
Block a user