planner.py

Go to the documentation of this file.

#!/usr/bin/env python3
# -*- coding: utf-8 -*-
# oc_chat/tools/planner.py
 
"""
Multi-step task planner for agentic goal execution.
 
Purpose
-------
Process-singleton plan state that the model uses to break a complex
goal into discrete steps and track progress across ReAct iterations.
The model calls :func:`create_plan` to set up a plan, :func:`mark_step`
as each step completes, :func:`get_status` for progress, and
:func:`stop_plan` to halt execution.
 
Responsibilities
----------------
- :class:`Step` / :class:`Plan` — dataclasses describing a single
  step (description, status, result) and the active plan (goal,
  steps, completion check).
- :func:`create_plan` — start a new plan; replaces any existing one.
- :func:`mark_step` — transition a step to ``done`` / ``failed`` with
  an optional result string.
- :func:`get_status` — return the current plan as a dict.
- :func:`stop_plan` — clear active plan state and set the stop flag.
 
Diagnostics
-----------
Domain: OC_CHAT.TOOLS
Levels:
    L2 — lifecycle events
    L3 — semantic details
    L4 — deep tracing
 
Contracts
---------
- Module-level ``_active_plan`` and ``_stopped`` are process-singleton
  state — only one plan is active at a time per REPL.
- ``create_plan`` overwrites any existing plan without warning.
- ``mark_step`` validates the step index; out-of-range indices return
  an error dict rather than raising.
- ``stop_plan`` is idempotent — calling it twice has no extra effect.
 
Dependencies
------------
- oc_diagnostics (structured debug logging via _dbg)
 
See Also
--------
oc_chat/docs/conventions/CONTRACTS.md §8 Plan / Step — owner of the planner
dataclasses; §3 Tool registry contract; §4 Tool return-value conventions.
"""
from dataclasses import dataclass, field
from typing import Literal
 
from oc_diagnostics import _dbg
 
_active_plan = None
_stopped = False
 
 
@dataclass
class Step:
    """A single step within a :class:`Plan`."""
    description: str
    status: Literal["pending", "running", "done", "failed"] = "pending"
    result: str = ""
 
 
@dataclass
class Plan:
    """A multi-step plan with a goal and ordered steps."""
    goal: str
    steps: list = field(default_factory=list)
 
    @property
    def is_done(self) -> bool:
        """Return ``True`` when every step has completed (done or failed)."""
        return bool(self.steps) and all(s.status in ("done", "failed") for s in self.steps)
 
    @property
    def next_pending_index(self) -> int | None:
        """Return the index of the first pending step, or ``None``."""
        for i, s in enumerate(self.steps):
            if s.status == "pending":
                return i
        return None
 
 
# ── Tool functions ─────────────────────────────────────────────────────────────
 
#: Keys a model commonly nests a step description under when it (wrongly) sends
#: ``steps`` as a list of objects instead of a list of strings (F28).
_STEP_TEXT_KEYS = ("description", "step", "title", "text", "name", "task")
 
 
def _coerce_steps(steps: object) -> "tuple[list[str] | None, str | None]":
    """Coerce a model-supplied *steps* value into a ``list[str]``, or return an error.
 
    F28: ``create_plan`` used to ``len(steps)`` / iterate ``steps`` blindly, so a
    ``None`` raised ``TypeError`` (invisible crash) and a bare STRING was silently
    char-split into one step per character. This tolerates the shapes the model
    actually sends — a list of dicts (``{'description'|'step'|'title'|...: ...}``),
    non-string scalars (stringified) — and hard-rejects only the crash/corruption
    shapes (``None`` and a bare string). An empty list is allowed (an empty plan is
    a valid, supported state).
 
    Returns ``(steps_list, None)`` on success or ``(None, error_message)``.
    """
    if steps is None:
        return None, "steps is required: pass a non-empty list of step description strings."
    if isinstance(steps, str):
        return None, (
            "steps must be a LIST of step strings, not a single string "
            "(a string would be split into one step per character). "
            'Pass e.g. ["first step", "second step"].'
        )
    try:
        items = list(steps)
    except TypeError:
        return None, f"steps must be a list of step strings; got {type(steps).__name__}."
 
    out: list[str] = []
    for item in items:
        if isinstance(item, str):
            text = item.strip()
        elif isinstance(item, dict):
            text = ""
            for key in _STEP_TEXT_KEYS:
                val = item.get(key)
                if isinstance(val, str) and val.strip():
                    text = val.strip()
                    break
            if not text:
                text = str(item).strip()
        else:
            text = str(item).strip()
        if text:
            out.append(text)
 
    # An empty list yields an empty plan — a valid, supported state (a plan is
    # simply "not done"); only the crash/corruption shapes (None, a bare string)
    # are hard-rejected above.
    return out, None
 
 
