Kairo-Jinkendo/backend/steering/proposals/sprint_commit.py
Lars 1bbc5dbfce
All checks were successful
Deploy Development / deploy (push) Successful in 46s
Test Suite / pytest-backend (push) Successful in 4m8s
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 19s
Test Suite / playwright-smoke (push) Successful in 13s
refactor(steering): Sprint-Vorschlaege mit austauschbarem Ranker und Abhaengigkeiten
Keine feste Bug-zuerst-Policy: heuristic_v0 als Fallback, agent_v1-Slot vorbereitet, parent_action-Abhaengigkeiten und factors[] fuer spaetere KI-Steuerung.

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

122 lines
3.9 KiB
Python

"""Sprint commit proposals — ADP P6 / pluggable ranking."""
from __future__ import annotations
from typing import Any
from steering.eval_context import SteeringEvalContext
# Ranker registration side effects
import steering.proposals.sprint_commit_agent_v1 # noqa: F401
import steering.proposals.sprint_commit_heuristic_v0 # noqa: F401
from steering.proposals.sprint_commit_context import (
build_sprint_commit_candidates,
resolve_sprint_commit_ranker,
resolve_target_work_cycle,
)
def propose_sprint_commit(
ctx: SteeringEvalContext,
*,
read_models: dict[str, Any] | None = None,
) -> list[dict[str, Any]]:
"""Ranked backlog items for sprint commit — ranker is swappable, no auto-mutation."""
read_models = read_models or {}
target = resolve_target_work_cycle(ctx)
if not target:
return []
cycle_id = str(target["id"])
cycle_title = target.get("title") or "Sprint"
candidates = build_sprint_commit_candidates(ctx, target_work_cycle_id=cycle_id)
if not candidates:
return []
ranker = resolve_sprint_commit_ranker(ctx)
ranked = ranker.rank(
ctx,
candidates=candidates,
read_models=read_models,
target_work_cycle_id=cycle_id,
)
proposals: list[dict[str, Any]] = []
for row in ranked:
item = row.candidate.item
proposals.append(
{
"proposal_key": "sprint_commit",
"scope_type": "backlog_item",
"scope_id": str(item["id"]),
"rank": row.rank,
"score": row.score,
"reason_code": row.reason_code,
"summary": row.summary,
"title": item.get("title") or "Backlog-Item",
"item_kind": item.get("item_kind") or "story",
"priority": item.get("priority") or "normal",
"work_cycle_id": cycle_id,
"work_cycle_title": cycle_title,
"initiative_id": ctx.initiative_id,
"ranker_key": row.ranker_key,
"confidence": "agent" if row.ranker_key.startswith("agent") else "heuristic",
"factors": row.factors,
"dependency_refs": row.candidate.dependency_refs,
"dependency_blocked": row.candidate.dependency_blocked,
"data_source": "steering_kernel",
}
)
return proposals
def sprint_commit_attention_items(
*,
initiative_id: str,
proposals: list[dict[str, Any]],
actions: list[dict[str, Any]],
) -> list[dict[str, Any]]:
"""Attention when sprint has room for proposed commits."""
from steering.proposals.sprint_commit_context import OPEN_ACTION
if not proposals:
return []
target_id = proposals[0].get("work_cycle_id")
if not target_id:
return []
sprint_actions = [
a
for a in actions
if a.get("work_cycle_id") and str(a["work_cycle_id"]) == str(target_id)
]
open_in_sprint = sum(1 for a in sprint_actions if a.get("status") in OPEN_ACTION)
ready = [p for p in proposals if not p.get("dependency_blocked")]
items: list[dict[str, Any]] = []
cycle_title = proposals[0].get("work_cycle_title") or "Sprint"
if open_in_sprint == 0 and ready:
ranker = proposals[0].get("ranker_key", "heuristic_v0")
items.append(
{
"kind": "sprint_commit_suggested",
"severity": "info",
"title": cycle_title,
"summary": (
f"{len(ready)} commit-bereite Item(s) — "
f"Vorschlag via Ranker {ranker}"
),
"scope_type": "work_cycle",
"scope_id": str(target_id),
"initiative_id": initiative_id,
"reason_code": "sprint_commit_suggested",
"data_source": "sprint_commit_proposal",
}
)
return items