plan_header.py

Go to the documentation of this file.

#!/usr/bin/env python3
# -*- coding: utf-8 -*-
# oc_chat/registry/render/plan_header.py
 
# OCT-LINT: disable=tests_exist reason="Covered by oc_chat/tests/registry/test_render_plan_header.py"
"""
Renderer for the registry-owned header block of plan Markdown files.
 
Purpose
-------
Materialise current registry metadata into the leading header block of plan
Markdown files while preserving the authored body.
 
Responsibilities
----------------
- Render deterministic branch/worktree/status/assignment metadata.
- Replace the leading H1/header block without touching the body section.
- Write plan files through the shared atomic artifact helper.
 
Diagnostics
-----------
Domain: OC_CHAT.MAIN
Levels:
    L2 - plan-header render lifecycle events
    L3 - artifact drift details
    L4 - header projection tracing
 
Contracts
---------
- The first `##` section and everything after it is preserved.
- Missing plan ids raise `KeyError`.
 
Dependencies
------------
- oc_diagnostics (structured debug logging via _dbg)
- oc_chat.registry.render._common (projection and write helpers)
- oc_chat.registry.api (Registry facade the plan rows are read from)
"""
from __future__ import annotations
 
from pathlib import Path
 
from oc_diagnostics import _dbg
 
from oc_chat.registry.api import Registry
from oc_chat.registry.render._common import (
    RenderResult,
    fetch_plan_rows,
    slot_label,
    write_text_artifact,
)
 
 
_dbg("OC_CHAT.MAIN", "module", "loaded", 4)
 
 
def render_plan_header(registry: Registry, *, plan_id: str) -> str:
    """Render a deterministic Markdown header for ``plan_id``."""
 
    row = _plan_row(registry, plan_id)
    if row is None:
        raise KeyError(f"plan not found: {plan_id}")
    title = str(row.get("title") or plan_id)
    lines = [
        f"# {title}",
        "",
        f"**Plan ID:** `{row['plan_id']}`",
        f"**Branch:** `{row.get('branch') or ''}`",
        f"**Worktree:** `{row.get('worktree') or ''}`",
        f"**Plan Type:** `{row.get('plan_type') or 'feature'}`",
        f"**Status:** `{row.get('status')}`",
        f"**Assigned Slot:** `{slot_label(row.get('assigned_slot'))}`",
        "",
    ]
    return "\n".join(lines)
 
 
def write_plan_header(
    registry: Registry,
    *,
    plan_id: str,
    plan_file: Path,
    prefer_local: bool = False,
    dry_run: bool = False,
) -> RenderResult:
    """Replace the leading header block of ``plan_file``."""
 
    header = render_plan_header(registry, plan_id=plan_id)
    if plan_file.exists():
        body = _body_without_header(plan_file.read_text(encoding="utf-8"))
    else:
        body = ""
    text = f"{header}{body.lstrip()}"
    return write_text_artifact(
        plan_file,
        text,
        artifact_kind="plan_header",
        prefer_local=prefer_local,
        dry_run=dry_run,
    )
 
 
def _body_without_header(text: str) -> str:
    normalized = text.replace("\r\n", "\n").replace("\r", "\n")
    lines = normalized.splitlines()
    if not lines or not lines[0].startswith("# "):
        return normalized
    for index, line in enumerate(lines[1:], start=1):
        if line.startswith("## "):
            return "\n".join(lines[index:]).rstrip() + "\n"
    return ""
 
 
def _plan_row(registry: Registry, plan_id: str) -> dict[str, object] | None:
    for row in fetch_plan_rows(registry):
        if str(row["plan_id"]) == plan_id:
            return row
    return None