Lars
b747200f3a
All checks were successful
Deploy Development / deploy (push) Successful in 46s
Test Suite / pytest-backend (push) Successful in 4m13s
Test Suite / lint-backend (push) Successful in 3s
Test Suite / compose-smoke (push) Has been skipped
Test Suite / k6 /api/health Baseline (push) Successful in 19s
Test Suite / playwright-smoke (push) Successful in 12s
Methodenuebergreifende Provider auf Kernel v0.3; verbindliche Coding Rules fuer Agenten in ADP und .cursor/rules. Co-authored-by: Cursor <cursoragent@cursor.com>
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
|