Compare commits
3
Commits
| Author | SHA1 | Date | |
|---|---|---|---|
|
|
db248af8f1 | ||
|
|
e62c9f07ae | ||
|
|
6f7b57cffb |
+26
-3
@@ -36,8 +36,11 @@ Optional environment variables:
|
||||
|------|-------------|
|
||||
| `/` | Home / operator overview |
|
||||
| `/health` | JSON liveness (`status`, `service`, `mode`, `timestamp`) |
|
||||
| `/projects` | Stub — registry (#427) |
|
||||
| `/prompts` | Stub — prompt library (#428) |
|
||||
| `/projects` | Project registry list (#427) |
|
||||
| `/projects/{id}` | Project detail + onboarding checklist |
|
||||
| `/api/projects` | JSON registry export |
|
||||
| `/prompts` | Prompt library with per-prompt copy buttons (#428) |
|
||||
| `/api/prompts` | JSON prompt export with workflow hashes |
|
||||
| `/runtime` | Stub — MCP runtime health (#430) |
|
||||
| `/audit` | Stub — report audit paste (#431) |
|
||||
| `/worktrees` | Stub — hygiene dashboard (#432) |
|
||||
@@ -46,8 +49,28 @@ Optional environment variables:
|
||||
All routes are GET-only. POST/PUT/PATCH/DELETE return `405` with
|
||||
`read-only-mvp`.
|
||||
|
||||
## Project registry (#427)
|
||||
|
||||
Versioned registry file: `webui/data/projects.registry.json` (schema version `1`).
|
||||
|
||||
Override path with `WEBUI_PROJECT_REGISTRY` when operators keep a machine-local
|
||||
copy outside git. The registry stores repo identity, remotes, profile names,
|
||||
workflow/schema path references, and onboarding checklist steps — never tokens
|
||||
or credentials.
|
||||
|
||||
Seed entry: **Gitea-Tools** on `https://gitea.prgs.cc` with `prgs-author`,
|
||||
`prgs-reviewer`, and `prgs-reconciler` profiles.
|
||||
|
||||
## Prompt library (#428)
|
||||
|
||||
Prompts are generated at load time from canonical workflow files under
|
||||
`skills/llm-project-workflow/workflows/`. SHA-256 hashes are computed from
|
||||
`WEBUI_REPO_ROOT` (defaults to the repository root). Prompt bodies are short
|
||||
copy/paste starters; canonical workflow files remain the only full policy
|
||||
source.
|
||||
|
||||
## Tests
|
||||
|
||||
```bash
|
||||
pytest tests/test_webui_skeleton.py -q
|
||||
pytest tests/test_webui_skeleton.py tests/test_webui_project_registry.py tests/test_webui_prompt_library.py -q
|
||||
```
|
||||
@@ -0,0 +1,111 @@
|
||||
"""Tests for web UI project registry (#427)."""
|
||||
import json
|
||||
import sys
|
||||
import tempfile
|
||||
import unittest
|
||||
from pathlib import Path
|
||||
|
||||
sys.path.insert(0, str(Path(__file__).resolve().parent.parent))
|
||||
|
||||
from starlette.testclient import TestClient
|
||||
|
||||
from webui.app import create_app
|
||||
from webui.project_registry import (
|
||||
default_registry_path,
|
||||
load_registry,
|
||||
project_to_dict,
|
||||
)
|
||||
|
||||
|
||||
class TestProjectRegistryLoader(unittest.TestCase):
|
||||
def test_default_registry_loads_gitea_tools(self):
|
||||
registry = load_registry()
|
||||
self.assertEqual(registry.version, 1)
|
||||
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.remote_host, "https://gitea.prgs.cc")
|
||||
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):
|
||||
payload = {
|
||||
"version": 1,
|
||||
"projects": [
|
||||
{
|
||||
"id": "bad",
|
||||
"repo_name": "Bad",
|
||||
"gitea_owner": "Org",
|
||||
"remote_host": "https://gitea.example.invalid",
|
||||
"default_branch": "main",
|
||||
"local_checkout_path": ".",
|
||||
"profiles": {
|
||||
"author": "a",
|
||||
"reviewer": "r",
|
||||
"reconciler": "c",
|
||||
},
|
||||
"workflow_paths": {"skill": "skills/x.md"},
|
||||
"api_token": "secret",
|
||||
}
|
||||
],
|
||||
}
|
||||
with tempfile.NamedTemporaryFile("w", suffix=".json", delete=False) as handle:
|
||||
json.dump(payload, handle)
|
||||
path = Path(handle.name)
|
||||
try:
|
||||
with self.assertRaises(ValueError):
|
||||
load_registry(path)
|
||||
finally:
|
||||
path.unlink(missing_ok=True)
|
||||
|
||||
def test_default_registry_path_points_at_packaged_data(self):
|
||||
path = default_registry_path()
|
||||
self.assertTrue(path.name == "projects.registry.json")
|
||||
self.assertTrue(path.parent.name == "data")
|
||||
|
||||
|
||||
class TestProjectRegistryRoutes(unittest.TestCase):
|
||||
def setUp(self):
|
||||
self.client = TestClient(create_app())
|
||||
|
||||
def test_projects_page_lists_gitea_tools(self):
|
||||
response = self.client.get("/projects")
|
||||
self.assertEqual(response.status_code, 200)
|
||||
self.assertIn("Gitea-Tools", response.text)
|
||||
self.assertIn("Scaled-Tech-Consulting", response.text)
|
||||
self.assertIn("prgs-author", response.text)
|
||||
self.assertNotIn("child issue", response.text.lower())
|
||||
|
||||
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)
|
||||
|
||||
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):
|
||||
response = self.client.get("/api/projects")
|
||||
self.assertEqual(response.status_code, 200)
|
||||
data = response.json()
|
||||
self.assertEqual(data["version"], 1)
|
||||
self.assertEqual(len(data["projects"]), 1)
|
||||
self.assertEqual(data["projects"][0]["id"], "gitea-tools")
|
||||
self.assertIn("onboarding_checklist", data["projects"][0])
|
||||
|
||||
def test_project_to_dict_is_json_safe(self):
|
||||
registry = load_registry()
|
||||
encoded = json.dumps(project_to_dict(registry.projects[0]))
|
||||
self.assertIn("gitea-tools", encoded)
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
unittest.main()
|
||||
@@ -0,0 +1,90 @@
|
||||
"""Tests for web UI prompt library (#428)."""
|
||||
import json
|
||||
import sys
|
||||
import unittest
|
||||
from pathlib import Path
|
||||
|
||||
sys.path.insert(0, str(Path(__file__).resolve().parent.parent))
|
||||
|
||||
from starlette.testclient import TestClient
|
||||
|
||||
from webui.app import create_app
|
||||
from webui.prompt_library import find_prompt, library_to_dict, load_prompt_library, prompt_to_dict
|
||||
|
||||
REQUIRED_PROMPT_SLUGS = frozenset({
|
||||
"review-pr",
|
||||
"work-issue",
|
||||
"create-issue",
|
||||
"comment-issue",
|
||||
"cleanup",
|
||||
"audit",
|
||||
"onboarding",
|
||||
})
|
||||
|
||||
|
||||
class TestPromptLibraryLoader(unittest.TestCase):
|
||||
def test_library_loads_required_prompts(self):
|
||||
entries = load_prompt_library()
|
||||
slugs = {entry.slug for entry in entries}
|
||||
self.assertEqual(slugs, REQUIRED_PROMPT_SLUGS)
|
||||
|
||||
def test_workflow_hashes_present(self):
|
||||
review = find_prompt("review-pr")
|
||||
self.assertIsNotNone(review)
|
||||
assert review is not None
|
||||
self.assertTrue(review.workflow_hash)
|
||||
self.assertEqual(len(review.workflow_hash), 64)
|
||||
|
||||
def test_prompt_text_is_short(self):
|
||||
for entry in load_prompt_library():
|
||||
self.assertLess(len(entry.prompt_text), 400)
|
||||
self.assertNotIn("Do not improvise around the gates", entry.prompt_text)
|
||||
|
||||
|
||||
class TestPromptLibraryRoutes(unittest.TestCase):
|
||||
def setUp(self):
|
||||
self.client = TestClient(create_app())
|
||||
|
||||
def test_prompts_page_lists_all_entries(self):
|
||||
response = self.client.get("/prompts")
|
||||
self.assertEqual(response.status_code, 200)
|
||||
for label in (
|
||||
"Review PR",
|
||||
"Work issue",
|
||||
"Create issue",
|
||||
"Comment on issue",
|
||||
"Post-merge cleanup",
|
||||
"Reconciliation audit",
|
||||
"Project onboarding",
|
||||
):
|
||||
self.assertIn(label, response.text)
|
||||
self.assertIn("Copy prompt", response.text)
|
||||
self.assertIn("sha256:", response.text)
|
||||
self.assertNotIn("child issue", response.text.lower())
|
||||
|
||||
def test_prompt_detail_route(self):
|
||||
response = self.client.get("/prompts/work-issue")
|
||||
self.assertEqual(response.status_code, 200)
|
||||
self.assertIn("work-issue.md", response.text)
|
||||
self.assertIn("Copy prompt", response.text)
|
||||
|
||||
def test_prompt_detail_404(self):
|
||||
self.assertEqual(self.client.get("/prompts/missing").status_code, 404)
|
||||
|
||||
def test_api_prompts_json(self):
|
||||
response = self.client.get("/api/prompts")
|
||||
self.assertEqual(response.status_code, 200)
|
||||
data = response.json()
|
||||
self.assertEqual(data["count"], 7)
|
||||
review = next(item for item in data["prompts"] if item["slug"] == "review-pr")
|
||||
self.assertIn("workflow_hash", review)
|
||||
self.assertIn("review-merge-pr.md", review["workflow_path"])
|
||||
|
||||
def test_prompt_to_dict_roundtrip(self):
|
||||
entry = load_prompt_library()[0]
|
||||
encoded = json.dumps(prompt_to_dict(entry))
|
||||
self.assertIn("workflow_path", encoded)
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
unittest.main()
|
||||
@@ -30,12 +30,22 @@ class TestWebuiSkeleton(unittest.TestCase):
|
||||
self.assertIn("Read-only MVP", response.text)
|
||||
|
||||
def test_route_stubs_render(self):
|
||||
for path in ("/projects", "/prompts", "/runtime", "/audit"):
|
||||
for path in ("/runtime", "/audit"):
|
||||
with self.subTest(path=path):
|
||||
response = self.client.get(path)
|
||||
self.assertEqual(response.status_code, 200)
|
||||
self.assertIn("child issue", response.text.lower())
|
||||
|
||||
def test_prompts_is_implemented(self):
|
||||
response = self.client.get("/prompts")
|
||||
self.assertEqual(response.status_code, 200)
|
||||
self.assertIn("Prompt library", response.text)
|
||||
|
||||
def test_projects_is_implemented(self):
|
||||
response = self.client.get("/projects")
|
||||
self.assertEqual(response.status_code, 200)
|
||||
self.assertIn("Gitea-Tools", response.text)
|
||||
|
||||
def test_extra_stub_routes(self):
|
||||
for path in ("/worktrees", "/leases"):
|
||||
with self.subTest(path=path):
|
||||
|
||||
+57
-8
@@ -10,6 +10,10 @@ from starlette.responses import HTMLResponse, JSONResponse, Response
|
||||
from starlette.routing import Route
|
||||
|
||||
from webui.layout import render_page
|
||||
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
|
||||
|
||||
_READ_ONLY_METHODS = frozenset({"GET", "HEAD", "OPTIONS"})
|
||||
|
||||
@@ -49,17 +53,58 @@ async def health(_request: Request) -> JSONResponse:
|
||||
|
||||
|
||||
async def projects(_request: Request) -> HTMLResponse:
|
||||
return _stub_page(
|
||||
"Projects",
|
||||
"Project registry and onboarding model will list configured repos and profiles.",
|
||||
)
|
||||
registry = load_registry()
|
||||
return HTMLResponse(render_projects_list(registry))
|
||||
|
||||
|
||||
async def project_detail(request: Request) -> HTMLResponse:
|
||||
project_id = request.path_params["project_id"]
|
||||
registry = load_registry()
|
||||
project = find_project(registry, project_id)
|
||||
if project is None:
|
||||
return HTMLResponse(
|
||||
render_page(
|
||||
title="Project not found",
|
||||
body_html=(
|
||||
"<h2>Project not found</h2>"
|
||||
f"<p>No registry entry for <code>{project_id}</code>.</p>"
|
||||
'<p><a href="/projects">← All projects</a></p>'
|
||||
),
|
||||
),
|
||||
status_code=404,
|
||||
)
|
||||
return HTMLResponse(render_project_detail(project))
|
||||
|
||||
|
||||
async def api_projects(_request: Request) -> JSONResponse:
|
||||
registry = load_registry()
|
||||
return JSONResponse(registry_to_dict(registry))
|
||||
|
||||
|
||||
async def prompts(_request: Request) -> HTMLResponse:
|
||||
return _stub_page(
|
||||
"Prompts",
|
||||
"Prompt library will surface canonical workflows from skills/llm-project-workflow/.",
|
||||
)
|
||||
return HTMLResponse(render_prompts_page())
|
||||
|
||||
|
||||
async def prompt_detail(request: Request) -> HTMLResponse:
|
||||
prompt_id = request.path_params["prompt_id"]
|
||||
prompt = find_prompt(prompt_id)
|
||||
if prompt is None:
|
||||
return HTMLResponse(
|
||||
render_page(
|
||||
title="Prompt not found",
|
||||
body_html=(
|
||||
"<h2>Prompt not found</h2>"
|
||||
f"<p>No library entry for <code>{prompt_id}</code>.</p>"
|
||||
'<p><a href="/prompts">← All prompts</a></p>'
|
||||
),
|
||||
),
|
||||
status_code=404,
|
||||
)
|
||||
return HTMLResponse(render_prompt_detail(prompt))
|
||||
|
||||
|
||||
async def api_prompts(_request: Request) -> JSONResponse:
|
||||
return JSONResponse(library_to_dict())
|
||||
|
||||
|
||||
async def runtime(_request: Request) -> HTMLResponse:
|
||||
@@ -107,7 +152,11 @@ def create_app() -> Starlette:
|
||||
Route("/", home, methods=["GET"]),
|
||||
Route("/health", health, methods=["GET"]),
|
||||
Route("/projects", projects, methods=["GET"]),
|
||||
Route("/projects/{project_id}", project_detail, methods=["GET"]),
|
||||
Route("/api/projects", api_projects, methods=["GET"]),
|
||||
Route("/prompts", prompts, methods=["GET"]),
|
||||
Route("/prompts/{prompt_id}", prompt_detail, methods=["GET"]),
|
||||
Route("/api/prompts", api_prompts, methods=["GET"]),
|
||||
Route("/runtime", runtime, methods=["GET"]),
|
||||
Route("/audit", audit, methods=["GET"]),
|
||||
Route("/worktrees", worktrees, methods=["GET"]),
|
||||
|
||||
@@ -0,0 +1,49 @@
|
||||
{
|
||||
"version": 1,
|
||||
"projects": [
|
||||
{
|
||||
"id": "gitea-tools",
|
||||
"repo_name": "Gitea-Tools",
|
||||
"gitea_owner": "Scaled-Tech-Consulting",
|
||||
"remote_host": "https://gitea.prgs.cc",
|
||||
"default_branch": "master",
|
||||
"local_checkout_path": ".",
|
||||
"profiles": {
|
||||
"author": "prgs-author",
|
||||
"reviewer": "prgs-reviewer",
|
||||
"reconciler": "prgs-reconciler"
|
||||
},
|
||||
"workflow_paths": {
|
||||
"skill": "skills/llm-project-workflow/SKILL.md",
|
||||
"work_issue": "skills/llm-project-workflow/workflows/work-issue.md",
|
||||
"review_merge": "skills/llm-project-workflow/workflows/review-merge-pr.md"
|
||||
},
|
||||
"schema_paths": {
|
||||
"mcp_config_v2": "gitea-mcp.v2-contexts.example.json",
|
||||
"mcp_config_v1": "gitea-mcp.example.json"
|
||||
},
|
||||
"onboarding_checklist": [
|
||||
{
|
||||
"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."
|
||||
},
|
||||
{
|
||||
"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."
|
||||
},
|
||||
{
|
||||
"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."
|
||||
},
|
||||
{
|
||||
"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."
|
||||
}
|
||||
]
|
||||
}
|
||||
]
|
||||
}
|
||||
+60
-1
@@ -18,7 +18,7 @@ MVP_NOTICE = (
|
||||
)
|
||||
|
||||
|
||||
def render_page(*, title: str, body_html: str) -> str:
|
||||
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
|
||||
)
|
||||
@@ -87,7 +87,66 @@ def render_page(*, title: str, body_html: str) -> str:
|
||||
padding-left: 0.85rem;
|
||||
margin: 1rem 0;
|
||||
}}
|
||||
.meta {{ font-size: 0.85rem; }}
|
||||
table.registry, table.detail {{
|
||||
width: 100%;
|
||||
border-collapse: collapse;
|
||||
margin: 1rem 0;
|
||||
font-size: 0.9rem;
|
||||
}}
|
||||
table.registry th, table.registry td,
|
||||
table.detail th, table.detail td {{
|
||||
text-align: left;
|
||||
padding: 0.45rem 0.6rem;
|
||||
border-bottom: 1px solid var(--border);
|
||||
}}
|
||||
table.registry th, table.detail th {{
|
||||
color: var(--muted);
|
||||
font-weight: 500;
|
||||
}}
|
||||
code {{
|
||||
font-family: ui-monospace, monospace;
|
||||
font-size: 0.85em;
|
||||
color: var(--text);
|
||||
}}
|
||||
ol.checklist {{
|
||||
padding-left: 1.25rem;
|
||||
margin: 0.5rem 0 1.5rem;
|
||||
}}
|
||||
ol.checklist li {{ margin-bottom: 0.85rem; }}
|
||||
ol.checklist p {{ margin: 0.25rem 0 0; font-size: 0.9rem; }}
|
||||
.prompt-card {{
|
||||
margin: 1.25rem 0 1.75rem;
|
||||
padding: 1rem 1.1rem;
|
||||
border: 1px solid var(--border);
|
||||
border-radius: 8px;
|
||||
background: var(--surface);
|
||||
}}
|
||||
.prompt-card h3 {{ margin: 0 0 0.5rem; font-size: 1.05rem; }}
|
||||
pre.prompt-text {{
|
||||
white-space: pre-wrap;
|
||||
word-break: break-word;
|
||||
margin: 0.75rem 0;
|
||||
padding: 0.75rem 0.85rem;
|
||||
border-radius: 6px;
|
||||
background: var(--bg);
|
||||
border: 1px solid var(--border);
|
||||
font-size: 0.9rem;
|
||||
color: var(--text);
|
||||
}}
|
||||
.copy-btn {{
|
||||
background: var(--accent);
|
||||
color: #0b1219;
|
||||
border: none;
|
||||
border-radius: 6px;
|
||||
padding: 0.4rem 0.85rem;
|
||||
font-size: 0.85rem;
|
||||
cursor: pointer;
|
||||
}}
|
||||
.copy-btn:hover {{ filter: brightness(1.08); }}
|
||||
.muted {{ color: var(--muted); }}
|
||||
</style>
|
||||
{extra_head}
|
||||
</head>
|
||||
<body>
|
||||
<header>
|
||||
|
||||
@@ -0,0 +1,196 @@
|
||||
"""Load and validate the web UI project registry (#427)."""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import json
|
||||
import os
|
||||
from dataclasses import dataclass
|
||||
from pathlib import Path
|
||||
from typing import Any
|
||||
|
||||
_FORBIDDEN_EXACT_KEYS = frozenset({
|
||||
"token",
|
||||
"password",
|
||||
"secret",
|
||||
"credential",
|
||||
"auth",
|
||||
"api_key",
|
||||
"api-key",
|
||||
})
|
||||
_FORBIDDEN_KEY_PREFIXES = ("auth_", "api_key_", "api-key_")
|
||||
_FORBIDDEN_KEY_SUFFIXES = ("_token", "_secret", "_password", "_credential", "_auth")
|
||||
|
||||
|
||||
def _is_forbidden_key(key: str) -> bool:
|
||||
lowered = key.lower()
|
||||
if lowered in _FORBIDDEN_EXACT_KEYS:
|
||||
return True
|
||||
return (
|
||||
lowered.startswith(_FORBIDDEN_KEY_PREFIXES)
|
||||
or lowered.endswith(_FORBIDDEN_KEY_SUFFIXES)
|
||||
)
|
||||
|
||||
_REQUIRED_PROJECT_FIELDS = (
|
||||
"id",
|
||||
"repo_name",
|
||||
"gitea_owner",
|
||||
"remote_host",
|
||||
"default_branch",
|
||||
"local_checkout_path",
|
||||
"profiles",
|
||||
"workflow_paths",
|
||||
)
|
||||
|
||||
_REQUIRED_PROFILE_ROLES = ("author", "reviewer", "reconciler")
|
||||
|
||||
|
||||
@dataclass(frozen=True)
|
||||
class OnboardingStep:
|
||||
id: str
|
||||
title: str
|
||||
description: str
|
||||
|
||||
|
||||
@dataclass(frozen=True)
|
||||
class ProjectRecord:
|
||||
id: str
|
||||
repo_name: str
|
||||
gitea_owner: str
|
||||
remote_host: str
|
||||
default_branch: str
|
||||
local_checkout_path: str
|
||||
profiles: dict[str, str]
|
||||
workflow_paths: dict[str, str]
|
||||
schema_paths: dict[str, str]
|
||||
onboarding_checklist: tuple[OnboardingStep, ...]
|
||||
|
||||
|
||||
@dataclass(frozen=True)
|
||||
class ProjectRegistry:
|
||||
version: int
|
||||
projects: tuple[ProjectRecord, ...]
|
||||
source_path: Path
|
||||
|
||||
|
||||
def default_registry_path() -> Path:
|
||||
override = os.environ.get("WEBUI_PROJECT_REGISTRY", "").strip()
|
||||
if override:
|
||||
return Path(override).expanduser().resolve()
|
||||
return (Path(__file__).resolve().parent / "data" / "projects.registry.json").resolve()
|
||||
|
||||
|
||||
def _reject_credential_keys(obj: Any, *, path: str = "") -> None:
|
||||
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 ValueError(f"registry must not store credentials ({key_path})")
|
||||
_reject_credential_keys(value, path=key_path)
|
||||
elif isinstance(obj, list):
|
||||
for index, item in enumerate(obj):
|
||||
_reject_credential_keys(item, path=f"{path}[{index}]")
|
||||
|
||||
|
||||
def _parse_onboarding(raw: list[dict[str, Any]] | None) -> tuple[OnboardingStep, ...]:
|
||||
if not raw:
|
||||
return ()
|
||||
steps: list[OnboardingStep] = []
|
||||
for item in raw:
|
||||
steps.append(
|
||||
OnboardingStep(
|
||||
id=str(item["id"]),
|
||||
title=str(item["title"]),
|
||||
description=str(item["description"]),
|
||||
)
|
||||
)
|
||||
return tuple(steps)
|
||||
|
||||
|
||||
def _parse_project(raw: dict[str, Any]) -> ProjectRecord:
|
||||
missing = [field for field in _REQUIRED_PROJECT_FIELDS if field not in raw]
|
||||
if missing:
|
||||
raise ValueError(f"project missing required fields: {', '.join(missing)}")
|
||||
|
||||
profiles = raw["profiles"]
|
||||
if not isinstance(profiles, dict):
|
||||
raise ValueError("profiles must be an object")
|
||||
for role in _REQUIRED_PROFILE_ROLES:
|
||||
if role not in profiles or not profiles[role]:
|
||||
raise ValueError(f"profiles.{role} is required")
|
||||
|
||||
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")
|
||||
|
||||
schema_paths = raw.get("schema_paths") or {}
|
||||
if not isinstance(schema_paths, dict):
|
||||
raise ValueError("schema_paths must be an object when present")
|
||||
|
||||
return ProjectRecord(
|
||||
id=str(raw["id"]),
|
||||
repo_name=str(raw["repo_name"]),
|
||||
gitea_owner=str(raw["gitea_owner"]),
|
||||
remote_host=str(raw["remote_host"]),
|
||||
default_branch=str(raw["default_branch"]),
|
||||
local_checkout_path=str(raw["local_checkout_path"]),
|
||||
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")),
|
||||
)
|
||||
|
||||
|
||||
def load_registry(path: Path | None = None) -> ProjectRegistry:
|
||||
"""Load the versioned project registry from disk."""
|
||||
source = (path or default_registry_path()).resolve()
|
||||
raw_text = source.read_text(encoding="utf-8")
|
||||
payload = json.loads(raw_text)
|
||||
if not isinstance(payload, dict):
|
||||
raise ValueError("registry root must be an object")
|
||||
|
||||
version = payload.get("version")
|
||||
if version != 1:
|
||||
raise ValueError(f"unsupported registry version: {version!r}")
|
||||
|
||||
_reject_credential_keys(payload)
|
||||
|
||||
projects_raw = payload.get("projects")
|
||||
if not isinstance(projects_raw, list) or not projects_raw:
|
||||
raise ValueError("projects must be a non-empty array")
|
||||
|
||||
projects = tuple(_parse_project(item) for item in projects_raw)
|
||||
return ProjectRegistry(version=version, projects=projects, source_path=source)
|
||||
|
||||
|
||||
def project_to_dict(project: ProjectRecord) -> dict[str, Any]:
|
||||
"""Serialize a project for JSON API responses."""
|
||||
return {
|
||||
"id": project.id,
|
||||
"repo_name": project.repo_name,
|
||||
"gitea_owner": project.gitea_owner,
|
||||
"remote_host": project.remote_host,
|
||||
"default_branch": project.default_branch,
|
||||
"local_checkout_path": project.local_checkout_path,
|
||||
"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}
|
||||
for step in project.onboarding_checklist
|
||||
],
|
||||
}
|
||||
|
||||
|
||||
def registry_to_dict(registry: ProjectRegistry) -> dict[str, Any]:
|
||||
return {
|
||||
"version": registry.version,
|
||||
"source_path": str(registry.source_path),
|
||||
"projects": [project_to_dict(project) for project in registry.projects],
|
||||
}
|
||||
|
||||
|
||||
def find_project(registry: ProjectRegistry, project_id: str) -> ProjectRecord | None:
|
||||
for project in registry.projects:
|
||||
if project.id == project_id:
|
||||
return project
|
||||
return None
|
||||
@@ -0,0 +1,93 @@
|
||||
"""HTML views for project registry pages (#427)."""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import html
|
||||
|
||||
from webui.layout import render_page
|
||||
from webui.project_registry import ProjectRecord, ProjectRegistry
|
||||
|
||||
|
||||
def _escape(text: str) -> str:
|
||||
return html.escape(text, quote=True)
|
||||
|
||||
|
||||
def render_projects_list(registry: ProjectRegistry) -> str:
|
||||
rows = []
|
||||
for project in registry.projects:
|
||||
rows.append(
|
||||
"<tr>"
|
||||
f"<td><a href=\"/projects/{_escape(project.id)}\">{_escape(project.repo_name)}</a></td>"
|
||||
f"<td>{_escape(project.gitea_owner)}</td>"
|
||||
f"<td>{_escape(project.remote_host)}</td>"
|
||||
f"<td>{_escape(project.default_branch)}</td>"
|
||||
f"<td><code>{_escape(project.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>"
|
||||
"</tr></thead>"
|
||||
f"<tbody>{''.join(rows)}</tbody></table>"
|
||||
)
|
||||
body = (
|
||||
"<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"{table}"
|
||||
"<p><a href=\"/api/projects\">JSON API</a></p>"
|
||||
)
|
||||
return render_page(title="Projects", body_html=body)
|
||||
|
||||
|
||||
def render_project_detail(project: ProjectRecord) -> str:
|
||||
profile_rows = "".join(
|
||||
f"<tr><th>{_escape(role)}</th><td><code>{_escape(name)}</code></td></tr>"
|
||||
for role, name in project.profiles.items()
|
||||
)
|
||||
workflow_rows = "".join(
|
||||
f"<tr><th>{_escape(key)}</th><td><code>{_escape(path)}</code></td></tr>"
|
||||
for key, path in project.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()
|
||||
)
|
||||
checklist_items = []
|
||||
for index, step in enumerate(project.onboarding_checklist, start=1):
|
||||
checklist_items.append(
|
||||
"<li>"
|
||||
f"<strong>{index}. {_escape(step.title)}</strong>"
|
||||
f"<p>{_escape(step.description)}</p>"
|
||||
"</li>"
|
||||
)
|
||||
checklist_html = (
|
||||
"<ol class=\"checklist\">" + "".join(checklist_items) + "</ol>"
|
||||
if checklist_items
|
||||
else "<p>No onboarding steps defined.</p>"
|
||||
)
|
||||
body = (
|
||||
f"<h2>{_escape(project.repo_name)}</h2>"
|
||||
"<p><a href=\"/projects\">← All projects</a></p>"
|
||||
"<h3>Identity</h3>"
|
||||
"<table class=\"detail\">"
|
||||
f"<tr><th>Registry id</th><td><code>{_escape(project.id)}</code></td></tr>"
|
||||
f"<tr><th>Gitea owner</th><td>{_escape(project.gitea_owner)}</td></tr>"
|
||||
f"<tr><th>Remote host</th><td>{_escape(project.remote_host)}</td></tr>"
|
||||
f"<tr><th>Default branch</th><td><code>{_escape(project.default_branch)}</code></td></tr>"
|
||||
f"<tr><th>Local checkout</th><td><code>{_escape(project.local_checkout_path)}</code></td></tr>"
|
||||
"</table>"
|
||||
"<h3>Profiles</h3>"
|
||||
f"<table class=\"detail\">{profile_rows}</table>"
|
||||
"<h3>Workflow paths</h3>"
|
||||
f"<table class=\"detail\">{workflow_rows}</table>"
|
||||
"<h3>Schema paths</h3>"
|
||||
f"<table class=\"detail\">{schema_rows}</table>"
|
||||
"<h3>Onboarding checklist</h3>"
|
||||
"<p class=\"meta\">Read-only MVP — complete these steps outside the UI.</p>"
|
||||
f"{checklist_html}"
|
||||
)
|
||||
return render_page(title=project.repo_name, body_html=body)
|
||||
@@ -0,0 +1,197 @@
|
||||
"""Canonical workflow prompt library for the internal web UI (#428)."""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import hashlib
|
||||
import os
|
||||
import re
|
||||
from dataclasses import dataclass
|
||||
from pathlib import Path
|
||||
from typing import Any
|
||||
|
||||
_WORKFLOW_ROOT = Path("skills/llm-project-workflow/workflows")
|
||||
|
||||
_DEFAULT_PROMPT_RE = re.compile(
|
||||
r"\*\*Default task prompt:\*\*\s*\n+>\s*(.+?)(?=\n\n|\nDo not improvise)",
|
||||
re.DOTALL,
|
||||
)
|
||||
_FRONTMATTER_TASK_MODE_RE = re.compile(r"^task_mode:\s*(\S+)", re.MULTILINE)
|
||||
|
||||
|
||||
@dataclass(frozen=True)
|
||||
class PromptEntry:
|
||||
slug: str
|
||||
label: str
|
||||
prompt_text: str
|
||||
workflow_path: str
|
||||
task_mode: str | None
|
||||
workflow_hash: str | None
|
||||
source_note: str
|
||||
|
||||
|
||||
def _repo_root() -> Path:
|
||||
override = (os.environ.get("WEBUI_REPO_ROOT") or "").strip()
|
||||
if override:
|
||||
return Path(override).resolve()
|
||||
return Path(__file__).resolve().parent.parent
|
||||
|
||||
|
||||
def _workflow_file(path: str) -> Path:
|
||||
return _repo_root() / path
|
||||
|
||||
|
||||
def _sha256_hex(content: str) -> str:
|
||||
return hashlib.sha256(content.encode("utf-8")).hexdigest()
|
||||
|
||||
|
||||
def _read_workflow(path: str) -> tuple[str, str]:
|
||||
file_path = _workflow_file(path)
|
||||
text = file_path.read_text(encoding="utf-8")
|
||||
return text, _sha256_hex(text)
|
||||
|
||||
|
||||
def _extract_default_prompt(markdown: str) -> str | None:
|
||||
match = _DEFAULT_PROMPT_RE.search(markdown)
|
||||
if not match:
|
||||
return None
|
||||
lines = [line.strip() for line in match.group(1).splitlines()]
|
||||
return " ".join(line for line in lines if line)
|
||||
|
||||
|
||||
def _extract_task_mode(markdown: str) -> str | None:
|
||||
match = _FRONTMATTER_TASK_MODE_RE.search(markdown)
|
||||
return match.group(1) if match else None
|
||||
|
||||
|
||||
def _entry_from_workflow(
|
||||
*,
|
||||
slug: str,
|
||||
label: str,
|
||||
workflow_path: str,
|
||||
prompt_override: str | None = None,
|
||||
source_note: str = "",
|
||||
) -> PromptEntry:
|
||||
markdown, digest = _read_workflow(workflow_path)
|
||||
prompt_text = prompt_override or _extract_default_prompt(markdown)
|
||||
if not prompt_text:
|
||||
raise ValueError(f"No default task prompt found in {workflow_path}")
|
||||
return PromptEntry(
|
||||
slug=slug,
|
||||
label=label,
|
||||
prompt_text=prompt_text,
|
||||
workflow_path=workflow_path,
|
||||
task_mode=_extract_task_mode(markdown),
|
||||
workflow_hash=digest,
|
||||
source_note=source_note,
|
||||
)
|
||||
|
||||
|
||||
def _static_entry(
|
||||
*,
|
||||
slug: str,
|
||||
label: str,
|
||||
prompt_text: str,
|
||||
workflow_path: str,
|
||||
source_note: str,
|
||||
) -> PromptEntry:
|
||||
path = _workflow_file(workflow_path)
|
||||
digest = _sha256_hex(path.read_text(encoding="utf-8")) if path.is_file() else None
|
||||
markdown = path.read_text(encoding="utf-8") if path.is_file() else ""
|
||||
return PromptEntry(
|
||||
slug=slug,
|
||||
label=label,
|
||||
prompt_text=prompt_text,
|
||||
workflow_path=workflow_path,
|
||||
task_mode=_extract_task_mode(markdown) if markdown else None,
|
||||
workflow_hash=digest,
|
||||
source_note=source_note,
|
||||
)
|
||||
|
||||
|
||||
def load_prompt_library() -> tuple[PromptEntry, ...]:
|
||||
"""Load operator prompts derived from canonical workflows."""
|
||||
entries = (
|
||||
_entry_from_workflow(
|
||||
slug="review-pr",
|
||||
label="Review PR",
|
||||
workflow_path=str(_WORKFLOW_ROOT / "review-merge-pr.md"),
|
||||
),
|
||||
_entry_from_workflow(
|
||||
slug="work-issue",
|
||||
label="Work issue",
|
||||
workflow_path=str(_WORKFLOW_ROOT / "work-issue.md"),
|
||||
),
|
||||
_entry_from_workflow(
|
||||
slug="create-issue",
|
||||
label="Create issue",
|
||||
workflow_path=str(_WORKFLOW_ROOT / "create-issue.md"),
|
||||
),
|
||||
_static_entry(
|
||||
slug="comment-issue",
|
||||
label="Comment on issue",
|
||||
workflow_path=str(_WORKFLOW_ROOT / "create-issue.md"),
|
||||
prompt_text=(
|
||||
"Comment on the target Gitea issue only if exact comment_issue "
|
||||
"capability is proven. Load the canonical create-issue workflow "
|
||||
"first and follow §16 (comment-on-existing issue rule). Include "
|
||||
"specific evidence; do not duplicate existing comments."
|
||||
),
|
||||
source_note="Derived from create-issue.md §16; full policy remains in the workflow file.",
|
||||
),
|
||||
_static_entry(
|
||||
slug="cleanup",
|
||||
label="Post-merge cleanup",
|
||||
workflow_path="skills/llm-project-workflow/templates/worktree-cleanup.md",
|
||||
prompt_text=(
|
||||
"Task: clean up branch/worktree for PR #<pr> / issue #<n> after merge. "
|
||||
"Confirm the merge on remote master before any deletion; never "
|
||||
"force-remove a dirty worktree."
|
||||
),
|
||||
source_note="Full cleanup steps live in templates/worktree-cleanup.md.",
|
||||
),
|
||||
_entry_from_workflow(
|
||||
slug="audit",
|
||||
label="Reconciliation audit",
|
||||
workflow_path=str(_WORKFLOW_ROOT / "reconcile-landed-pr.md"),
|
||||
),
|
||||
_static_entry(
|
||||
slug="onboarding",
|
||||
label="Project onboarding",
|
||||
workflow_path="skills/llm-project-workflow/SKILL.md",
|
||||
prompt_text=(
|
||||
"Onboard this repository into the MCP Control Plane: prove identity "
|
||||
"and task capability, configure author/reviewer/reconciler profiles "
|
||||
"in separate namespaces, then complete the checklist at /projects. "
|
||||
"Canonical router: skills/llm-project-workflow/SKILL.md."
|
||||
),
|
||||
source_note="Checklist details live in webui/data/projects.registry.json and /projects.",
|
||||
),
|
||||
)
|
||||
return entries
|
||||
|
||||
|
||||
def find_prompt(slug: str) -> PromptEntry | None:
|
||||
for entry in load_prompt_library():
|
||||
if entry.slug == slug:
|
||||
return entry
|
||||
return None
|
||||
|
||||
|
||||
def prompt_to_dict(entry: PromptEntry) -> dict[str, Any]:
|
||||
return {
|
||||
"slug": entry.slug,
|
||||
"label": entry.label,
|
||||
"prompt_text": entry.prompt_text,
|
||||
"workflow_path": entry.workflow_path,
|
||||
"task_mode": entry.task_mode,
|
||||
"workflow_hash": entry.workflow_hash,
|
||||
"source_note": entry.source_note,
|
||||
}
|
||||
|
||||
|
||||
def library_to_dict() -> dict[str, Any]:
|
||||
entries = load_prompt_library()
|
||||
return {
|
||||
"count": len(entries),
|
||||
"prompts": [prompt_to_dict(entry) for entry in entries],
|
||||
}
|
||||
@@ -0,0 +1,87 @@
|
||||
"""HTML views for the prompt library (#428)."""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import html
|
||||
|
||||
from webui.layout import render_page
|
||||
from webui.prompt_library import PromptEntry, load_prompt_library
|
||||
|
||||
|
||||
def _escape(text: str) -> str:
|
||||
return html.escape(text, quote=True)
|
||||
|
||||
|
||||
def _render_prompt_card(entry: PromptEntry) -> str:
|
||||
hash_short = (
|
||||
f"<code>{_escape(entry.workflow_hash[:12])}</code>"
|
||||
if entry.workflow_hash
|
||||
else "<span class=\"muted\">n/a</span>"
|
||||
)
|
||||
task_mode = (
|
||||
f"<code>{_escape(entry.task_mode)}</code>"
|
||||
if entry.task_mode
|
||||
else "<span class=\"muted\">n/a</span>"
|
||||
)
|
||||
note = (
|
||||
f'<p class="meta">{_escape(entry.source_note)}</p>'
|
||||
if entry.source_note
|
||||
else ""
|
||||
)
|
||||
prompt_id = f"prompt-{entry.slug}"
|
||||
return (
|
||||
f'<section class="prompt-card" id="{_escape(entry.slug)}">'
|
||||
f"<h3>{_escape(entry.label)}</h3>"
|
||||
f'<p class="meta">Workflow: <code>{_escape(entry.workflow_path)}</code> · '
|
||||
f"task_mode: {task_mode} · sha256: {hash_short}</p>"
|
||||
f'<pre class="prompt-text" id="{prompt_id}">{_escape(entry.prompt_text)}</pre>'
|
||||
f'<button type="button" class="copy-btn" data-copy-target="{prompt_id}">'
|
||||
"Copy prompt</button>"
|
||||
f"{note}"
|
||||
"</section>"
|
||||
)
|
||||
|
||||
|
||||
PROMPT_PAGE_SCRIPT = """
|
||||
<script>
|
||||
document.querySelectorAll('.copy-btn').forEach((btn) => {
|
||||
btn.addEventListener('click', async () => {
|
||||
const targetId = btn.getAttribute('data-copy-target');
|
||||
const node = document.getElementById(targetId);
|
||||
if (!node) return;
|
||||
const text = node.textContent || '';
|
||||
try {
|
||||
await navigator.clipboard.writeText(text);
|
||||
const prior = btn.textContent;
|
||||
btn.textContent = 'Copied';
|
||||
setTimeout(() => { btn.textContent = prior; }, 1200);
|
||||
} catch (_err) {
|
||||
btn.textContent = 'Copy failed';
|
||||
}
|
||||
});
|
||||
});
|
||||
</script>
|
||||
"""
|
||||
|
||||
|
||||
def render_prompts_page() -> str:
|
||||
entries = load_prompt_library()
|
||||
cards = "".join(_render_prompt_card(entry) for entry in entries)
|
||||
body = (
|
||||
"<h2>Prompt library</h2>"
|
||||
"<p>Short copy/paste task prompts derived from canonical workflows. "
|
||||
"Full policy remains in the cited workflow files — not duplicated here.</p>"
|
||||
f"{cards}"
|
||||
"<p><a href=\"/api/prompts\">JSON API</a></p>"
|
||||
f"{PROMPT_PAGE_SCRIPT}"
|
||||
)
|
||||
return render_page(title="Prompts", body_html=body)
|
||||
|
||||
|
||||
def render_prompt_detail(entry: PromptEntry) -> str:
|
||||
body = (
|
||||
f"<p><a href=\"/prompts\">← All prompts</a></p>"
|
||||
f"{_render_prompt_card(entry)}"
|
||||
f"{PROMPT_PAGE_SCRIPT}"
|
||||
)
|
||||
return render_page(title=entry.label, body_html=body)
|
||||
Reference in New Issue
Block a user