feat(webui): worker registry and configuration schema (Closes #798)

Add the declarative worker registry that epic #797 makes the source of
truth for the scheduled multi-LLM worker fleet.

Providers and configured workers are modelled as separate entities so a
provider can be listed with no worker configured, and so provider facts
are not copied into every worker record. A worker records provider,
model, project, role, namespace, profile, workflow, schedule, timeout,
enabled state, and scheduler metadata.

Validation fails closed: unknown fields are refused rather than ignored,
so a typo cannot silently disable a timeout; a worker naming an
undeclared provider is rejected; worker ids, provider ids, and
LaunchAgent labels must be unique.

Persistence is atomic (temp file in the same directory, fsync, replace).
Every superseded document is retained as a numbered revision, and
rollback republishes a chosen revision as a new head, so history stays
append-only and a rollback is itself reversible.

The credential-rejection guard is extracted to webui/registry_safety.py
so both registries share one implementation instead of two copies of a
security check; project_registry.py keeps identical behaviour.

Scope: data model, validation, persistence only. No routes, scheduler,
process control, or provider probing - those are #799/#800/#804/#805.
The workers array ships empty because populating it is #808.

Tests: tests/test_webui_worker_registry.py, 44 cases.

Co-Authored-By: Claude Opus 4.8 (1M context) <[email protected]>
This commit is contained in:
2026-07-22 06:10:34 -05:00
co-authored by Claude Opus 4.8
parent 5032965e3a
commit 5463f58933
5 changed files with 1215 additions and 33 deletions
+458
View File
@@ -0,0 +1,458 @@
"""Tests for the worker registry and configuration schema (#798, epic #797)."""
import json
import sys
import tempfile
import unittest
from pathlib import Path
sys.path.insert(0, str(Path(__file__).resolve().parent.parent))
from webui.worker_registry import (
ALLOWED_ROLES,
SCHEMA_VERSION,
RegistryValidationError,
WorkerRegistry,
default_registry_path,
find_provider,
find_worker,
history_dir,
list_revisions,
load_registry,
registry_to_dict,
registry_to_document,
rollback_to_revision,
save_registry,
validate_payload,
worker_to_dict,
workers_for_provider,
)
_EXPECTED_PROVIDER_IDS = ("claude", "grok", "codex", "agy", "kimi-k")
def _provider(provider_id: str = "claude", **overrides) -> dict:
payload = {
"id": provider_id,
"display_name": "Claude",
"vendor": "Anthropic",
"executable": "claude",
"available": True,
"models": ["claude-opus-4-8"],
"notes": "",
}
payload.update(overrides)
return payload
def _worker(worker_id: str = "claude-author", **overrides) -> dict:
payload = {
"id": worker_id,
"display_name": "Claude author",
"provider": "claude",
"model": "claude-opus-4-8",
"project": "gitea-tools",
"role": "author",
"namespace": "gitea-author",
"profile": "prgs-author",
"workflow": "skills/llm-project-workflow/workflows/work-issue.md",
"schedule": {"kind": "cron", "expression": "0 * * * *"},
"timeout_seconds": 3600,
"enabled": True,
"scheduler": {"kind": "launchd", "label": "cc.prgs.claude.author"},
"notes": "",
}
payload.update(overrides)
return payload
def _document(providers=None, workers=None, **overrides) -> dict:
payload = {
"version": SCHEMA_VERSION,
"revision": 1,
"updated_at": "2026-07-22T00:00:00Z",
"providers": providers if providers is not None else [_provider()],
"workers": workers if workers is not None else [_worker()],
}
payload.update(overrides)
return payload
class _TempRegistryCase(unittest.TestCase):
"""Base case giving each test an isolated registry file."""
def setUp(self):
self._tmp = tempfile.TemporaryDirectory()
self.addCleanup(self._tmp.cleanup)
self.path = Path(self._tmp.name) / "workers.registry.json"
def write(self, document: dict) -> Path:
self.path.write_text(json.dumps(document, indent=2) + "\n", encoding="utf-8")
return self.path
def parse(self, document: dict) -> WorkerRegistry:
return validate_payload(document, source_path=self.path)
class TestPackagedRegistry(unittest.TestCase):
"""AC: the declarative registry is the source of truth and ships with the app."""
def test_default_path_points_at_packaged_data(self):
path = default_registry_path()
self.assertEqual(path.name, "workers.registry.json")
self.assertEqual(path.parent.name, "data")
def test_packaged_registry_loads_and_validates(self):
registry = load_registry()
self.assertEqual(registry.version, SCHEMA_VERSION)
self.assertGreaterEqual(registry.revision, 1)
def test_packaged_registry_declares_all_five_providers(self):
registry = load_registry()
self.assertEqual(
tuple(provider.id for provider in registry.providers),
_EXPECTED_PROVIDER_IDS,
)
def test_packaged_registry_carries_no_credentials(self):
raw = default_registry_path().read_text(encoding="utf-8").lower()
for marker in ("token", "password", "secret", "api_key", "credential"):
self.assertNotIn(marker, raw)
class TestSeparateEntities(_TempRegistryCase):
"""AC: providers and configured workers are separate entities."""
def test_provider_may_exist_with_no_workers(self):
registry = self.parse(
_document(providers=[_provider("grok", display_name="Grok")], workers=[])
)
self.assertEqual(len(registry.providers), 1)
self.assertEqual(registry.workers, ())
self.assertEqual(workers_for_provider(registry, "grok"), ())
def test_many_workers_may_share_one_provider(self):
registry = self.parse(
_document(
workers=[
_worker("claude-author"),
_worker(
"claude-reviewer",
role="reviewer",
namespace="gitea-reviewer",
profile="prgs-reviewer",
scheduler={"kind": "launchd", "label": "cc.prgs.claude.reviewer"},
),
]
)
)
self.assertEqual(len(workers_for_provider(registry, "claude")), 2)
self.assertEqual(len(registry.providers), 1)
def test_worker_referencing_unknown_provider_is_refused(self):
with self.assertRaises(RegistryValidationError) as ctx:
self.parse(_document(workers=[_worker(provider="mystery")]))
self.assertIn("unknown provider", str(ctx.exception))
def test_lookup_helpers(self):
registry = self.parse(_document())
self.assertIsNotNone(find_worker(registry, "claude-author"))
self.assertIsNone(find_worker(registry, "absent"))
self.assertIsNotNone(find_provider(registry, "claude"))
self.assertIsNone(find_provider(registry, "absent"))
class TestRecordedFields(_TempRegistryCase):
"""AC: records provider, model, project, role, namespace/profile, workflow,
schedule, timeout, enabled state, and scheduler metadata."""
def test_every_required_field_is_recorded(self):
registry = self.parse(_document())
worker = registry.workers[0]
self.assertEqual(worker.provider, "claude")
self.assertEqual(worker.model, "claude-opus-4-8")
self.assertEqual(worker.project, "gitea-tools")
self.assertEqual(worker.role, "author")
self.assertEqual(worker.namespace, "gitea-author")
self.assertEqual(worker.profile, "prgs-author")
self.assertEqual(worker.workflow, "skills/llm-project-workflow/workflows/work-issue.md")
self.assertEqual(worker.schedule.kind, "cron")
self.assertEqual(worker.schedule.expression, "0 * * * *")
self.assertEqual(worker.timeout_seconds, 3600)
self.assertTrue(worker.enabled)
self.assertEqual(worker.scheduler.kind, "launchd")
self.assertEqual(worker.scheduler.label, "cc.prgs.claude.author")
def test_each_required_field_is_individually_required(self):
for field in (
"provider", "model", "project", "role", "namespace",
"profile", "workflow", "schedule", "timeout_seconds",
"enabled", "scheduler", "id", "display_name",
):
with self.subTest(field=field):
worker = _worker()
worker.pop(field)
with self.assertRaises(RegistryValidationError):
self.parse(_document(workers=[worker]))
def test_all_sanctioned_roles_are_accepted(self):
for role in ALLOWED_ROLES:
with self.subTest(role=role):
registry = self.parse(_document(workers=[_worker(role=role)]))
self.assertEqual(registry.workers[0].role, role)
def test_unsanctioned_role_is_refused(self):
with self.assertRaises(RegistryValidationError) as ctx:
self.parse(_document(workers=[_worker(role="admin")]))
self.assertIn("role must be one of", str(ctx.exception))
def test_worker_dict_round_trips_every_field(self):
registry = self.parse(_document())
encoded = worker_to_dict(registry.workers[0])
self.assertEqual(encoded, _worker())
json.dumps(encoded) # must stay JSON-safe for the #799 API
class TestSchemaValidation(_TempRegistryCase):
"""AC: supports schema validation — and fails closed."""
def test_unsupported_version_is_refused(self):
with self.assertRaises(RegistryValidationError):
self.parse(_document(version=2))
def test_root_must_be_an_object(self):
with self.assertRaises(RegistryValidationError):
validate_payload([], source_path=self.path)
def test_providers_must_be_non_empty(self):
with self.assertRaises(RegistryValidationError):
self.parse(_document(providers=[]))
def test_unknown_top_level_field_is_refused(self):
with self.assertRaises(RegistryValidationError) as ctx:
self.parse(_document(fleet=[]))
self.assertIn("unknown fields", str(ctx.exception))
def test_unknown_worker_field_is_refused_not_ignored(self):
# A typo'd field must not be silently dropped: "timeout_second" would
# otherwise read as "no timeout declared".
worker = _worker()
worker["timeout_second"] = 30
with self.assertRaises(RegistryValidationError) as ctx:
self.parse(_document(workers=[worker]))
self.assertIn("timeout_second", str(ctx.exception))
def test_credentials_are_refused_anywhere_in_the_document(self):
for label, mutate in (
("provider.api_token", lambda doc: doc["providers"][0].__setitem__("api_token", "x")),
("worker.password", lambda doc: doc["workers"][0].__setitem__("password", "x")),
("root.secret", lambda doc: doc.__setitem__("secret", "x")),
):
with self.subTest(field=label):
document = _document()
mutate(document)
with self.assertRaises(ValueError) as ctx:
self.parse(document)
self.assertIn("credential", str(ctx.exception).lower())
def test_duplicate_worker_id_is_refused(self):
workers = [_worker("dup"), _worker("dup", scheduler={"kind": "manual"})]
with self.assertRaises(RegistryValidationError) as ctx:
self.parse(_document(workers=workers))
self.assertIn("duplicate worker id", str(ctx.exception))
def test_duplicate_provider_id_is_refused(self):
with self.assertRaises(RegistryValidationError) as ctx:
self.parse(_document(providers=[_provider("claude"), _provider("claude")], workers=[]))
self.assertIn("duplicate provider id", str(ctx.exception))
def test_duplicate_launchagent_label_is_refused(self):
# Two workers sharing a label would silently overwrite each other's agent.
workers = [
_worker("a", scheduler={"kind": "launchd", "label": "cc.prgs.same"}),
_worker("b", scheduler={"kind": "launchd", "label": "cc.prgs.same"}),
]
with self.assertRaises(RegistryValidationError) as ctx:
self.parse(_document(workers=workers))
self.assertIn("duplicate scheduler label", str(ctx.exception))
def test_manual_scheduler_needs_no_label_and_many_may_coexist(self):
workers = [
_worker("a", scheduler={"kind": "manual"}),
_worker("b", scheduler={"kind": "manual"}),
]
registry = self.parse(_document(workers=workers))
self.assertEqual([w.scheduler.label for w in registry.workers], [None, None])
def test_launchd_scheduler_requires_a_label(self):
with self.assertRaises(RegistryValidationError) as ctx:
self.parse(_document(workers=[_worker(scheduler={"kind": "launchd"})]))
self.assertIn("label is required", str(ctx.exception))
def test_unknown_scheduler_kind_is_refused(self):
with self.assertRaises(RegistryValidationError):
self.parse(_document(workers=[_worker(scheduler={"kind": "systemd", "label": "x"})]))
def test_timeout_must_be_a_positive_bounded_integer(self):
for bad in (0, -1, "3600", 1.5, True, 86_401):
with self.subTest(timeout=bad):
with self.assertRaises(RegistryValidationError):
self.parse(_document(workers=[_worker(timeout_seconds=bad)]))
def test_enabled_must_be_a_real_boolean(self):
for bad in ("true", 1, None):
with self.subTest(enabled=bad):
with self.assertRaises(RegistryValidationError):
self.parse(_document(workers=[_worker(enabled=bad)]))
def test_identifier_shape_is_enforced(self):
for bad in ("Claude Author", "-leading", "UPPER", ""):
with self.subTest(worker_id=bad):
with self.assertRaises(RegistryValidationError):
self.parse(_document(workers=[_worker(bad)]))
class TestScheduleValidation(_TempRegistryCase):
"""Schedules are declarations; next-run computation belongs to #803."""
def test_interval_schedule_requires_positive_seconds(self):
registry = self.parse(
_document(workers=[_worker(schedule={"kind": "interval", "seconds": 900})])
)
self.assertEqual(registry.workers[0].schedule.seconds, 900)
with self.assertRaises(RegistryValidationError):
self.parse(_document(workers=[_worker(schedule={"kind": "interval"})]))
with self.assertRaises(RegistryValidationError):
self.parse(_document(workers=[_worker(schedule={"kind": "interval", "seconds": 0})]))
def test_cron_schedule_requires_five_fields(self):
with self.assertRaises(RegistryValidationError) as ctx:
self.parse(_document(workers=[_worker(schedule={"kind": "cron", "expression": "0 *"})]))
self.assertIn("five crontab fields", str(ctx.exception))
def test_manual_schedule_needs_no_timing(self):
registry = self.parse(_document(workers=[_worker(schedule={"kind": "manual"})]))
schedule = registry.workers[0].schedule
self.assertEqual(schedule.kind, "manual")
self.assertIsNone(schedule.seconds)
self.assertIsNone(schedule.expression)
def test_fields_from_the_wrong_kind_are_refused(self):
with self.assertRaises(RegistryValidationError) as ctx:
self.parse(_document(workers=[_worker(schedule={"kind": "manual", "seconds": 60})]))
self.assertIn("not valid for kind", str(ctx.exception))
def test_unknown_schedule_kind_is_refused(self):
with self.assertRaises(RegistryValidationError):
self.parse(_document(workers=[_worker(schedule={"kind": "hourly"})]))
class TestAtomicPersistence(_TempRegistryCase):
"""AC: atomic persistence."""
def test_save_then_load_round_trips(self):
registry = self.parse(_document())
save_registry(registry, self.path)
reloaded = load_registry(self.path)
self.assertEqual(
[worker_to_dict(w) for w in reloaded.workers],
[worker_to_dict(w) for w in registry.workers],
)
def test_save_leaves_no_temp_files_behind(self):
registry = self.parse(_document())
save_registry(registry, self.path)
save_registry(registry, self.path)
leftovers = [p.name for p in self.path.parent.iterdir() if p.name.startswith(".")]
self.assertEqual(leftovers, [])
def test_save_refuses_to_persist_an_invalid_document(self):
registry = self.parse(_document())
broken = WorkerRegistry(
version=registry.version,
revision=registry.revision,
updated_at=registry.updated_at,
providers=registry.providers,
# A worker whose provider is not declared in the registry.
workers=tuple(
type(worker)(**{**worker.__dict__, "provider": "vanished"})
for worker in registry.workers
),
source_path=self.path,
)
with self.assertRaises(RegistryValidationError):
save_registry(broken, self.path)
self.assertFalse(self.path.exists(), "invalid save must not create the file")
def test_document_shape_excludes_local_paths_but_api_shape_includes_it(self):
registry = self.parse(_document())
self.assertNotIn("source_path", registry_to_document(registry))
self.assertEqual(registry_to_dict(registry)["source_path"], str(self.path))
class TestVersioningAndRollback(_TempRegistryCase):
"""AC: versioning and rollback."""
def _seed(self) -> WorkerRegistry:
self.write(_document())
return load_registry(self.path)
def test_revision_increments_on_each_save(self):
registry = self._seed()
self.assertEqual(registry.revision, 1)
second = save_registry(registry, self.path)
self.assertEqual(second.revision, 2)
third = save_registry(second, self.path)
self.assertEqual(third.revision, 3)
def test_updated_at_is_refreshed_and_utc(self):
registry = self._seed()
saved = save_registry(registry, self.path)
self.assertRegex(saved.updated_at, r"^\d{4}-\d{2}-\d{2}T\d{2}:\d{2}:\d{2}Z$")
def test_superseded_revisions_are_retained(self):
registry = self._seed()
second = save_registry(registry, self.path)
save_registry(second, self.path)
self.assertEqual(list_revisions(self.path), (1, 2))
self.assertTrue(history_dir(self.path).is_dir())
def test_rollback_restores_prior_content_as_a_new_revision(self):
self.write(_document(workers=[_worker("original")]))
registry = load_registry(self.path)
changed = WorkerRegistry(
version=registry.version,
revision=registry.revision,
updated_at=registry.updated_at,
providers=registry.providers,
workers=(), # operator deletes every worker
source_path=self.path,
)
save_registry(changed, self.path)
self.assertEqual(load_registry(self.path).workers, ())
restored = rollback_to_revision(1, self.path)
self.assertEqual([w.id for w in restored.workers], ["original"])
# Append-only: the rollback publishes a new head rather than rewinding.
self.assertGreater(restored.revision, 2)
self.assertEqual([w.id for w in load_registry(self.path).workers], ["original"])
def test_rollback_to_unknown_revision_fails_closed(self):
self._seed()
with self.assertRaises(RegistryValidationError) as ctx:
rollback_to_revision(99, self.path)
self.assertIn("not retained", str(ctx.exception))
def test_revision_must_be_a_positive_integer(self):
for bad in (0, -1, "1", None):
with self.subTest(revision=bad):
with self.assertRaises(RegistryValidationError):
self.parse(_document(revision=bad))
def test_history_is_empty_before_any_save(self):
self.write(_document())
self.assertEqual(list_revisions(self.path), ())
if __name__ == "__main__":
unittest.main()
+57
View File
@@ -0,0 +1,57 @@
{
"version": 1,
"revision": 1,
"updated_at": "2026-07-22T00:00:00Z",
"providers": [
{
"id": "claude",
"display_name": "Claude",
"vendor": "Anthropic",
"executable": "claude",
"available": true,
"models": [
"claude-opus-4-8",
"claude-sonnet-5",
"claude-haiku-4-5-20251001"
],
"notes": "Model list is a declaration. Live enumeration and version inspection belong to the provider adapter framework (#800)."
},
{
"id": "grok",
"display_name": "Grok",
"vendor": "xAI",
"executable": "grok",
"available": true,
"models": [],
"notes": "Models enumerated by the provider adapter (#800); not declared here."
},
{
"id": "codex",
"display_name": "Codex",
"vendor": "OpenAI",
"executable": "codex",
"available": true,
"models": [],
"notes": "Models enumerated by the provider adapter (#800); not declared here."
},
{
"id": "agy",
"display_name": "AGY",
"vendor": "Antigravity",
"executable": "agy",
"available": true,
"models": [],
"notes": "MCP allowlist gating applies to this provider; confirm server-side allowlist before configuring a worker."
},
{
"id": "kimi-k",
"display_name": "Kimi K",
"vendor": "Moonshot AI",
"executable": "kimi",
"available": true,
"models": [],
"notes": "Provider id is kimi-k; the executable on PATH is kimi. Models enumerated by the provider adapter (#800)."
}
],
"workers": []
}
+1 -33
View File
@@ -8,27 +8,7 @@ from dataclasses import dataclass
from pathlib import Path from pathlib import Path
from typing import Any from typing import Any
_FORBIDDEN_EXACT_KEYS = frozenset({ from webui.registry_safety import reject_credential_keys as _reject_credential_keys
"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 = ( _REQUIRED_PROJECT_FIELDS = (
"id", "id",
@@ -79,18 +59,6 @@ def default_registry_path() -> Path:
return (Path(__file__).resolve().parent / "data" / "projects.registry.json").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, ...]: def _parse_onboarding(raw: list[dict[str, Any]] | None) -> tuple[OnboardingStep, ...]:
if not raw: if not raw:
return () return ()
+52
View File
@@ -0,0 +1,52 @@
"""Shared credential-rejection guard for web UI registries (#427, #798).
Registries are operator-editable declarative files that the web UI loads and,
for the worker registry, writes back. None of them may ever carry a secret:
credentials belong in the keychain and reach worker processes through
environment injection, never through a file the browser layer can read.
The check is structural rather than value-based on purpose. A value scanner has
to guess what a secret looks like; a key scanner refuses the *shape* of a
credential field, so an operator cannot introduce one by accident and a later
loader cannot silently pass one through.
"""
from __future__ import annotations
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:
"""Return True when *key* names a credential field."""
lowered = key.lower()
if lowered in _FORBIDDEN_EXACT_KEYS:
return True
return (
lowered.startswith(_FORBIDDEN_KEY_PREFIXES)
or lowered.endswith(_FORBIDDEN_KEY_SUFFIXES)
)
def reject_credential_keys(obj: Any, *, path: str = "", subject: str = "registry") -> None:
"""Raise ValueError when *obj* carries a credential-shaped key at any depth."""
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"{subject} must not store credentials ({key_path})")
reject_credential_keys(value, path=key_path, subject=subject)
elif isinstance(obj, list):
for index, item in enumerate(obj):
reject_credential_keys(item, path=f"{path}[{index}]", subject=subject)
+647
View File
@@ -0,0 +1,647 @@
"""Declarative worker registry and configuration schema (#798, epic #797).
The registry is the single source of truth for the scheduled multi-LLM worker
fleet. It is a versioned JSON document holding two *separate* entity kinds:
* **Providers** — the LLM runtimes a worker can be built on (Claude, Grok,
Codex, AGY, Kimi K). A provider describes the runtime itself: vendor,
executable name, models it can serve, and whether it is available on this
machine. Providers exist whether or not any worker uses them.
* **Workers** — a configured *instance*: one provider, one model, one project,
one role, one MCP namespace/profile, one workflow, one schedule. Several
workers may share a provider; a worker naming an undeclared provider is
refused.
Keeping them separate is what lets #799 list all five providers even when a
provider currently has no configured worker, and it stops provider facts from
being copied into (and drifting across) every worker record.
Scope boundary. This module owns the data model, its validation, and its
persistence. It does **not** schedule anything, launch anything, probe provider
executables, or serve HTTP. Loading a registry never touches a process; the
live fields a dashboard wants (PID, elapsed time, next run) are derived
elsewhere (#799, #801, #803, #804) from these declarations.
Safety invariants:
* No credential may be stored (:mod:`webui.registry_safety`), so the registry
stays safe to render and to hand to a browser layer.
* Validation fails closed. Unknown fields are refused rather than ignored, so a
typo cannot silently disable a timeout or a role binding.
* Writes are atomic and every superseded document is retained as a numbered
revision, so a bad edit is recoverable by rollback rather than hand-repair.
"""
from __future__ import annotations
import json
import os
import re
import tempfile
from dataclasses import dataclass
from datetime import datetime, timezone
from pathlib import Path
from typing import Any
from webui.registry_safety import reject_credential_keys
SCHEMA_VERSION = 1
#: Roles a worker may hold. These mirror the sanctioned MCP role kinds; a
#: worker may not invent one, because the role selects the namespace/profile
#: whose capability gates constrain it.
ALLOWED_ROLES = ("author", "reviewer", "merger", "reconciler", "cleanup")
#: Scheduler backends the registry can describe. ``manual`` means the worker is
#: only ever started on request and has no recurring trigger.
ALLOWED_SCHEDULER_KINDS = ("launchd", "manual")
#: Schedule kinds. Next-run computation belongs to #803; this module only
#: guarantees the declaration is well formed.
ALLOWED_SCHEDULE_KINDS = ("interval", "cron", "manual")
_REQUIRED_PROVIDER_FIELDS = ("id", "display_name", "vendor", "executable", "available")
_OPTIONAL_PROVIDER_FIELDS = ("models", "notes")
_REQUIRED_WORKER_FIELDS = (
"id",
"display_name",
"provider",
"model",
"project",
"role",
"namespace",
"profile",
"workflow",
"schedule",
"timeout_seconds",
"enabled",
"scheduler",
)
_OPTIONAL_WORKER_FIELDS = ("notes",)
_ID_RE = re.compile(r"^[a-z0-9][a-z0-9._-]*$")
#: Guards against an operator writing a timeout that would let a worker hold a
#: lease effectively forever. 24h is far above any sanctioned cycle.
_MAX_TIMEOUT_SECONDS = 86_400
#: How many superseded revisions to retain beside the live file.
_HISTORY_LIMIT = 20
_TOP_LEVEL_FIELDS = frozenset({"version", "revision", "updated_at", "providers", "workers"})
class RegistryValidationError(ValueError):
"""Raised when a registry document violates the schema."""
@dataclass(frozen=True)
class ProviderRecord:
id: str
display_name: str
vendor: str
executable: str
available: bool
models: tuple[str, ...]
notes: str
@dataclass(frozen=True)
class ScheduleSpec:
kind: str
#: Set for ``interval`` schedules.
seconds: int | None
#: Set for ``cron`` schedules — a five-field crontab expression.
expression: str | None
@dataclass(frozen=True)
class SchedulerSpec:
kind: str
#: LaunchAgent label; required for ``launchd``, absent for ``manual``.
label: str | None
@dataclass(frozen=True)
class WorkerRecord:
id: str
display_name: str
provider: str
model: str
project: str
role: str
namespace: str
profile: str
workflow: str
schedule: ScheduleSpec
timeout_seconds: int
enabled: bool
scheduler: SchedulerSpec
notes: str
@dataclass(frozen=True)
class WorkerRegistry:
version: int
revision: int
updated_at: str
providers: tuple[ProviderRecord, ...]
workers: tuple[WorkerRecord, ...]
source_path: Path
# ── paths ────────────────────────────────────────────────────────────────────
def default_registry_path() -> Path:
"""Location of the packaged worker registry, overridable for tests/deploys."""
override = os.environ.get("WEBUI_WORKER_REGISTRY", "").strip()
if override:
return Path(override).expanduser().resolve()
return (Path(__file__).resolve().parent / "data" / "workers.registry.json").resolve()
def history_dir(path: Path | None = None) -> Path:
"""Directory holding superseded revisions of *path*."""
source = (path or default_registry_path()).resolve()
return source.parent / f"{source.name}.history"
# ── field helpers ────────────────────────────────────────────────────────────
def _require_exact_fields(
raw: Any,
*,
required: tuple[str, ...],
optional: tuple[str, ...],
subject: str,
) -> dict[str, Any]:
if not isinstance(raw, dict):
raise RegistryValidationError(f"{subject} must be an object")
missing = [field for field in required if field not in raw]
if missing:
raise RegistryValidationError(
f"{subject} missing required fields: {', '.join(sorted(missing))}"
)
unknown = sorted(set(raw) - set(required) - set(optional))
if unknown:
# Fail closed: silently dropping an unrecognized key is how a typo'd
# "timeout_second" ends up meaning "no timeout".
raise RegistryValidationError(f"{subject} has unknown fields: {', '.join(unknown)}")
return raw
def _require_identifier(value: Any, *, subject: str) -> str:
text = str(value).strip()
if not _ID_RE.match(text):
raise RegistryValidationError(
f"{subject} must be lowercase alphanumeric with '.', '_', or '-' (got {value!r})"
)
return text
def _require_text(value: Any, *, subject: str) -> str:
if not isinstance(value, str):
raise RegistryValidationError(f"{subject} must be a string (got {value!r})")
text = value.strip()
if not text:
raise RegistryValidationError(f"{subject} must be a non-empty string")
return text
def _require_bool(value: Any, *, subject: str) -> bool:
if not isinstance(value, bool):
raise RegistryValidationError(f"{subject} must be a boolean (got {value!r})")
return value
def _require_positive_int(value: Any, *, subject: str, maximum: int | None = None) -> int:
if isinstance(value, bool) or not isinstance(value, int):
raise RegistryValidationError(f"{subject} must be an integer (got {value!r})")
if value <= 0:
raise RegistryValidationError(f"{subject} must be greater than zero (got {value})")
if maximum is not None and value > maximum:
raise RegistryValidationError(f"{subject} must not exceed {maximum} (got {value})")
return value
# ── parsing ──────────────────────────────────────────────────────────────────
def _parse_provider(raw: Any) -> ProviderRecord:
data = _require_exact_fields(
raw,
required=_REQUIRED_PROVIDER_FIELDS,
optional=_OPTIONAL_PROVIDER_FIELDS,
subject="provider",
)
provider_id = _require_identifier(data["id"], subject="provider.id")
models_raw = data.get("models") or []
if not isinstance(models_raw, list):
raise RegistryValidationError(f"provider[{provider_id}].models must be an array")
models = tuple(
_require_text(item, subject=f"provider[{provider_id}].models[]") for item in models_raw
)
return ProviderRecord(
id=provider_id,
display_name=_require_text(
data["display_name"], subject=f"provider[{provider_id}].display_name"
),
vendor=_require_text(data["vendor"], subject=f"provider[{provider_id}].vendor"),
executable=_require_text(data["executable"], subject=f"provider[{provider_id}].executable"),
available=_require_bool(data["available"], subject=f"provider[{provider_id}].available"),
models=models,
notes=str(data.get("notes") or "").strip(),
)
def _parse_schedule(raw: Any, *, subject: str) -> ScheduleSpec:
if not isinstance(raw, dict):
raise RegistryValidationError(f"{subject} must be an object")
kind = _require_text(raw.get("kind"), subject=f"{subject}.kind")
if kind not in ALLOWED_SCHEDULE_KINDS:
raise RegistryValidationError(
f"{subject}.kind must be one of {', '.join(ALLOWED_SCHEDULE_KINDS)} (got {kind!r})"
)
seconds: int | None = None
expression: str | None = None
if kind == "interval":
if "seconds" not in raw:
raise RegistryValidationError(f"{subject}.seconds is required for interval schedules")
seconds = _require_positive_int(raw["seconds"], subject=f"{subject}.seconds")
elif kind == "cron":
if "expression" not in raw:
raise RegistryValidationError(f"{subject}.expression is required for cron schedules")
expression = _require_text(raw["expression"], subject=f"{subject}.expression")
if len(expression.split()) != 5:
raise RegistryValidationError(
f"{subject}.expression must have five crontab fields (got {expression!r})"
)
allowed = {"kind"}
if kind == "interval":
allowed.add("seconds")
elif kind == "cron":
allowed.add("expression")
unknown = sorted(set(raw) - allowed)
if unknown:
raise RegistryValidationError(
f"{subject} has fields not valid for kind {kind!r}: {', '.join(unknown)}"
)
return ScheduleSpec(kind=kind, seconds=seconds, expression=expression)
def _parse_scheduler(raw: Any, *, subject: str) -> SchedulerSpec:
if not isinstance(raw, dict):
raise RegistryValidationError(f"{subject} must be an object")
kind = _require_text(raw.get("kind"), subject=f"{subject}.kind")
if kind not in ALLOWED_SCHEDULER_KINDS:
raise RegistryValidationError(
f"{subject}.kind must be one of {', '.join(ALLOWED_SCHEDULER_KINDS)} (got {kind!r})"
)
label: str | None = None
if kind == "launchd":
if "label" not in raw:
raise RegistryValidationError(f"{subject}.label is required for launchd schedulers")
label = _require_text(raw["label"], subject=f"{subject}.label")
allowed = {"kind"}
if kind == "launchd":
allowed.add("label")
unknown = sorted(set(raw) - allowed)
if unknown:
raise RegistryValidationError(
f"{subject} has fields not valid for kind {kind!r}: {', '.join(unknown)}"
)
return SchedulerSpec(kind=kind, label=label)
def _parse_worker(raw: Any) -> WorkerRecord:
data = _require_exact_fields(
raw,
required=_REQUIRED_WORKER_FIELDS,
optional=_OPTIONAL_WORKER_FIELDS,
subject="worker",
)
worker_id = _require_identifier(data["id"], subject="worker.id")
role = _require_text(data["role"], subject=f"worker[{worker_id}].role")
if role not in ALLOWED_ROLES:
raise RegistryValidationError(
f"worker[{worker_id}].role must be one of {', '.join(ALLOWED_ROLES)} (got {role!r})"
)
return WorkerRecord(
id=worker_id,
display_name=_require_text(
data["display_name"], subject=f"worker[{worker_id}].display_name"
),
provider=_require_identifier(data["provider"], subject=f"worker[{worker_id}].provider"),
model=_require_text(data["model"], subject=f"worker[{worker_id}].model"),
project=_require_text(data["project"], subject=f"worker[{worker_id}].project"),
role=role,
namespace=_require_text(data["namespace"], subject=f"worker[{worker_id}].namespace"),
profile=_require_text(data["profile"], subject=f"worker[{worker_id}].profile"),
workflow=_require_text(data["workflow"], subject=f"worker[{worker_id}].workflow"),
schedule=_parse_schedule(data["schedule"], subject=f"worker[{worker_id}].schedule"),
timeout_seconds=_require_positive_int(
data["timeout_seconds"],
subject=f"worker[{worker_id}].timeout_seconds",
maximum=_MAX_TIMEOUT_SECONDS,
),
enabled=_require_bool(data["enabled"], subject=f"worker[{worker_id}].enabled"),
scheduler=_parse_scheduler(data["scheduler"], subject=f"worker[{worker_id}].scheduler"),
notes=str(data.get("notes") or "").strip(),
)
def _require_unique(values: list[str], *, subject: str) -> None:
seen: set[str] = set()
for value in values:
if value in seen:
raise RegistryValidationError(f"duplicate {subject}: {value}")
seen.add(value)
def validate_payload(payload: Any, *, source_path: Path) -> WorkerRegistry:
"""Validate a decoded registry document and return the typed registry.
Raises :class:`RegistryValidationError` on any violation; never partially
accepts a document.
"""
if not isinstance(payload, dict):
raise RegistryValidationError("registry root must be an object")
version = payload.get("version")
if version != SCHEMA_VERSION:
raise RegistryValidationError(f"unsupported registry version: {version!r}")
reject_credential_keys(payload, subject="worker registry")
unknown = sorted(set(payload) - _TOP_LEVEL_FIELDS)
if unknown:
raise RegistryValidationError(f"registry has unknown fields: {', '.join(unknown)}")
revision = _require_positive_int(payload.get("revision"), subject="revision")
updated_at = _require_text(payload.get("updated_at"), subject="updated_at")
providers_raw = payload.get("providers")
if not isinstance(providers_raw, list) or not providers_raw:
raise RegistryValidationError("providers must be a non-empty array")
providers = tuple(_parse_provider(item) for item in providers_raw)
_require_unique([provider.id for provider in providers], subject="provider id")
workers_raw = payload.get("workers")
if not isinstance(workers_raw, list):
raise RegistryValidationError("workers must be an array")
workers = tuple(_parse_worker(item) for item in workers_raw)
_require_unique([worker.id for worker in workers], subject="worker id")
# Referential integrity: a worker naming an undeclared provider would look
# configured while being unrunnable, which is exactly the ambiguous
# ownership the epic requires to fail closed.
known_providers = {provider.id for provider in providers}
for worker in workers:
if worker.provider not in known_providers:
raise RegistryValidationError(
f"worker[{worker.id}].provider references unknown provider {worker.provider!r}"
)
# A LaunchAgent label identifies a job to launchd; two workers sharing one
# would silently overwrite each other's agent.
_require_unique(
[worker.scheduler.label for worker in workers if worker.scheduler.label],
subject="scheduler label",
)
return WorkerRegistry(
version=version,
revision=revision,
updated_at=updated_at,
providers=providers,
workers=workers,
source_path=source_path,
)
def load_registry(path: Path | None = None) -> WorkerRegistry:
"""Load and validate the worker registry from disk."""
source = (path or default_registry_path()).resolve()
payload = json.loads(source.read_text(encoding="utf-8"))
return validate_payload(payload, source_path=source)
# ── serialization ────────────────────────────────────────────────────────────
def provider_to_dict(provider: ProviderRecord) -> dict[str, Any]:
return {
"id": provider.id,
"display_name": provider.display_name,
"vendor": provider.vendor,
"executable": provider.executable,
"available": provider.available,
"models": list(provider.models),
"notes": provider.notes,
}
def _schedule_to_dict(schedule: ScheduleSpec) -> dict[str, Any]:
payload: dict[str, Any] = {"kind": schedule.kind}
if schedule.kind == "interval":
payload["seconds"] = schedule.seconds
elif schedule.kind == "cron":
payload["expression"] = schedule.expression
return payload
def _scheduler_to_dict(scheduler: SchedulerSpec) -> dict[str, Any]:
payload: dict[str, Any] = {"kind": scheduler.kind}
if scheduler.kind == "launchd":
payload["label"] = scheduler.label
return payload
def worker_to_dict(worker: WorkerRecord) -> dict[str, Any]:
return {
"id": worker.id,
"display_name": worker.display_name,
"provider": worker.provider,
"model": worker.model,
"project": worker.project,
"role": worker.role,
"namespace": worker.namespace,
"profile": worker.profile,
"workflow": worker.workflow,
"schedule": _schedule_to_dict(worker.schedule),
"timeout_seconds": worker.timeout_seconds,
"enabled": worker.enabled,
"scheduler": _scheduler_to_dict(worker.scheduler),
"notes": worker.notes,
}
def registry_to_document(registry: WorkerRegistry) -> dict[str, Any]:
"""Serialize to the on-disk document shape (no local paths embedded)."""
return {
"version": registry.version,
"revision": registry.revision,
"updated_at": registry.updated_at,
"providers": [provider_to_dict(provider) for provider in registry.providers],
"workers": [worker_to_dict(worker) for worker in registry.workers],
}
def registry_to_dict(registry: WorkerRegistry) -> dict[str, Any]:
"""Serialize for JSON API responses (adds the resolved source path)."""
document = registry_to_document(registry)
document["source_path"] = str(registry.source_path)
return document
def find_worker(registry: WorkerRegistry, worker_id: str) -> WorkerRecord | None:
for worker in registry.workers:
if worker.id == worker_id:
return worker
return None
def find_provider(registry: WorkerRegistry, provider_id: str) -> ProviderRecord | None:
for provider in registry.providers:
if provider.id == provider_id:
return provider
return None
def workers_for_provider(registry: WorkerRegistry, provider_id: str) -> tuple[WorkerRecord, ...]:
return tuple(worker for worker in registry.workers if worker.provider == provider_id)
# ── persistence ──────────────────────────────────────────────────────────────
def _utc_now() -> str:
return datetime.now(timezone.utc).strftime("%Y-%m-%dT%H:%M:%SZ")
def _atomic_write(path: Path, payload: str) -> None:
"""Write *payload* to *path* atomically: temp file in the same dir, fsync, replace."""
parent = path.parent
parent.mkdir(parents=True, exist_ok=True)
fd, temp_path = tempfile.mkstemp(prefix=f".{path.name}-", suffix=".tmp", dir=parent)
try:
with os.fdopen(fd, "w", encoding="utf-8") as handle:
handle.write(payload)
handle.flush()
os.fsync(handle.fileno())
os.replace(temp_path, path)
finally:
if os.path.exists(temp_path):
try:
os.remove(temp_path)
except OSError:
pass
def _revision_path(directory: Path, revision: int) -> Path:
return directory / f"rev-{revision:06d}.json"
def _prune_history(path: Path) -> None:
directory = history_dir(path)
revisions = list_revisions(path)
excess = len(revisions) - _HISTORY_LIMIT
for revision in revisions[: max(0, excess)]:
_revision_path(directory, revision).unlink(missing_ok=True)
def _archive_current(path: Path) -> int | None:
"""Copy the live document into the history dir under its own revision number."""
if not path.exists():
return None
try:
existing = json.loads(path.read_text(encoding="utf-8"))
revision = int(existing.get("revision", 0))
except (json.JSONDecodeError, TypeError, ValueError, AttributeError):
# An unreadable live file has no trustworthy revision number to file it
# under, so it cannot join the history chain.
return None
if revision <= 0:
return None
_atomic_write(
_revision_path(history_dir(path), revision),
json.dumps(existing, indent=2, sort_keys=True) + "\n",
)
_prune_history(path)
return revision
def list_revisions(path: Path | None = None) -> tuple[int, ...]:
"""Revision numbers retained in history for *path*, oldest first."""
directory = history_dir(path)
if not directory.is_dir():
return ()
revisions: list[int] = []
for entry in directory.glob("rev-*.json"):
try:
revisions.append(int(entry.stem.split("-", 1)[1]))
except (IndexError, ValueError):
continue
return tuple(sorted(revisions))
def save_registry(
registry: WorkerRegistry,
path: Path | None = None,
*,
updated_at: str | None = None,
) -> WorkerRegistry:
"""Validate, archive the superseded revision, then atomically persist a new one.
The stored revision is always the previous revision plus one, so a reader
can tell two documents apart even when their content is otherwise equal.
Returns the registry exactly as persisted.
"""
target = (path or registry.source_path or default_registry_path()).resolve()
document = registry_to_document(registry)
# Re-validate before writing: a registry assembled in memory has not
# necessarily been through the loader.
validate_payload(document, source_path=target)
archived = _archive_current(target)
document["revision"] = (archived + 1) if archived is not None else registry.revision
document["updated_at"] = updated_at or _utc_now()
persisted = validate_payload(document, source_path=target)
_atomic_write(target, json.dumps(document, indent=2, sort_keys=True) + "\n")
return persisted
def rollback_to_revision(revision: int, path: Path | None = None) -> WorkerRegistry:
"""Restore a retained *revision* as a new head revision.
History is append-only: rolling back does not delete the revisions in
between, it republishes the chosen one under the next revision number, so a
rollback is itself reversible.
"""
target = (path or default_registry_path()).resolve()
snapshot_path = _revision_path(history_dir(target), revision)
if not snapshot_path.exists():
available = ", ".join(str(item) for item in list_revisions(target)) or "(none)"
raise RegistryValidationError(
f"revision {revision} is not retained for {target.name}; available: {available}"
)
payload = json.loads(snapshot_path.read_text(encoding="utf-8"))
restored = validate_payload(payload, source_path=target)
return save_registry(restored, target)