jcwalker3andClaude Opus 4.8
5463f58933
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]>
459 lines
18 KiB
Python
459 lines
18 KiB
Python
"""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()
|