plan_status.py

Go to the documentation of this file.

#!/usr/bin/env python3
# -*- coding: utf-8 -*-
# oc_chat/registry/render/plan_status.py
 
# OCT-LINT: disable=tests_exist reason="Covered by oc_chat/tests/registry/test_render_plan_status.py"
"""
Renderer for per-worktree ``.PLAN_STATUS.md`` files.
 
Purpose
-------
Render the registry's current plan assignment into the per-worktree status
file used by workers and the legacy plan-status helper.
 
Responsibilities
----------------
- Map registry lifecycle statuses to the legacy `.PLAN_STATUS.md` enum.
- Render valid schema-v1.0 YAML frontmatter for a worktree.
- Write status files through the shared atomic artifact helper.
 
Diagnostics
-----------
Domain: OC_CHAT.MAIN
Levels:
    L2 - plan-status render lifecycle events
    L3 - artifact drift details
    L4 - per-field projection tracing
 
Contracts
---------
- Output validates with `oc_chat.scripts.plan_status.read`.
- No wall-clock values are invented by the renderer.
 
Dependencies
------------
- oc_diagnostics (structured debug logging via _dbg)
- yaml (frontmatter emission)
- oc_chat.registry.render._common (projection and write helpers)
- oc_chat.registry.api (Registry facade the plan row is read from)
"""
from __future__ import annotations
 
from pathlib import Path
 
from oc_diagnostics import _dbg
import yaml
 
from oc_chat.registry.api import Registry
from oc_chat.registry.render._common import (
    RenderResult,
    agent_letter,
    plan_for_worktree,
    write_text_artifact,
)
 
 
_dbg("OC_CHAT.MAIN", "module", "loaded", 4)
 
STATUS_FILENAME = ".PLAN_STATUS.md"
 
_STATUS_TO_LEGACY = {
    "draft": "Queued",
    "pending": "Queued",
    "implementing": "Implementing",
    "ready_for_merge": "Completed",
    "integrating": "Completed",
    "completed": "Completed",
    "partial": "Partially Completed",
    "failed": "Failed",
    "superseded": "Failed",
}
 
 
def render_plan_status(registry: Registry, *, worktree: Path) -> str:
    """Render the status file for the plan assigned to ``worktree``."""
 
    row = plan_for_worktree(registry, worktree)
    if row is None:
        return _render_empty(worktree)
    status = _STATUS_TO_LEGACY.get(str(row["status"]), "Queued")
    data: dict[str, object] = {
        "schema_version": "1.0",
        "plan_id": row["plan_id"],
        "branch": row.get("branch") or "",
        "worktree": row.get("worktree") or str(worktree),
        "agent": agent_letter(row.get("assigned_slot")),
        "model": row.get("model_override") or "gpt-5.5",
        "thinking": _thinking(row.get("thinking_override")),
        "session": _session(row.get("session_override")),
        "status": status,
        "progress": {
            "steps_total": None,
            "steps_done": None,
            "current_step": _current_step(status),
        },
    }
    if row.get("created_at"):
        data["queued_at"] = str(row["created_at"])
    if status == "Implementing" and row.get("updated_at"):
        data["started_at"] = str(row["updated_at"])
    if status in {"Completed", "Partially Completed", "Failed"}:
        data["finished_at"] = row.get("body_observed_at") or row.get("updated_at")
    body = f"# {row['plan_id']}\n\nRegistry-rendered status snapshot.\n"
    return _frontmatter(data, body)
 
 
def write_plan_status(
    registry: Registry,
    *,
    worktree: Path,
    prefer_local: bool = False,
    dry_run: bool = False,
) -> RenderResult:
    """Write ``.PLAN_STATUS.md`` for ``worktree``."""
 
    return write_text_artifact(
        worktree / STATUS_FILENAME,
        render_plan_status(registry, worktree=worktree),
        artifact_kind="plan_status",
        prefer_local=prefer_local,
        dry_run=dry_run,
    )
 
 
def _render_empty(worktree: Path) -> str:
    data: dict[str, object] = {
        "schema_version": "1.0",
        "plan_id": "(none)",
        "branch": "",
        "worktree": str(worktree),
        "agent": "B",
        "model": "gpt-5.5",
        "thinking": "high",
        "session": "either",
        "status": "Queued",
        "progress": {
            "steps_total": None,
            "steps_done": None,
            "current_step": "No assigned registry plan",
        },
    }
    return _frontmatter(data, "# No Assigned Plan\n\nRegistry has no plan for this worktree.\n")
 
 
def _frontmatter(data: dict[str, object], body: str) -> str:
    yaml_text = yaml.safe_dump(
        data,
        sort_keys=False,
        allow_unicode=False,
        default_flow_style=False,
    ).rstrip()
    return f"---\n{yaml_text}\n---\n\n{body.rstrip()}\n"
 
 
def _thinking(value: object) -> str:
    text = "high" if value is None else str(value)
    return text if text in {"off", "low", "medium", "high", "max"} else "high"
 
 
def _session(value: object) -> str:
    text = "either" if value is None else str(value)
    return text if text in {"fresh", "continue", "either"} else "either"
 
 
def _current_step(status: str) -> str:
    if status == "Implementing":
        return "Implementing"
    if status == "Queued":
        return "Queued"
    return "Completed - ready to merge"