plan_inventory.py

Go to the documentation of this file.

#!/usr/bin/env python3
# -*- coding: utf-8 -*-
# oc_orchestrator/plan_inventory.py
 
"""
Plan inventory storage abstraction.
 
Purpose
-------
Single source of truth for plan state and metadata. A small JSON file
at ``oc_chat/plans/.inventory.json`` accessed through the stable
:class:`PlanInventory` interface. Future SQLite/Postgres backends are
sibling classes; nothing in ``plans.py`` or ``commands.py`` changes.
 
Responsibilities
----------------
- :class:`PlanRecord` — frozen dataclass holding plan_id, status,
  description, timestamps, and originating session_name.
- :class:`PlanInventory` (ABC) — register/get/update_status/remove/all/
  list_by_status/next_prefix.
- :class:`JsonPlanInventory` — concrete JSON-file implementation with
  atomic write-temp-then-rename semantics; tolerates missing file by
  auto-creating an empty inventory.
- :func:`get_inventory` — process-wide singleton factory.
- :func:`reconcile` — heal disk/inventory drift (inventory → filesystem
  direction). Filenames get renamed to mirror inventory.status; orphan
  files (on disk but not in inventory) get absorbed.
 
Diagnostics
-----------
Domain: OC_CHAT.MAIN
Levels:
    L2 — inventory lifecycle events (load, save, reconcile actions)
    L3 — per-record semantic details (status changes, prefix allocation)
    L4 — deep tracing (every read/write)
 
Contracts
---------
- All public mutators are atomic at the inventory level (write-to-temp +
  rename). Crash mid-op leaves the on-disk inventory untouched.
- ``PlanRecord`` is frozen; updates produce new records via
  :meth:`PlanRecord.with_status` / :meth:`with_completed_at`.
- ``status`` is one of ``{"active", "paused", "complete", "aborted"}``.
- ``next_prefix`` raises :class:`RuntimeError` when 999 is taken
  (archive-at-999 deferred to a future phase).
- ``reconcile`` is idempotent.
 
Dependencies
------------
- oc_diagnostics (structured debug logging via _dbg)
"""
from __future__ import annotations
 
import json
import os
import re
import tempfile
from abc import ABC, abstractmethod
from dataclasses import asdict, dataclass, field, replace
from datetime import datetime
from pathlib import Path
from typing import Any
 
from oc_diagnostics import _dbg
 
VALID_STATUSES = ("active", "paused", "complete", "aborted")
TERMINAL_STATUSES = ("complete", "aborted")
 
#: Where plan markdown files live, relative to the process cwd. Plan 257
#: P3: this and :data:`DEFAULT_INVENTORY_NAME` are the layout assumption,
#: stated once. It lives in this module rather than in ``plans.py``
#: because ``plans.py`` imports this one, and the inventory path is
#: derived from it -- two literals encoding the same layout is exactly
#: what the A2 residue was.
DEFAULT_PLANS_DIR = Path("oc_chat/plans")
 
#: The inventory file's name inside the plans directory.
DEFAULT_INVENTORY_NAME = ".inventory.json"
 
#: Typed receipts from the owner atomic op (plan 152 R11-05). They tell a
#: caller what is TRUE of disk and memory after a durable mutation attempt,
#: so a failure can be reported precisely instead of guessed at.
NOT_APPLIED = "NOT_APPLIED"   # durable write failed; memory AND disk are OLD
APPLIED = "APPLIED"           # disk committed and memory published; they agree
UNKNOWN = "UNKNOWN"           # durable write landed, publish did not complete
 
