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>
106 lines
3.4 KiB
Python
106 lines
3.4 KiB
Python
"""Gate / execution graph attention — planning_debt, execution_waiting."""
|
|
|
|
from __future__ import annotations
|
|
|
|
from steering.attention.context import AttentionEvalContext
|
|
from steering.attention.contributors.registry import (
|
|
AttentionContributor,
|
|
register_attention_contributor,
|
|
)
|
|
from steering.graph.execution_engine import (
|
|
execution_waiting_to_attention_items,
|
|
planning_debt_to_attention_items,
|
|
)
|
|
|
|
_GATE_METHODS = frozenset(
|
|
{"sequential_dependency", "program_delivery", "maturity_progression"}
|
|
)
|
|
|
|
|
|
def _planning_debt(ctx: AttentionEvalContext) -> list[dict]:
|
|
if ctx.binding.primary_method_key not in _GATE_METHODS:
|
|
return []
|
|
debts = ctx.read_models.get("planning_debt")
|
|
if debts is None:
|
|
from steering.graph.execution_engine import compute_planning_debt
|
|
|
|
gate_scope = ctx.gate_scope
|
|
scoped_roadmap = ctx.roadmap_items
|
|
scoped_actions = ctx.actions
|
|
if gate_scope:
|
|
scoped_roadmap = [
|
|
ri for ri in ctx.roadmap_items if str(ri.get("id")) == str(gate_scope)
|
|
]
|
|
scoped_actions = [
|
|
a
|
|
for a in ctx.actions
|
|
if a.get("roadmap_item_id")
|
|
and str(a["roadmap_item_id"]) == str(gate_scope)
|
|
]
|
|
debts = compute_planning_debt(
|
|
actions=scoped_actions, roadmap_items=scoped_roadmap
|
|
)
|
|
return planning_debt_to_attention_items(
|
|
initiative_id=ctx.initiative_id, debts=debts or []
|
|
)
|
|
|
|
|
|
def _execution_waiting(ctx: AttentionEvalContext) -> list[dict]:
|
|
if "critical_path" not in ctx.steering_elements and "gate_fulfillment" not in ctx.steering_elements:
|
|
return []
|
|
return execution_waiting_to_attention_items(
|
|
initiative_id=ctx.initiative_id,
|
|
actions=ctx.actions,
|
|
graph_state=ctx.graph_state,
|
|
)
|
|
|
|
|
|
def _horizon_no_ready_work(ctx: AttentionEvalContext) -> list[dict]:
|
|
if ctx.next_work:
|
|
return []
|
|
if ctx.binding.primary_method_key not in _GATE_METHODS:
|
|
return []
|
|
if ctx.horizon.kind != "gate":
|
|
return []
|
|
|
|
debts = ctx.read_models.get("planning_debt") or []
|
|
has_planning = len(debts) > 0
|
|
has_waiting = bool(ctx.graph_state.get("blocked_actions"))
|
|
if not has_planning and not has_waiting and not ctx.graph_state.get("ready_actions"):
|
|
return [
|
|
{
|
|
"kind": "initiative_without_next_action",
|
|
"severity": "info",
|
|
"title": ctx.horizon.gate_title or "Aktiver Horizont",
|
|
"summary": "Keine ausführungsbereite Arbeit im aktiven Horizont",
|
|
"scope_type": "milestone",
|
|
"scope_id": ctx.horizon.gate_roadmap_item_id,
|
|
"initiative_id": ctx.initiative_id,
|
|
"reason_code": "horizon_no_ready_work",
|
|
}
|
|
]
|
|
return []
|
|
|
|
|
|
register_attention_contributor(
|
|
AttentionContributor(
|
|
key="planning_debt",
|
|
requires_elements=frozenset({"gate_fulfillment"}),
|
|
contribute=_planning_debt,
|
|
)
|
|
)
|
|
register_attention_contributor(
|
|
AttentionContributor(
|
|
key="execution_waiting",
|
|
requires_any_element=frozenset({"critical_path", "gate_fulfillment"}),
|
|
contribute=_execution_waiting,
|
|
)
|
|
)
|
|
register_attention_contributor(
|
|
AttentionContributor(
|
|
key="horizon_no_ready_work",
|
|
requires_elements=frozenset({"gate_fulfillment"}),
|
|
contribute=_horizon_no_ready_work,
|
|
)
|
|
)
|