68 lines
1.9 KiB
Python
68 lines
1.9 KiB
Python
|
"""Horizon resolution (Q1)."""
|
||
|
|
|
||
|
|
from __future__ import annotations
|
||
|
|
|
||
|
|
from steering.kernel.models import HorizonMarker
|
||
|
|
from steering.strategies.next_action.execution_ready import (
|
||
|
|
first_active_gate_id,
|
||
|
|
gate_horizon_scope_for_method,
|
||
|
|
)
|
||
|
|
from services.work_cycle import get_active_work_cycle
|
||
|
|
from tenant_context import TenantContext
|
||
|
|
|
||
|
|
|
||
|
|
def resolve_horizon(
|
||
|
|
ctx: TenantContext,
|
||
|
|
*,
|
||
|
|
initiative_id: str,
|
||
|
|
primary_method_key: str,
|
||
|
|
composition_modifier: str | None,
|
||
|
|
) -> HorizonMarker:
|
||
|
|
gate_id = gate_horizon_scope_for_method(
|
||
|
|
primary_method_key, ctx, initiative_id
|
||
|
|
)
|
||
|
|
gate_title: str | None = None
|
||
|
|
if gate_id:
|
||
|
|
gate_title = _load_gate_title(ctx, gate_id)
|
||
|
|
|
||
|
|
active_cycle = get_active_work_cycle(
|
||
|
|
tenant_id=ctx.tenant_id, initiative_id=initiative_id
|
||
|
|
)
|
||
|
|
if composition_modifier and active_cycle:
|
||
|
|
return HorizonMarker(
|
||
|
|
kind="work_cycle",
|
||
|
|
gate_roadmap_item_id=gate_id,
|
||
|
|
gate_title=gate_title,
|
||
|
|
work_cycle_id=str(active_cycle["id"]),
|
||
|
|
work_cycle_title=active_cycle.get("title"),
|
||
|
|
)
|
||
|
|
|
||
|
|
if gate_id:
|
||
|
|
return HorizonMarker(
|
||
|
|
kind="gate",
|
||
|
|
gate_roadmap_item_id=gate_id,
|
||
|
|
gate_title=gate_title,
|
||
|
|
)
|
||
|
|
|
||
|
|
return HorizonMarker(kind="none")
|
||
|
|
|
||
|
|
|
||
|
|
def _load_gate_title(ctx: TenantContext, gate_id: str) -> str | None:
|
||
|
|
from db import get_connection
|
||
|
|
from psycopg2.extras import RealDictCursor
|
||
|
|
|
||
|
|
conn = get_connection()
|
||
|
|
try:
|
||
|
|
with conn.cursor(cursor_factory=RealDictCursor) as cur:
|
||
|
|
cur.execute(
|
||
|
|
"""
|
||
|
|
SELECT title FROM roadmap_items
|
||
|
|
WHERE id = %s AND tenant_id = %s
|
||
|
|
""",
|
||
|
|
(gate_id, ctx.tenant_id),
|
||
|
|
)
|
||
|
|
row = cur.fetchone()
|
||
|
|
return row["title"] if row else None
|
||
|
|
finally:
|
||
|
|
conn.close()
|