# Filename regex: NNN-stem.md or NNN-stem_<state>.md
_FILENAME_RE = re.compile(
    # OCT-LINT: disable=no_hardcoded_secrets reason="plan-filename regex (NNN-stem[_state].md); a parser pattern, no credential content"
    r"^(?P<stem>\d{3}-[a-z0-9-]+?)(?:_(?P<state>paused|complete|aborted))?\.md$"
)
 
 
@dataclass(frozen=True)
class PlanPaths:
    """All caller-controlled filesystem inputs to one plan operation.
 
    Plan 257 P3, in the ``DispatchRequest`` shape the dispatch layer
    already uses: a frozen dataclass whose defaults are today's values,
    constructed at the boundary and passed whole.
 
    ``plans_dir`` is late-bound through a default factory rather than
    captured at class-creation time, so a caller that monkeypatches
    :data:`DEFAULT_PLANS_DIR` is still honoured -- the same reason
    ``PrepareRequest.interpreter`` uses a factory in
    ``worktree_env.py``.
    """
 
    plans_dir: Path = field(default_factory=lambda: DEFAULT_PLANS_DIR)
    #: Overrides the derived ``<plans_dir>/.inventory.json`` when set.
    inventory_path_override: Path | None = None
 
    @property
    def inventory_path(self) -> Path:
        """The inventory file: the override, else derived from *plans_dir*."""
        if self.inventory_path_override is not None:
            return self.inventory_path_override
        return self.plans_dir / DEFAULT_INVENTORY_NAME
 
    def plan_body_relpath(self, plan_id: str) -> Path:
        """Where *plan_id*'s markdown file lives, relative to a repo root.
 
        Plan 270 P3. ``plan_status.py`` and ``plan_manage.py`` each built
        this by hand from a literal ``Path("oc_chat") / "plans"``, which
        is a CONSUMER's layout and cannot be hard-coded in a standalone
        package. Both now ask here, so the layout is stated once beside
        the inventory path it sits next to, and a consumer with a
        different tree overrides it by constructing a :class:`PlanPaths`
        with its own ``plans_dir``.
 
        The value is deliberately RELATIVE: it is recorded in registry
        rows and in manifests as the plan's identity on disk, where an
        absolute path would differ per machine and per worktree.
        """
        return self.plans_dir / f"{plan_id}.md"
 
 
def _now_iso() -> str:
    # OCT-LINT: disable=no_hardcoded_secrets reason="English docstring prose; no credential content"
    """Return a local-with-offset ISO 8601 timestamp.
 
    Form: ``YYYY-MM-DDTHH:MM:SS±HH:MM`` (e.g. ``2026-05-17T00:25:42+02:00``).
    Local wall-clock time + explicit zone offset per the project-wide
    convention in :doc:`CHANGELOG-format.md` §"Timestamps".
    """
    return datetime.now().astimezone().replace(microsecond=0).isoformat()
 
 
@dataclass(frozen=True)
class PlanRecord:
    """A single plan's metadata and current status.
 
    Status values: ``active``, ``paused``, ``complete``, ``aborted``.
    ``plan_id`` is the filename stem without ``.md`` and without state
    suffix — e.g., ``032-sunny-weaving-honey``.
    """
    plan_id: str
    description: str
    status: str = "active"
    created_at: str = field(default_factory=_now_iso)
    completed_at: str | None = None
    aborted_at: str | None = None
    session_name: str | None = None
 
    def to_dict(self) -> dict[str, Any]:
        """Serialise for JSON storage. Omits ``plan_id`` (it's the dict key)."""
        d = asdict(self)
        d.pop("plan_id")
        return d
 
    @classmethod
    def from_dict(cls, plan_id: str, data: dict[str, Any]) -> PlanRecord:
        """Reconstruct from JSON. Missing optional fields get defaults."""
        return cls(
            plan_id=plan_id,
            description=data.get("description", ""),
            status=data.get("status", "active"),
            created_at=data.get("created_at") or _now_iso(),
            completed_at=data.get("completed_at"),
            aborted_at=data.get("aborted_at"),
            session_name=data.get("session_name"),
        )
 
 
class PlanInventory(ABC):
    """Backend-agnostic interface for plan storage."""
 
    @abstractmethod
    def register(
        self,
        plan_id: str,
        description: str,
        session_name: str | None = None,
    ) -> PlanRecord:
        """Insert a new active record. Raises if plan_id already exists."""
 
    @abstractmethod
    def get(self, plan_id: str) -> PlanRecord | None:
        """Return the record for plan_id, or None if missing."""
 
    @abstractmethod
    def update_status(self, plan_id: str, new_status: str) -> PlanRecord:
        """Transition a record's status. Sets completed_at/aborted_at as
        appropriate. Raises if plan_id is unknown or status is invalid."""
 
    @abstractmethod
    def remove(self, plan_id: str) -> None:
        """Delete a record. No-op if plan_id is unknown."""
 
    @abstractmethod
    def all(self) -> list[PlanRecord]:
        """Return all records, unsorted."""
 
    @abstractmethod
    def list_by_status(self, status: str) -> list[PlanRecord]:
        """Return records with the given status."""
 
    @abstractmethod
    def next_prefix(self) -> int:
        """Return the next available 3-digit prefix (1-based). Considers
        BOTH inventory records and on-disk plan files. Raises
        :class:`RuntimeError` if 999 is taken."""
 
 
