Lars
20636f88d2
All checks were successful
Deploy Development / deploy (push) Successful in 47s
Test Suite / pytest-backend (push) Successful in 4m11s
Test Suite / lint-backend (push) Successful in 2s
Test Suite / compose-smoke (push) Has been skipped
Test Suite / k6 /api/health Baseline (push) Successful in 18s
Test Suite / playwright-smoke (push) Successful in 15s
Erweitert evaluate_steering um read_models und proposals, migriert epic_rollup auf Registry, liefert sprint_commit-Vorschlaege und UI auf Plan-Sprint. Co-authored-by: Cursor <cursoragent@cursor.com>
97 lines
3.0 KiB
Python
97 lines
3.0 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)
|
|
|
|
@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 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
|
|
|
|
|
|
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,
|
|
)
|