sections.py

Go to the documentation of this file.

#!/usr/bin/env python3
# -*- coding: utf-8 -*-
# oc_chat/registry/render/roadmap/sections.py
 
# OCT-LINT: disable=tests_exist reason="Covered by oc_chat/tests/registry/roadmap/test_render_living_set.py"
"""
Per-section body renderers shared by the lean V4 + EXTENDED V3 views (P1.6, §6).
 
Purpose
-------
Render each ``roadmap_render_sections`` body section by ``query_kind`` — the
milestone (m_phase) table, per-phase sub-phase detail, the non-phase item
tables, and the era overview — so the lean and EXTENDED views are produced
**deterministically from the seeded render-section manifest** (the P1.4↔P1.6
seam, umbrella §12 wave-3). All enum→glyph mapping flows through ``_common``
(§7, AC §8.7); all reads flow through ``store_read`` (the §8.30 guard).
 
Responsibilities
----------------
- ``render_section(conn, query_kind, section_path, params)`` — dispatch one
  manifest section to its body markdown (a list of lines).
- The four v1 body query kinds: ``prose_only`` (era overview),
  ``m_phase_table``, ``phase_section_with_subrows``, ``backlog_track_table`` /
  ``git_workflow_table`` (item tables).
 
Diagnostics
-----------
Domain: OC_CHAT.MAIN
Levels:
    L2 - section lifecycle
    L3 - section selection
    L4 - per-row rendering
 
Contracts
---------
- Deterministic: identical store state → identical bytes (no wall-clock).
- A live-scope phase-detail section emits detail only for non-terminal phases;
  an ``all`` scope emits every phase (DESIGN §3 lean-vs-EXTENDED split).
 
Dependencies
------------
- oc_diagnostics (structured debug logging via _dbg)
- oc_chat.registry.render.roadmap.store_read (guarded reads of eras, milestones, sub-items and non-phase items)
- oc_chat.registry.render.roadmap._common (status and priority glyphs, status labels and terminal statuses)
"""
from __future__ import annotations
 
import sqlite3
from typing import Any, Optional
 
from oc_diagnostics import _dbg
 
from oc_chat.registry.render.roadmap import store_read
from oc_chat.registry.render.roadmap._common import (
    TERMINAL_STATUSES,
    priority_glyph,
    status_glyph,
    status_label,
)
 
_dbg("OC_CHAT.MAIN", "module", "loaded", 4)
 
_MILESTONE_HEADER = (
    "| Phase | Priority | State | Done | Depends On | Title | Purpose |\n"
    "|-------|----------|-------|------|------------|-------|---------|"
)
_ITEM_HEADER = (
    "| State | Priority | Item | Title | Purpose |\n"
    "|-------|----------|------|-------|---------|"
)
_SUBROW_HEADER = "| Sub-phase | State | Description |\n|-----------|-------|-------------|"
_ERA_HEADER = (
    "| Era | Phases | Started | Ended | Note |\n"
    "|-----|--------|---------|-------|------|"
)
 
 
def escape(value: object) -> str:
    """Collapse a cell value to a single safe markdown-table line."""
    text = "" if value is None else str(value)
    return text.replace("|", "\\|").replace("\n", " ").strip()
 
 
def _done_cell(done: int, total: int) -> str:
    return f"{done}/{total}" if total else "—"
 
 
def render_section(
    conn: sqlite3.Connection,
    *,
    query_kind: str,
    section_path: str,
    params: Optional[dict[str, Any]],
) -> list[str]:
    """Render one manifest section's body markdown (heading + content)."""
    func = "render_section"
    params = params or {}
    if query_kind == "prose_only" and section_path == "era-overview":
        return _era_overview(conn)
    if query_kind == "m_phase_table":
        return _milestone_table(conn)
    if query_kind == "phase_section_with_subrows":
        return _phase_detail(conn, scope=str(params.get("scope", "all")))
    if query_kind in ("backlog_track_table", "git_workflow_table"):
        return _item_tables(conn, query_kind, params)
    _dbg("OC_CHAT.MAIN", func, f"unhandled {query_kind}/{section_path}", 2)
    return []
 
 
def _era_overview(conn: sqlite3.Connection) -> list[str]:
    rows = store_read.era_rows(conn)
    if not rows:
        return []
    lines = ["## Era Overview", "", _ERA_HEADER]
    for row in rows:
        lines.append(
            "| {era} | {label} | {started} | {ended} | {note} |".format(
                era=escape(row["era_id"]),
                label=escape(row["era_label"]),
                started=escape(row["started_at"]),
                ended=escape(row["ended_at"] or "—"),
                note=escape(row["major_bump_reason"] or ""),
            )
        )
    return lines
 
 
def _milestone_table(conn: sqlite3.Connection) -> list[str]:
    milestones = store_read.milestone_rows(conn)
    if not milestones:
        return []
    lines = ["## Milestones", "", _MILESTONE_HEADER]
    for milestone in milestones:
        has_successor = milestone.successor is not None
        lines.append(
            "| **{phase}** | {prio} | {state} | {done} | {deps} | {title} | {purpose} |".format(
                phase=escape(milestone.phase_code),
                prio=priority_glyph(milestone.priority),
                state=status_glyph(milestone.status, has_successor=has_successor),
                done=_done_cell(milestone.done, milestone.total),
                deps=escape(", ".join(milestone.depends_on) or "—"),
                title=escape(milestone.title),
                purpose=escape(milestone.purpose),
            )
        )
    return lines
 
 
