plans.py

Go to the documentation of this file.

#!/usr/bin/env python3
# -*- coding: utf-8 -*-
# oc_orchestrator/plans.py
 
"""
High-level plan operations on top of the plan inventory.
 
Purpose
-------
Public functions that combine inventory writes with filename mirroring.
Every state-change operation follows the protocol:
 
    1. Inventory write FIRST (atomic).
    2. Filename rename SECOND.
    3. On filename failure, inventory still correct; next call's
       reconcile() heals the filename.
 
Responsibilities
----------------
- :func:`create_plan` — allocate next prefix, generate kebab-case stem,
  register inventory record, return ``(plan_id, path)``. Does NOT
  create the file on disk; the agent is expected to ``write_file`` it.
- :func:`pause_plan`, :func:`continue_plan`, :func:`complete_plan`,
  :func:`abort_plan` — lifecycle transitions.
- :func:`list_plans` — return all records (from inventory), grouped or
  not by status as the caller wishes.
- :func:`find_plan_by_id` — resolve a 3-digit prefix OR full stem to a
  ``PlanRecord``. Returns ``None`` for ambiguous prefix.
- :func:`get_plan_status` — convenience wrapper for ``inventory.get(id).status``.
- :func:`parse_plan_filename` — regex helper exposed for callers.
 
Diagnostics
-----------
Domain: OC_CHAT.MAIN
Levels:
    L2 — high-level operation lifecycle
    L3 — semantic details (plan_id, prefix, status transition)
    L4 — deep tracing (filesystem rename attempts, reconcile diffs)
 
Contracts
---------
- All public mutators raise :class:`KeyError` for unknown plan_ids.
- :func:`continue_plan` raises :class:`ValueError` if the plan is not
  currently paused. :func:`complete_plan` / :func:`abort_plan` raise on
  terminal-to-terminal transitions.
- :func:`create_plan` writes to inventory and chooses a filename, but
  does NOT create the file on disk.
- :func:`find_plan_by_id` returns the matching record or ``None``.
  Raises :class:`ValueError` only when the input is ambiguous
  (multiple matches by prefix).
 
Dependencies
------------
- oc_diagnostics (structured debug logging via _dbg)
- oc_orchestrator.plan_inventory (PlanInventory, get_inventory, reconcile)
- oc_orchestrator.wordgen (random_plan_filename)
"""
from __future__ import annotations
 
import re
from pathlib import Path
 
from oc_diagnostics import _dbg
 
from oc_orchestrator.plan_inventory import (
    DEFAULT_PLANS_DIR,
    PlanInventory,
    PlanPaths,
    PlanRecord,
    TERMINAL_STATUSES,
    _FILENAME_RE,
    _expected_filename,
    get_inventory,
    reconcile,
)
from oc_orchestrator.wordgen import random_plan_filename
 
#: Re-exported, not redefined (plan 257 P3). The literal lives beside the
#: inventory filename it is joined to, so the layout assumption is stated
#: once. Existing callers -- ``scripts/plan_manage.py`` and
#: ``core/commands.py`` -- and the test that monkeypatches
#: ``oc_orchestrator.plans.DEFAULT_PLANS_DIR`` are unaffected: the name is
#: still bound in this module and still read at call time.
 
 
def parse_plan_filename(name: str) -> tuple[str, str] | None:
    """Parse a plan filename into ``(stem, state)``.
 
    State is ``"active"`` when no suffix is present; otherwise one of
    ``"paused"``, ``"complete"``, ``"aborted"``. Returns ``None`` for
    filenames that don't match the plan-file pattern.
    """
    m = _FILENAME_RE.match(name)
    if not m:
        return None
    state = m.group("state") or "active"
    return (m.group("stem"), state)
 
 
def _find_plan_file(plans_dir: Path, plan_id: str) -> Path | None:
    """Locate the on-disk file for ``plan_id`` (any suffix). Returns None."""
    for f in plans_dir.glob(f"{plan_id}*.md"):
        parsed = parse_plan_filename(f.name)
        if parsed is not None and parsed[0] == plan_id:
            return f
    return None
 
 
