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>
125 lines
4.1 KiB
Python
125 lines
4.1 KiB
Python
"""Shared evaluation context for steering read models and proposals (K-Ext-1)."""
|
|
|
|
from __future__ import annotations
|
|
|
|
from dataclasses import dataclass, field
|
|
from typing import Any
|
|
|
|
from steering.kernel.models import HorizonMarker, LifecycleContext, SteeringBinding
|
|
from tenant_context import TenantContext
|
|
|
|
|
|
@dataclass
|
|
class SteeringEvalContext:
|
|
"""Tenant-scoped context — lazy OM loads for extension providers."""
|
|
|
|
tenant_ctx: TenantContext
|
|
initiative_id: str
|
|
binding: SteeringBinding
|
|
horizon: HorizonMarker
|
|
lifecycle: LifecycleContext
|
|
operating_context: dict[str, Any]
|
|
|
|
_actions: list[dict[str, Any]] | None = field(default=None, repr=False)
|
|
_backlog_items: list[dict[str, Any]] | None = field(default=None, repr=False)
|
|
_work_cycles: list[dict[str, Any]] | None = field(default=None, repr=False)
|
|
_reviews: list[dict[str, Any]] | None = field(default=None, repr=False)
|
|
_recurring_elements: list[dict[str, Any]] | None = field(default=None, repr=False)
|
|
|
|
@property
|
|
def steering_elements(self) -> frozenset[str]:
|
|
return frozenset(self.operating_context.get("steering_elements") or [])
|
|
|
|
@property
|
|
def data_slices(self) -> list[str]:
|
|
return list(self.operating_context.get("data_slices") or [])
|
|
|
|
@property
|
|
def backlog_vocabulary(self) -> dict[str, Any]:
|
|
return dict(self.operating_context.get("backlog_vocabulary") or {})
|
|
|
|
@property
|
|
def agent_slot_config(self) -> dict[str, Any]:
|
|
return dict(self.operating_context.get("agent_slot_config") or {})
|
|
|
|
@property
|
|
def actions(self) -> list[dict[str, Any]]:
|
|
if self._actions is None:
|
|
from services import actions as action_service
|
|
|
|
self._actions = action_service.list_actions_for_initiative(
|
|
tenant_id=self.tenant_ctx.tenant_id,
|
|
initiative_id=self.initiative_id,
|
|
)
|
|
return self._actions
|
|
|
|
@property
|
|
def backlog_items(self) -> list[dict[str, Any]]:
|
|
if self._backlog_items is None:
|
|
from services import backlog as backlog_service
|
|
|
|
self._backlog_items = backlog_service.list_backlog_for_initiative(
|
|
tenant_id=self.tenant_ctx.tenant_id,
|
|
initiative_id=self.initiative_id,
|
|
)
|
|
return self._backlog_items
|
|
|
|
@property
|
|
def work_cycles(self) -> list[dict[str, Any]]:
|
|
if self._work_cycles is None:
|
|
from services.work_cycle import list_work_cycles_for_initiative
|
|
|
|
self._work_cycles = list_work_cycles_for_initiative(
|
|
tenant_id=self.tenant_ctx.tenant_id,
|
|
initiative_id=self.initiative_id,
|
|
)
|
|
return self._work_cycles
|
|
|
|
@property
|
|
def reviews(self) -> list[dict[str, Any]]:
|
|
if self._reviews is None:
|
|
from services.reviews import list_reviews_for_initiative
|
|
|
|
self._reviews = list_reviews_for_initiative(
|
|
tenant_id=self.tenant_ctx.tenant_id,
|
|
initiative_id=self.initiative_id,
|
|
)
|
|
return self._reviews
|
|
|
|
@property
|
|
def recurring_elements(self) -> list[dict[str, Any]]:
|
|
if self._recurring_elements is None:
|
|
from services.recurring import list_recurring_for_initiative
|
|
|
|
self._recurring_elements = list_recurring_for_initiative(
|
|
tenant_id=self.tenant_ctx.tenant_id,
|
|
initiative_id=self.initiative_id,
|
|
)
|
|
return self._recurring_elements
|
|
|
|
|
|
def build_steering_eval_context(
|
|
tenant_ctx: TenantContext,
|
|
*,
|
|
initiative_id: str,
|
|
binding: SteeringBinding,
|
|
horizon: HorizonMarker,
|
|
lifecycle: LifecycleContext,
|
|
) -> SteeringEvalContext:
|
|
from services.operating_context import get_operating_context
|
|
|
|
operating = (
|
|
get_operating_context(
|
|
tenant_id=tenant_ctx.tenant_id, initiative_id=initiative_id
|
|
)
|
|
or {}
|
|
)
|
|
return SteeringEvalContext(
|
|
tenant_ctx=tenant_ctx,
|
|
initiative_id=initiative_id,
|
|
binding=binding,
|
|
horizon=horizon,
|
|
lifecycle=lifecycle,
|
|
operating_context=operating,
|
|
)
|