class JsonPlanInventory(PlanInventory):
    """JSON-file-backed plan inventory.
 
    File format (version 1)::
 
        {
          "version": 1,
          "plans": {
            "<plan_id>": { "description": str, "status": str,
                            "created_at": str, "completed_at": str|None,
                            "aborted_at": str|None,
                            "session_name": str|None },
            ...
          }
        }
 
    Writes are atomic: a fresh temp file is written and then renamed
    over the inventory path. A corrupt or missing file is recovered
    transparently with an empty inventory.
    """
 
    def __init__(self, path: Path, plans_dir: Path | None = None) -> None:
        """Open (or create) the inventory at *path*.
 
        ``plans_dir`` defaults to the parent of *path* and is used to
        check existing on-disk filenames when allocating new prefixes.
        """
        func = "JsonPlanInventory.__init__"
        self._path = Path(path)
        self._plans_dir = Path(plans_dir) if plans_dir is not None else self._path.parent
        self._records: dict[str, PlanRecord] = {}
        self._load_or_init()
        _dbg("OC_CHAT.MAIN", func,
             f"opened {self._path} ({len(self._records)} record(s))", 2)
 
    # ── Internal IO ────────────────────────────────────────────────────
 
    def _load_or_init(self) -> None:
        """Load the inventory file. Missing or corrupt → empty inventory."""
        func = "JsonPlanInventory._load_or_init"
        if not self._path.exists():
            _dbg("OC_CHAT.MAIN", func, f"no inventory at {self._path}; starting empty", 2)
            self._records = {}
            return
        try:
            raw = self._path.read_bytes()
            if raw.startswith(b"\xef\xbb\xbf"):
                raw = raw[3:]
            data = json.loads(raw.decode("utf-8"))
        except (OSError, json.JSONDecodeError) as exc:
            _dbg("OC_CHAT.MAIN", func,
                 # OCT-LINT: disable=no_hardcoded_secrets reason="diagnostic message string emitted via _dbg; no credential content"
                 f"SYSTEM_ERROR: inventory unreadable, starting empty: {exc}",
                 1, override=True)
            self._records = {}
            return
 
        plans = data.get("plans", {})
        self._records = {
            pid: PlanRecord.from_dict(pid, rec)
            for pid, rec in plans.items()
        }
 
    def _write(self) -> None:
        """Atomically persist the current records to disk."""
        func = "JsonPlanInventory._write"
        self._commit_records(dict(self._records))
        _dbg("OC_CHAT.MAIN", func, f"wrote {len(self._records)} record(s)", 4)
 
    def _commit_records(self, candidate: dict) -> None:
        """Durably write *candidate* to disk. Does NOT publish it in memory.
 
        The durable half of the owner atomic op (plan 152 R11-05). Splitting it
        out is what lets callers write BEFORE publishing: previously ``register``
        and ``update_status`` mutated the live ``_records`` map and only then
        called this, so a serialization / temp-write / replace failure left
        memory NEW while disk stayed OLD — and a compensating second write is
        itself fallible, so wrapping the call sites could not fix it.
 
        The temp file is flushed AND fsynced before the replace. Without the
        fsync, ``Path.replace`` can be durable while the temp file's contents
        are still only in the page cache, so a crash at the wrong moment leaves
        the inventory pointing at a truncated or empty file — atomic in name
        only. Temp cleanup runs in a ``finally`` so a failure never strews
        ``.inventory_*.tmp`` files.
        """
        self._path.parent.mkdir(parents=True, exist_ok=True)
        payload = {
            "version": 1,
            "plans": {pid: r.to_dict() for pid, r in candidate.items()},
        }
        tmp_path: Path | None = None
        try:
            with tempfile.NamedTemporaryFile(
                mode="w", encoding="utf-8", dir=self._path.parent,
                prefix=".inventory_", suffix=".tmp", delete=False,
            ) as tmp:
                tmp_path = Path(tmp.name)
                json.dump(payload, tmp, indent=2, ensure_ascii=False)
                tmp.flush()
                os.fsync(tmp.fileno())
            tmp_path.replace(self._path)   # <- the DURABLE COMMIT POINT
            tmp_path = None                # replaced; nothing left to clean up
        finally:
            if tmp_path is not None and tmp_path.exists():
                try:
                    tmp_path.unlink()
                except OSError:
                    pass
 
    def commit_atomic(self, candidate: dict) -> str:
        """Durably commit *candidate*, publishing in memory ONLY after it lands.
 
        The owner atomic op. Returns a typed receipt:
 
        - ``NOT_APPLIED`` — the durable write failed; memory AND disk are OLD, so
          the caller may retry or report cleanly.
        - ``APPLIED`` — disk committed and memory published; they agree.
        - ``UNKNOWN`` — the durable write landed but publishing did not complete.
          Fail closed: the caller must report an indeterminate state rather than
          claim success.
 
        Ordering is the whole point: disk first, memory second. The reverse — the
        pre-existing shape — could leave the in-process view claiming a plan
        exists that no restart would ever see.
        """
        func = "JsonPlanInventory.commit_atomic"
        try:
            self._commit_records(candidate)
        except Exception as exc:
            _dbg("OC_CHAT.MAIN", func,
                 # OCT-LINT: disable=no_hardcoded_secrets reason="diagnostic message string emitted via _dbg; no credential content"
                 f"EXTERNAL_ERROR: durable write failed, nothing published: "
                 f"{type(exc).__name__}: {exc}", 2, override=True)
            return NOT_APPLIED
        try:
            self._records = candidate
        except Exception as exc:                      # pragma: no cover - defensive
            _dbg("OC_CHAT.MAIN", func,
                 f"SYSTEM_ERROR: durable write landed but publish failed: {exc}",
                 1, override=True)
            return UNKNOWN
        return APPLIED
 
    # ── Public API ─────────────────────────────────────────────────────
 
    def register(
        self,
        plan_id: str,
        description: str,
        session_name: str | None = None,
    ) -> PlanRecord:
        func = "JsonPlanInventory.register"
        if plan_id in self._records:
            raise ValueError(f"plan_id already registered: {plan_id!r}")
        rec = PlanRecord(
            plan_id=plan_id,
            description=description,
            status="active",
            session_name=session_name,
        )
        # Build the candidate map WITHOUT publishing, commit it durably, and
        # publish in memory only after the durable commit point (R11-05).
        candidate = dict(self._records)
        candidate[plan_id] = rec
        receipt = self.commit_atomic(candidate)
        if receipt != APPLIED:
            raise OSError(
                f"could not durably register plan {plan_id!r} (receipt={receipt})"
            )
        _dbg("OC_CHAT.MAIN", func, f"registered {plan_id!r}", 3)
        return rec
 
    def get(self, plan_id: str) -> PlanRecord | None:
        return self._records.get(plan_id)
 
    def update_status(self, plan_id: str, new_status: str) -> PlanRecord:
        func = "JsonPlanInventory.update_status"
        if new_status not in VALID_STATUSES:
            raise ValueError(
                f"invalid status {new_status!r}; expected one of {VALID_STATUSES}"
            )
        existing = self._records.get(plan_id)
        if existing is None:
            raise KeyError(f"unknown plan_id: {plan_id!r}")
 
        updates: dict[str, Any] = {"status": new_status}
        if new_status == "complete" and existing.completed_at is None:
            updates["completed_at"] = _now_iso()
        elif new_status == "aborted" and existing.aborted_at is None:
            updates["aborted_at"] = _now_iso()
 
        new_rec = replace(existing, **updates)
        candidate = dict(self._records)
        candidate[plan_id] = new_rec
        receipt = self.commit_atomic(candidate)
        if receipt != APPLIED:
            raise OSError(
                f"could not durably update plan {plan_id!r} (receipt={receipt})"
            )
        _dbg("OC_CHAT.MAIN", func,
             f"{plan_id!r}: {existing.status} → {new_status}", 3)
        return new_rec
 
    def remove(self, plan_id: str) -> None:
        func = "JsonPlanInventory.remove"
        if plan_id in self._records:
            del self._records[plan_id]
            self._write()
            _dbg("OC_CHAT.MAIN", func, f"removed {plan_id!r}", 3)
 
    def all(self) -> list[PlanRecord]:
        return list(self._records.values())
 
    def list_by_status(self, status: str) -> list[PlanRecord]:
        return [r for r in self._records.values() if r.status == status]
 
    def next_prefix(self) -> int:
        """Return the next prefix not used by inventory or on-disk files.
 
        TODO(phase-future): archive-and-reset when 999 is reached. For
        now, raise loudly so we fail rather than silently overflow.
        """
        used: set[int] = set()
        for plan_id in self._records:
            try:
                used.add(int(plan_id.split("-", 1)[0]))
            except ValueError:
                continue
        if self._plans_dir.is_dir():
            for f in self._plans_dir.glob("*.md"):
                m = _FILENAME_RE.match(f.name)
                if m:
                    try:
                        used.add(int(m.group("stem").split("-", 1)[0]))
                    except ValueError:
                        continue
        for n in range(1, 1000):
            if n not in used:
                return n
        raise RuntimeError(
            "Plan directory full (999 prefixes used); manual archive required"
        )
 
 