def _rename_to_state(plans_dir: Path, plan_id: str, new_status: str) -> None:
    """Rename the on-disk file to match the new inventory status.
 
    No-op if the file is missing or already at the expected name.
    Raises :class:`OSError` if the rename fails (caller may swallow and
    rely on :func:`reconcile` to heal later).
    """
    func = "_rename_to_state"
    target_name = _expected_filename(plan_id, new_status)
    target = plans_dir / target_name
    src = _find_plan_file(plans_dir, plan_id)
    if src is None:
        _dbg("OC_CHAT.MAIN", func,
             f"no on-disk file for {plan_id!r}; skipping rename", 3)
        return
    if src == target:
        return
    src.rename(target)
    _dbg("OC_CHAT.MAIN", func, f"{src.name} → {target_name}", 3)
 
 
def _resolve_inventory(
    inventory: PlanInventory | None,
    plans_dir: Path,
) -> tuple[PlanInventory, Path]:
    """Internal: pick the active inventory and resolve plans_dir."""
    if inventory is None:
        inventory = get_inventory()
    return inventory, plans_dir
 
 
def create_plan(
    description: str,
    session_name: str | None = None,
    *,
    inventory: PlanInventory | None = None,
    plans_dir: Path | None = None,
) -> tuple[str, Path]:
    """Allocate a new plan and register its inventory record.
 
    Returns ``(plan_id, path)``. The on-disk file is **not** created;
    the caller (typically the agent via ``write_file``) writes the
    Markdown content to ``path``.
 
    Refusal logic (e.g., "another plan is active") is the caller's
    responsibility — this function does not consult any session state.
    """
    func = "create_plan"
    plans_dir = Path(plans_dir) if plans_dir is not None else DEFAULT_PLANS_DIR
    inv, plans_dir = _resolve_inventory(inventory, plans_dir)
    plans_dir.mkdir(parents=True, exist_ok=True)
 
    # Collision-retry: 5 attempts then 4-digit suffix fallback handled
    # internally by random_plan_filename; here we just retry against
    # existing files for the (very rare) double-collision case.
    for attempt in range(5):
        prefix = inv.next_prefix()
        candidate = random_plan_filename(prefix)
        plan_id = candidate[:-3]  # strip ".md"
        if inv.get(plan_id) is None and not (plans_dir / candidate).exists():
            break
        _dbg("OC_CHAT.MAIN", func,
             f"collision on {plan_id!r}; retrying (attempt {attempt + 1})", 3)
    else:
        raise RuntimeError("create_plan: 5 collisions in a row — investigate")
 
    inv.register(plan_id, description=description, session_name=session_name)
    path = plans_dir / candidate
    _dbg("OC_CHAT.MAIN", func, f"created {plan_id!r} → {path}", 2)
    return plan_id, path
 
 
def _transition(
    plan_id: str,
    new_status: str,
    inventory: PlanInventory | None,
    plans_dir: Path | None,
) -> PlanRecord:
    """Shared implementation: inventory update first, then rename."""
    func = "_transition"
    plans_dir = Path(plans_dir) if plans_dir is not None else DEFAULT_PLANS_DIR
    inv, plans_dir = _resolve_inventory(inventory, plans_dir)
    rec = inv.update_status(plan_id, new_status)
    try:
        _rename_to_state(plans_dir, plan_id, new_status)
    except OSError as exc:
        _dbg("OC_CHAT.MAIN", func,
             f"SYSTEM_ERROR: rename failed for {plan_id!r}: {exc}; "
             "inventory is correct, reconcile will heal", 1, override=True)
    return rec
 
 
def pause_plan(
    plan_id: str,
    *,
    inventory: PlanInventory | None = None,
    plans_dir: Path | None = None,
) -> PlanRecord:
    """Pause an active plan. Idempotent on already-paused."""
    inv, plans_dir = _resolve_inventory(inventory, Path(plans_dir or DEFAULT_PLANS_DIR))
    rec = inv.get(plan_id)
    if rec is None:
        raise KeyError(f"unknown plan_id: {plan_id!r}")
    if rec.status == "paused":
        return rec
    if rec.status in TERMINAL_STATUSES:
        raise ValueError(
            f"cannot pause plan in terminal status {rec.status!r}: {plan_id!r}"
        )
    return _transition(plan_id, "paused", inv, plans_dir)
 
 