def _blocker_names(
    blocked_by: tuple[str, ...],
    status: str,
    titles: dict[str, str],
    statuses: dict[str, str],
) -> str:
    """The "🔒 Blocked by `<id> — <title>`[, ...]" annotation for a non-terminal
    item with blocks/gates predecessors. Only an *incomplete* predecessor is an
    active block — a satisfied (terminal) predecessor is filtered out (a done
    dependency does not block); empty for a terminal/unblocked item or one whose
    blockers are all satisfied."""
    if status in TERMINAL_STATUSES or not blocked_by:
        return ""
    active = [bid for bid in blocked_by if statuses.get(bid, "") not in TERMINAL_STATUSES]
    if not active:
        return ""
    named = ", ".join(f"`{bid} — {titles.get(bid, bid)}`" for bid in active)
    return f"🔒 Blocked by {named}"
 
 
def _phase_status_line(
    milestone: "store_read.Milestone", titles: dict[str, str], statuses: dict[str, str]
) -> str:
    """The phase status line. When the status is ``blocked`` AND a named blocker
    exists, the named blocker REPLACES the generic "🔒 Blocked" label (one lock):
    ``🔒 Blocked by `M2 — …` · **1/9** complete``. A ``blocked`` item with no named
    blocker keeps ``🔒 Blocked``; any other item appends the blocker after progress."""
    done = _done_cell(milestone.done, milestone.total)
    names = _blocker_names(milestone.depends_on, milestone.status, titles, statuses)
    if milestone.status == "blocked" and names:
        return f"{names} · **{done}** complete"
    glyph = status_glyph(milestone.status, has_successor=milestone.successor is not None)
    label = status_label(
        milestone.status, successor=milestone.successor, superseded_by=milestone.superseded_by
    )
    suffix = f" · {names}" if names else ""
    return f"{glyph} {label} · **{done}** complete{suffix}"
 
 
def _phase_detail(conn: sqlite3.Connection, *, scope: str) -> list[str]:
    milestones = store_read.milestone_rows(conn)
    titles = store_read.item_titles(conn)
    statuses = store_read.item_statuses(conn)
    blocks: list[str] = []
    for milestone in milestones:
        if scope == "live" and milestone.status in TERMINAL_STATUSES:
            continue
        children = store_read.sub_items(conn, milestone.phase_code)
        # The EXTENDED (scope="all") view emits the verbatim per-phase prose
        # (the legacy `# PHASE` section: metadata + the full deliverables table
        # with each item's status provenance + acceptance) when present; the lean
        # view keeps the compact structured sub-phase table (DESIGN §3 split).
        prose = (milestone.prose or "").strip() if scope == "all" else ""
        if not children and not prose:
            continue
        blocks.extend(
            [
                f"### {milestone.phase_code} — {escape(milestone.title)}",
                "",
                _phase_status_line(milestone, titles, statuses),
                "",
            ]
        )
        if prose:
            blocks.extend(prose.split("\n"))
            blocks.append("")
            continue
        blocks.append(_SUBROW_HEADER)
        for child in children:
            child_glyph = status_glyph(child.status, has_successor=child.successor is not None)
            child_names = _blocker_names(child.blocked_by, child.status, titles, statuses)
            child_suffix = f" · {child_names}" if child_names else ""
            blocks.append(
                "| {sub} | {state} | {desc} |".format(
                    sub=escape(child.item_id),
                    state=f"{child_glyph} {status_label(child.status, successor=child.successor, superseded_by=child.superseded_by)}",
                    desc=escape(f"{child.purpose}{child_suffix}"),
                )
            )
        blocks.append("")
    if not blocks:
        return []
    heading = "Live Phase Detail" if scope == "live" else "Phase Detail"
    return ["## " + heading, "", *blocks[:-1]]
 
 
def _item_tables(
    conn: sqlite3.Connection, query_kind: str, params: dict[str, Any]
) -> list[str]:
    if query_kind == "git_workflow_table":
        track_kinds: tuple[str, ...] = ("git_workflow",)
        heading = "Git Workflow"
    else:
        track_kinds = tuple(params.get("track_kinds", ()))
        heading = "Backlog & Non-Phase Items"
    items = store_read.non_phase_rows(conn, track_kinds)
    if not items:
        return []
    lines = [f"## {heading}", "", _ITEM_HEADER]
    for item in items:
        glyph = status_glyph(item.status, has_successor=item.successor is not None)
        lines.append(
            "| {state} | {prio} | {item} | {title} | {purpose} |".format(
                state=glyph,
                prio=priority_glyph(item.priority),
                item=escape(item.item_id),
                title=escape(item.title),
                purpose=escape(item.purpose),
            )
        )
    return lines
 
 
__all__ = ["escape", "render_section"]