Lars
a5154cbd3b
Some checks failed
Deploy Development / deploy (push) Successful in 47s
Test Suite / pytest-backend (push) Failing after 4m15s
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
K-Ext-4/P8 deklarative Agent-Slots; P7 tech_debt Read Model und Vokabular; Review-Attention in den Kernel verlagert. Co-authored-by: Cursor <cursoragent@cursor.com>
108 lines
3.5 KiB
Python
108 lines
3.5 KiB
Python
"""Agent slot provider registry — K-Ext-4."""
|
|
|
|
from __future__ import annotations
|
|
|
|
from dataclasses import dataclass, field
|
|
from typing import Any, Callable
|
|
|
|
from steering.eval_context import SteeringEvalContext
|
|
|
|
AgentSlotResolve = Callable[[SteeringEvalContext, dict[str, Any], dict[str, Any]], list[dict]]
|
|
|
|
|
|
@dataclass(frozen=True)
|
|
class AgentSlotProvider:
|
|
slot_key: str
|
|
requires_elements: frozenset[str]
|
|
requires_data_slices: frozenset[str]
|
|
resolve: AgentSlotResolve
|
|
requires_any_element: frozenset[str] = field(default_factory=frozenset)
|
|
excludes_elements: frozenset[str] = field(default_factory=frozenset)
|
|
|
|
|
|
_PROVIDERS: list[AgentSlotProvider] = []
|
|
|
|
|
|
def register_agent_slot(provider: AgentSlotProvider) -> None:
|
|
_PROVIDERS.append(provider)
|
|
|
|
|
|
def _slot_enabled(ctx: SteeringEvalContext, slot_key: str) -> bool:
|
|
config = ctx.agent_slot_config
|
|
enabled = config.get("enabled_slots")
|
|
if enabled is None:
|
|
return True
|
|
return slot_key in enabled
|
|
|
|
|
|
def compute_agent_slots(
|
|
ctx: SteeringEvalContext,
|
|
*,
|
|
read_models: dict[str, Any],
|
|
proposals: dict[str, Any],
|
|
) -> list[dict[str, Any]]:
|
|
elements = ctx.steering_elements
|
|
slices = frozenset(ctx.data_slices)
|
|
slots: list[dict[str, Any]] = []
|
|
|
|
for provider in _PROVIDERS:
|
|
if not _slot_enabled(ctx, provider.slot_key):
|
|
continue
|
|
if provider.requires_elements and not provider.requires_elements <= elements:
|
|
continue
|
|
if provider.excludes_elements and provider.excludes_elements & elements:
|
|
continue
|
|
if provider.requires_any_element and not (
|
|
provider.requires_any_element & elements
|
|
):
|
|
continue
|
|
if provider.requires_data_slices and not provider.requires_data_slices <= slices:
|
|
continue
|
|
slots.extend(provider.resolve(ctx, read_models, proposals))
|
|
|
|
for slot in slots:
|
|
slot.setdefault("data_source", "steering_kernel")
|
|
slot.setdefault("initiative_id", ctx.initiative_id)
|
|
slot.setdefault("context_refs", ["steering_snapshot", "operating_context"])
|
|
|
|
slots.sort(
|
|
key=lambda s: {"high": 0, "normal": 1, "low": 2}.get(s.get("priority", "normal"), 9)
|
|
)
|
|
return slots
|
|
|
|
|
|
def _register_builtin_agent_slots() -> None:
|
|
from steering.agent_slots.guardrail_review import resolve_guardrail_review_slots
|
|
from steering.agent_slots.recurring_due import resolve_recurring_due_slots
|
|
from steering.agent_slots.review_due import resolve_review_due_slots
|
|
|
|
register_agent_slot(
|
|
AgentSlotProvider(
|
|
slot_key="review.due",
|
|
requires_elements=frozenset(),
|
|
requires_data_slices=frozenset({"reviews"}),
|
|
resolve=lambda ctx, read_models, proposals: resolve_review_due_slots(ctx),
|
|
)
|
|
)
|
|
register_agent_slot(
|
|
AgentSlotProvider(
|
|
slot_key="recurring.due",
|
|
requires_elements=frozenset({"recurring_rhythm"}),
|
|
requires_data_slices=frozenset({"recurring"}),
|
|
resolve=lambda ctx, read_models, proposals: resolve_recurring_due_slots(ctx),
|
|
)
|
|
)
|
|
register_agent_slot(
|
|
AgentSlotProvider(
|
|
slot_key="review.guardrail",
|
|
requires_elements=frozenset({"gate_fulfillment"}),
|
|
requires_data_slices=frozenset({"reviews", "roadmap"}),
|
|
resolve=lambda ctx, read_models, proposals: resolve_guardrail_review_slots(
|
|
ctx, read_models
|
|
),
|
|
)
|
|
)
|
|
|
|
|
|
_register_builtin_agent_slots()
|