Kairo-Jinkendo/backend/steering/proposals/gate_next_actions.py
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
feat(steering): K-Ext-2 Attention-Registry, Gate/Intake-Proposals, generische UI
Methodenuebergreifende Provider auf Kernel v0.3; verbindliche Coding Rules fuer Agenten in ADP und .cursor/rules.

Co-authored-by: Cursor <cursoragent@cursor.com>
2026-07-27 11:48:10 +02:00

128 lines
4.4 KiB
Python

"""Gate next actions proposal — sequential_dependency / program_delivery."""
from __future__ import annotations
from typing import Any
from steering.eval_context import SteeringEvalContext
COMMITTABLE = frozenset({"new", "triaged", "accepted"})
_PRIORITY = {"critical": 0, "high": 1, "normal": 2, "low": 3}
def _active_gate_id(ctx: SteeringEvalContext) -> str | None:
if ctx.horizon.kind == "gate" and ctx.horizon.gate_roadmap_item_id:
return str(ctx.horizon.gate_roadmap_item_id)
debts = []
return None
def _sort_key(item: dict[str, Any]) -> tuple:
return (
_PRIORITY.get(item.get("priority") or "normal", 9),
item.get("sort_order") if item.get("sort_order") is not None else 9999,
(item.get("title") or "").lower(),
)
def propose_gate_next_actions(
ctx: SteeringEvalContext,
*,
read_models: dict[str, Any] | None = None,
) -> list[dict[str, Any]]:
read_models = read_models or {}
gate_id = _active_gate_id(ctx)
proposals: list[dict[str, Any]] = []
rank = 0
for debt in read_models.get("planning_debt") or []:
rank += 1
gid = str(debt.get("roadmap_item_id"))
proposals.append(
{
"proposal_key": "gate_next_actions",
"scope_type": "roadmap_item",
"scope_id": gid,
"rank": rank,
"reason_code": "planning_debt",
"summary": debt.get("message") or "Gate ohne Durchführungsplan",
"title": debt.get("title") or "Zielzustand",
"roadmap_item_id": gid,
"initiative_id": ctx.initiative_id,
"ranker_key": "heuristic_v0",
"confidence": "heuristic",
"factors": [{"code": "planning_debt", "weight": 100, "label": "Planning Debt"}],
"dependency_refs": [],
"dependency_blocked": False,
"data_source": "steering_kernel",
}
)
vocabulary = ctx.backlog_vocabulary
convertible = frozenset(vocabulary.get("convertible_kinds") or ["story", "bug", "issue"])
candidates = [
item
for item in ctx.backlog_items
if item.get("status") in COMMITTABLE
and item.get("item_kind") != "epic"
and (item.get("item_kind") or "story") in convertible
and (not gate_id or str(item.get("roadmap_item_id") or "") == gate_id)
]
for item in sorted(candidates, key=_sort_key)[:15]:
rank += 1
proposals.append(
{
"proposal_key": "gate_next_actions",
"scope_type": "backlog_item",
"scope_id": str(item["id"]),
"rank": rank,
"reason_code": "gate_backlog_ready",
"summary": "Eingang-Item am aktiven Gate committen",
"title": item.get("title") or "Backlog-Item",
"item_kind": item.get("item_kind") or "story",
"priority": item.get("priority") or "normal",
"roadmap_item_id": (
str(item["roadmap_item_id"]) if item.get("roadmap_item_id") else gate_id
),
"initiative_id": ctx.initiative_id,
"ranker_key": "heuristic_v0",
"confidence": "heuristic",
"factors": [
{
"code": "gate_alignment",
"weight": 50,
"label": "Gate-Zuordnung",
}
],
"dependency_refs": [],
"dependency_blocked": False,
"data_source": "steering_kernel",
}
)
return proposals
def gate_next_actions_attention_items(
*,
initiative_id: str,
proposals: list[dict[str, Any]],
) -> list[dict[str, Any]]:
backlog_ready = [p for p in proposals if p.get("scope_type") == "backlog_item"]
if not backlog_ready:
return []
return [
{
"kind": "gate_commit_suggested",
"severity": "info",
"title": "Gate-Planung",
"summary": f"{len(backlog_ready)} Item(s) für Gate-Commit vorgeschlagen",
"scope_type": "initiative",
"scope_id": initiative_id,
"initiative_id": initiative_id,
"reason_code": "gate_commit_suggested",
"data_source": "gate_next_actions",
}
]