From 5463f589334f6a9520e2512a3c33929976d924ac Mon Sep 17 00:00:00 2001 From: jcwalker3 Date: Wed, 22 Jul 2026 06:10:34 -0500 Subject: [PATCH] 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) --- tests/test_webui_worker_registry.py | 458 ++++++++++++++++++++ webui/data/workers.registry.json | 57 +++ webui/project_registry.py | 34 +- webui/registry_safety.py | 52 +++ webui/worker_registry.py | 647 ++++++++++++++++++++++++++++ 5 files changed, 1215 insertions(+), 33 deletions(-) create mode 100644 tests/test_webui_worker_registry.py create mode 100644 webui/data/workers.registry.json create mode 100644 webui/registry_safety.py create mode 100644 webui/worker_registry.py diff --git a/tests/test_webui_worker_registry.py b/tests/test_webui_worker_registry.py new file mode 100644 index 0000000..b579204 --- /dev/null +++ b/tests/test_webui_worker_registry.py @@ -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() diff --git a/webui/data/workers.registry.json b/webui/data/workers.registry.json new file mode 100644 index 0000000..503b6ef --- /dev/null +++ b/webui/data/workers.registry.json @@ -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": [] +} diff --git a/webui/project_registry.py b/webui/project_registry.py index 2c5e889..afc5e15 100644 --- a/webui/project_registry.py +++ b/webui/project_registry.py @@ -8,27 +8,7 @@ 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) - ) +from webui.registry_safety import reject_credential_keys as _reject_credential_keys _REQUIRED_PROJECT_FIELDS = ( "id", @@ -79,18 +59,6 @@ def default_registry_path() -> Path: 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 () diff --git a/webui/registry_safety.py b/webui/registry_safety.py new file mode 100644 index 0000000..9ba741e --- /dev/null +++ b/webui/registry_safety.py @@ -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) diff --git a/webui/worker_registry.py b/webui/worker_registry.py new file mode 100644 index 0000000..ccc923a --- /dev/null +++ b/webui/worker_registry.py @@ -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) -- 2.43.7