Some checks failed
Deploy Development / deploy (push) Successful in 49s
Test Suite / pytest-backend (push) Failing after 3m12s
Test Suite / k6 /api/health Baseline (push) Has been skipped
Test Suite / playwright-smoke (push) Has been skipped
Test Suite / lint-backend (push) Successful in 2s
Test Suite / compose-smoke (push) Has been skipped
Co-authored-by: Cursor <cursoragent@cursor.com>
48 lines
1.1 KiB
Python
48 lines
1.1 KiB
Python
"""Method Registry — built-in steering methods."""
|
|
|
|
from __future__ import annotations
|
|
|
|
from dataclasses import dataclass
|
|
from typing import Literal
|
|
|
|
_METHODS: dict[str, "MethodDefinition"] = {}
|
|
|
|
|
|
@dataclass(frozen=True)
|
|
class MethodDefinition:
|
|
key: str
|
|
version: str
|
|
label: str
|
|
description: str
|
|
default_lifecycle_steps: tuple[str, ...]
|
|
next_action_strategy_key: str = "default"
|
|
data_slices: frozenset[str] = frozenset()
|
|
compatible_archetype_keys: frozenset[str] | Literal["*"] = "*"
|
|
|
|
|
|
def method_compatible_with_archetype(
|
|
method: MethodDefinition, archetype_key: str
|
|
) -> bool:
|
|
compat = method.compatible_archetype_keys
|
|
if compat == "*":
|
|
return True
|
|
return archetype_key in compat
|
|
|
|
|
|
def register_method(defn: MethodDefinition) -> None:
|
|
if defn.key in _METHODS:
|
|
raise ValueError(f"Method already registered: {defn.key}")
|
|
_METHODS[defn.key] = defn
|
|
|
|
|
|
def get_method(key: str) -> MethodDefinition | None:
|
|
return _METHODS.get(key)
|
|
|
|
|
|
def list_methods() -> tuple[MethodDefinition, ...]:
|
|
return tuple(_METHODS.values())
|
|
|
|
|
|
def clear_methods_for_tests() -> None:
|
|
_METHODS.clear()
|