110 lines
3.9 KiB
Python
110 lines
3.9 KiB
Python
|
"""Attention evaluation context — shared caches for contributors."""
|
||
|
|
|
||
|
|
from __future__ import annotations
|
||
|
|
|
||
|
|
from dataclasses import dataclass, field
|
||
|
|
from typing import Any
|
||
|
|
|
||
|
|
from steering.eval_context import SteeringEvalContext
|
||
|
|
from steering.kernel.models import HorizonMarker, SteeringBinding
|
||
|
|
from tenant_context import TenantContext
|
||
|
|
|
||
|
|
|
||
|
|
@dataclass
|
||
|
|
class AttentionEvalContext:
|
||
|
|
tenant_ctx: TenantContext
|
||
|
|
initiative_id: str
|
||
|
|
binding: SteeringBinding
|
||
|
|
horizon: HorizonMarker
|
||
|
|
next_work: list[dict[str, Any]]
|
||
|
|
read_models: dict[str, Any]
|
||
|
|
proposals: dict[str, Any]
|
||
|
|
eval_ctx: SteeringEvalContext | None = None
|
||
|
|
|
||
|
|
_actions: list[dict[str, Any]] | None = field(default=None, repr=False)
|
||
|
|
_blockers: list[dict[str, Any]] | None = field(default=None, repr=False)
|
||
|
|
_roadmap_items: list[dict[str, Any]] | None = field(default=None, repr=False)
|
||
|
|
_dependencies: list[dict[str, Any]] | None = field(default=None, repr=False)
|
||
|
|
_graph_state: dict[str, Any] | None = field(default=None, repr=False)
|
||
|
|
_gate_scope: str | None = field(default=None, repr=False)
|
||
|
|
|
||
|
|
@property
|
||
|
|
def steering_elements(self) -> frozenset[str]:
|
||
|
|
if self.eval_ctx:
|
||
|
|
return self.eval_ctx.steering_elements
|
||
|
|
return frozenset()
|
||
|
|
|
||
|
|
@property
|
||
|
|
def actions(self) -> list[dict[str, Any]]:
|
||
|
|
if self._actions is None:
|
||
|
|
if self.eval_ctx:
|
||
|
|
self._actions = self.eval_ctx.actions
|
||
|
|
else:
|
||
|
|
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 blockers(self) -> list[dict[str, Any]]:
|
||
|
|
if self._blockers is None:
|
||
|
|
from services import blockers as blocker_service
|
||
|
|
|
||
|
|
self._blockers = blocker_service.list_blockers_for_initiative(
|
||
|
|
tenant_id=self.tenant_ctx.tenant_id, initiative_id=self.initiative_id
|
||
|
|
)
|
||
|
|
return self._blockers
|
||
|
|
|
||
|
|
@property
|
||
|
|
def roadmap_items(self) -> list[dict[str, Any]]:
|
||
|
|
if self._roadmap_items is None:
|
||
|
|
from services import roadmap as roadmap_service
|
||
|
|
|
||
|
|
self._roadmap_items = roadmap_service.list_roadmap_items_for_initiative(
|
||
|
|
tenant_id=self.tenant_ctx.tenant_id, initiative_id=self.initiative_id
|
||
|
|
)
|
||
|
|
return self._roadmap_items
|
||
|
|
|
||
|
|
@property
|
||
|
|
def gate_scope(self) -> str | None:
|
||
|
|
if self._gate_scope is None:
|
||
|
|
from steering.strategies.next_action.execution_ready import (
|
||
|
|
gate_horizon_scope_for_method,
|
||
|
|
)
|
||
|
|
|
||
|
|
self._gate_scope = gate_horizon_scope_for_method(
|
||
|
|
self.binding.primary_method_key,
|
||
|
|
self.tenant_ctx,
|
||
|
|
self.initiative_id,
|
||
|
|
)
|
||
|
|
return self._gate_scope
|
||
|
|
|
||
|
|
@property
|
||
|
|
def dependencies(self) -> list[dict[str, Any]]:
|
||
|
|
if self._dependencies is None:
|
||
|
|
from services.execution_plan import list_dependencies_for_initiative
|
||
|
|
|
||
|
|
self._dependencies = list_dependencies_for_initiative(
|
||
|
|
tenant_id=self.tenant_ctx.tenant_id, initiative_id=self.initiative_id
|
||
|
|
)
|
||
|
|
return self._dependencies
|
||
|
|
|
||
|
|
@property
|
||
|
|
def graph_state(self) -> dict[str, Any]:
|
||
|
|
if self._graph_state is None:
|
||
|
|
cached = self.read_models.get("execution_graph")
|
||
|
|
if cached is not None:
|
||
|
|
self._graph_state = cached
|
||
|
|
else:
|
||
|
|
from steering.graph.execution_engine import compute_execution_graph_state
|
||
|
|
|
||
|
|
self._graph_state = compute_execution_graph_state(
|
||
|
|
actions=self.actions,
|
||
|
|
dependencies=self.dependencies,
|
||
|
|
scope_roadmap_item_id=self.gate_scope,
|
||
|
|
)
|
||
|
|
return self._graph_state
|