def continue_plan(
    plan_id: str,
    *,
    inventory: PlanInventory | None = None,
    plans_dir: Path | None = None,
) -> PlanRecord:
    """Resume a paused plan. Refuses if not currently paused."""
    inv, plans_dir = _resolve_inventory(inventory, Path(plans_dir or DEFAULT_PLANS_DIR))
    rec = inv.get(plan_id)
    if rec is None:
        raise KeyError(f"unknown plan_id: {plan_id!r}")
    if rec.status != "paused":
        raise ValueError(
            f"can only continue a paused plan; {plan_id!r} is {rec.status!r}"
        )
    return _transition(plan_id, "active", inv, plans_dir)
 
 
def complete_plan(
    plan_id: str,
    *,
    inventory: PlanInventory | None = None,
    plans_dir: Path | None = None,
) -> PlanRecord:
    """Mark a plan complete. Terminal."""
    inv, plans_dir = _resolve_inventory(inventory, Path(plans_dir or DEFAULT_PLANS_DIR))
    rec = inv.get(plan_id)
    if rec is None:
        raise KeyError(f"unknown plan_id: {plan_id!r}")
    if rec.status in TERMINAL_STATUSES:
        raise ValueError(
            f"plan {plan_id!r} is already in terminal status {rec.status!r}"
        )
    return _transition(plan_id, "complete", inv, plans_dir)
 
 
def abort_plan(
    plan_id: str,
    *,
    inventory: PlanInventory | None = None,
    plans_dir: Path | None = None,
) -> PlanRecord:
    """Mark a plan aborted. Terminal."""
    inv, plans_dir = _resolve_inventory(inventory, Path(plans_dir or DEFAULT_PLANS_DIR))
    rec = inv.get(plan_id)
    if rec is None:
        raise KeyError(f"unknown plan_id: {plan_id!r}")
    if rec.status in TERMINAL_STATUSES:
        raise ValueError(
            f"plan {plan_id!r} is already in terminal status {rec.status!r}"
        )
    return _transition(plan_id, "aborted", inv, plans_dir)
 
 
def list_plans(
    *,
    inventory: PlanInventory | None = None,
) -> list[PlanRecord]:
    """Return all records, sorted by ``created_at`` descending."""
    inv = inventory if inventory is not None else get_inventory()
    return sorted(inv.all(), key=lambda r: r.created_at, reverse=True)
 
 
def find_plan_by_id(
    plan_id_or_prefix: str,
    *,
    inventory: PlanInventory | None = None,
) -> PlanRecord | None:
    """Resolve a 3-digit prefix or full stem to a :class:`PlanRecord`.
 
    - If the input matches a full plan_id exactly, return that record.
    - If the input is a 3-digit prefix and exactly one record matches,
      return that record.
    - Returns ``None`` for no match.
    - Raises :class:`ValueError` for ambiguous prefix (multiple matches).
    """
    inv = inventory if inventory is not None else get_inventory()
    direct = inv.get(plan_id_or_prefix)
    if direct is not None:
        return direct
    if re.fullmatch(r"\d{1,3}", plan_id_or_prefix):
        prefix_num = int(plan_id_or_prefix)
        wanted_prefix = f"{prefix_num:03d}-"
        matches = [r for r in inv.all() if r.plan_id.startswith(wanted_prefix)]
        if not matches:
            return None
        if len(matches) > 1:
            raise ValueError(
                f"ambiguous prefix {plan_id_or_prefix!r}: "
                f"{len(matches)} plans match — provide full stem instead"
            )
        return matches[0]
    return None
 
 
def get_plan_status(
    plan_id: str,
    *,
    inventory: PlanInventory | None = None,
) -> str | None:
    """Return the status of ``plan_id`` from the inventory, or ``None``."""
    inv = inventory if inventory is not None else get_inventory()
    rec = inv.get(plan_id)
    return rec.status if rec is not None else None