def create_plan(goal: str, steps: list) -> dict:
    """Create a new plan with a list of step descriptions.
 
    Validates and coerces the model-supplied arguments (F28) so a malformed call
    returns a structured ``{"error": ...}`` the model can act on — it never raises
    (an invisible crash) or silently char-splits a bare string.
    """
    func = "create_plan"
    global _active_plan, _stopped
 
    goal_text = goal.strip() if isinstance(goal, str) else str(goal or "").strip()
    if not goal_text:
        _dbg("OC_CHAT.TOOLS", func, "VALIDATION_ERROR: empty goal", 3)
        return {"error": "goal is required: pass a short non-empty goal string."}
 
    coerced, err = _coerce_steps(steps)
    if err is not None:
        _dbg("OC_CHAT.TOOLS", func, f"VALIDATION_ERROR: {err}", 3)
        return {"error": err}
 
    _dbg("OC_CHAT.TOOLS", func, f"goal={goal_text!r} steps={len(coerced)}", 2)
    _active_plan = Plan(goal=goal_text, steps=[Step(description=s) for s in coerced])
    _stopped = False
    return {
        "created": True,
        "goal": goal_text,
        "step_count": len(coerced),
        "steps": [s.description for s in _active_plan.steps],
    }
 
 
def get_status() -> dict:
    """Return the current plan status including all step statuses."""
    if not _active_plan:
        return {"status": "no active plan"}
    return {
        "goal":    _active_plan.goal,
        "done":    _active_plan.is_done,
        "stopped": _stopped,
        "steps": [
            {"index": i, "description": s.description, "status": s.status, "result": s.result}
            for i, s in enumerate(_active_plan.steps)
        ],
    }
 
 
def mark_step(step_index: int, status: str, result: str = "") -> dict:
    """Update the status of a plan step.
 
    *status* must be one of ``pending``, ``running``, ``done``, ``failed``.
    """
    if not _active_plan:
        return {"error": "No active plan"}
    if step_index < 0 or step_index >= len(_active_plan.steps):
        return {"error": f"Invalid step index: {step_index}"}
    if status not in ("pending", "running", "done", "failed"):
        return {"error": f"Invalid status: {status}"}
 
    _active_plan.steps[step_index].status = status
    _active_plan.steps[step_index].result = result
    return {"ok": True, "step": step_index, "status": status}
 
 
def stop_plan() -> dict:
    """Halt the active plan.
 
    Call this when the goal is complete, impossible, or the user has
    requested a stop.
    """
    global _stopped
    _stopped = True
    return {"stopped": True}
 
 
TOOLS = {
    "create_plan": create_plan,
    "get_status":  get_status,
    "mark_step":   mark_step,
    "stop_plan":   stop_plan,
}
 
SCHEMAS = [
    {
        "type": "function",
        "function": {
            "name": "create_plan",
            "description": "Create a multi-step plan for a complex goal.",
            "parameters": {
                "type": "object",
                "properties": {
                    "goal":  {"type": "string", "description": "The overall goal"},
                    "steps": {
                        "type": "array",
                        "items": {"type": "string"},
                        "description": "Ordered list of step descriptions",
                    },
                },
                "required": ["goal", "steps"],
            },
        },
    },
    {
        "type": "function",
        "function": {
            "name": "get_status",
            "description": "Return the current plan status including all step statuses.",
            "parameters": {"type": "object", "properties": {}, "required": []},
        },
    },
    {
        "type": "function",
        "function": {
            "name": "mark_step",
            "description": "Update the status of a plan step.",
            "parameters": {
                "type": "object",
                "properties": {
                    "step_index": {"type": "integer", "description": "Zero-based step index"},
                    "status":     {"type": "string",  "description": "One of: pending, running, done, failed"},
                    "result":     {"type": "string",  "description": "Optional result or error message"},
                },
                "required": ["step_index", "status"],
            },
        },
    },
    {
        "type": "function",
        "function": {
            "name": "stop_plan",
            "description": "Halt the active plan. Call this when the goal is complete, impossible, or the user has requested a stop.",
            "parameters": {"type": "object", "properties": {}, "required": []},
        },
    },
]