# ── Process-wide singleton factory ─────────────────────────────────────
 
_default_inventory: PlanInventory | None = None
 
 
def get_inventory(
    path: Path | None = None,
    *,
    paths: PlanPaths | None = None,
) -> PlanInventory:
    """Return the process-wide default inventory.
 
    First call may pass an explicit ``path`` (used by tests). Subsequent
    calls return the cached instance regardless of ``path``. Pass
    ``path`` only on first call; pass ``None`` thereafter.
 
    Plan 257 P3: the default location is now read off :class:`PlanPaths`
    rather than from a second literal, and the unused
    ``persistence.DEFAULT_SESSIONS_DIR`` import that used to sit on the
    line above it is gone. That one dead line pulled 43 modules --
    session, config, defaults, startup and the rest -- into this
    module's import closure, and importing them ran ``load_dotenv`` and
    created a log directory as a side effect of asking for a plan list.
    """
    global _default_inventory
    if _default_inventory is None:
        if path is None:
            path = (paths or PlanPaths()).inventory_path
        _default_inventory = JsonPlanInventory(path)
    return _default_inventory
 
 
def reset_inventory_for_tests() -> None:
    """Test-only helper to clear the cached singleton."""
    global _default_inventory
    _default_inventory = None
 
 
# ── Reconcile: inventory → filesystem ──────────────────────────────────
 
 
def _expected_filename(plan_id: str, status: str) -> str:
    """Compute the expected on-disk filename for a record."""
    if status == "active":
        return f"{plan_id}.md"
    return f"{plan_id}_{status}.md"
 
 
def reconcile(inventory: PlanInventory, plans_dir: Path) -> dict[str, list[str]]:
    """Heal disk/inventory drift. Inventory wins.
 
    Two operations:
 
    1. For every inventory record, ensure exactly one file exists at the
       expected filename. If the file is at a different suffix, rename
       it. If no file exists for the plan_id, log a warning (we do not
       create empty files — the agent may not have written the plan
       yet).
 
    2. For every on-disk file matching the plan-filename regex that
       isn't in the inventory, absorb it: register a new record with
       status derived from the suffix and an empty description.
 
    Returns a dict with three lists for callers/tests to inspect:
    ``renamed``, ``orphans_absorbed``, ``missing_files``.
    """
    func = "reconcile"
    result: dict[str, list[str]] = {
        "renamed": [],
        "orphans_absorbed": [],
        "missing_files": [],
    }
    plans_dir = Path(plans_dir)
    if not plans_dir.is_dir():
        _dbg("OC_CHAT.MAIN", func, f"plans_dir missing: {plans_dir}", 2)
        return result
 
    # Pass 1: inventory → filesystem (rename mismatched files)
    for rec in inventory.all():
        expected = _expected_filename(rec.plan_id, rec.status)
        expected_path = plans_dir / expected
        if expected_path.exists():
            continue
        # Search for any file matching this plan_id with any suffix
        candidates = [
            p for p in plans_dir.glob(f"{rec.plan_id}*.md")
            if _FILENAME_RE.match(p.name)
            and _FILENAME_RE.match(p.name).group("stem") == rec.plan_id
        ]
        if not candidates:
            result["missing_files"].append(rec.plan_id)
            _dbg("OC_CHAT.MAIN", func,
                 f"inventory has {rec.plan_id!r} but no file on disk", 2)
            continue
        src = candidates[0]
        try:
            src.rename(expected_path)
            result["renamed"].append(f"{src.name} → {expected}")
            _dbg("OC_CHAT.MAIN", func, f"renamed {src.name} → {expected}", 3)
        except OSError as exc:
            _dbg("OC_CHAT.MAIN", func,
                 f"SYSTEM_ERROR: rename failed for {src}: {exc}",
                 1, override=True)
 
    # Pass 2: filesystem → inventory (absorb orphans)
    known_ids = {r.plan_id for r in inventory.all()}
    for f in plans_dir.glob("*.md"):
        m = _FILENAME_RE.match(f.name)
        if not m:
            continue
        stem = m.group("stem")
        if stem in known_ids:
            continue
        state = m.group("state") or "active"
        try:
            inventory.register(stem, description="", session_name=None)
            if state != "active":
                inventory.update_status(stem, state)
            result["orphans_absorbed"].append(f.name)
            _dbg("OC_CHAT.MAIN", func,
                 f"absorbed orphan {f.name} (status={state})", 3)
        except ValueError:
            # Race: another process registered between checks. Skip.
            continue
 
    return result