Chapter 16. Structured Plans and Verified Completion

# src/harness/plans/model.py
from __future__ import annotations

from dataclasses import dataclass, field
from datetime import datetime, timezone
from enum import Enum
from typing import Literal
from uuid import uuid4


class StepStatus(str, Enum):
    pending = "pending"
    in_progress = "in_progress"
    done = "done"
    blocked = "blocked"


@dataclass
class Step:
    id: str
    description: str
    status: StepStatus = StepStatus.pending
    evidence: str | None = None   # what proved the step done
    notes: str = ""

    def is_terminal(self) -> bool:
        return self.status in (StepStatus.done, StepStatus.blocked)


@dataclass
class Postcondition:
    description: str
    satisfied: bool = False
    evidence: str | None = None


@dataclass
class Plan:
    objective: str
    steps: list[Step] = field(default_factory=list)
    postconditions: list[Postcondition] = field(default_factory=list)
    created_at: datetime = field(default_factory=lambda: datetime.now(timezone.utc))
    id: str = field(default_factory=lambda: str(uuid4()))

    def all_steps_terminal(self) -> bool:
        return all(s.is_terminal() for s in self.steps)

    def all_postconditions_satisfied(self) -> bool:
        return all(pc.satisfied for pc in self.postconditions)

    def is_ready_to_finalize(self) -> bool:
        return self.all_steps_terminal() and self.all_postconditions_satisfied()

    def to_render(self) -> str:
        """Render the plan as a string the model can read."""
        lines = [f"# Plan: {self.objective}\n"]
        lines.append("## Steps")
        for i, s in enumerate(self.steps, start=1):
            mark = {"pending": "[ ]", "in_progress": "[.]",
                    "done": "[x]", "blocked": "[!]"}[s.status.value]
            lines.append(f"{i}. {mark} {s.description}")
            if s.evidence:
                lines.append(f"     evidence: {s.evidence}")
            if s.notes:
                lines.append(f"     notes: {s.notes}")
        lines.append("\n## Postconditions")
        for i, pc in enumerate(self.postconditions, start=1):
            mark = "[x]" if pc.satisfied else "[ ]"
            lines.append(f"{i}. {mark} {pc.description}")
            if pc.evidence:
                lines.append(f"     evidence: {pc.evidence}")
        return "\n".join(lines)

# src/harness/plans/tools.py
from __future__ import annotations

from dataclasses import dataclass

from ..tools.base import Tool
from ..tools.decorator import tool
from .model import Plan, Postcondition, Step, StepStatus


@dataclass
class PlanHolder:
    """Wraps a mutable Plan so tools can mutate it through a shared reference."""
    plan: Plan | None = None

    def require(self) -> Plan:
        if self.plan is None:
            raise RuntimeError("no active plan")
        return self.plan


def plan_tools(holder: PlanHolder) -> list[Tool]:

    @tool(side_effects={"write"})
    def plan_create(objective: str, steps: list[str],
                     postconditions: list[str]) -> str:
        """Create or replace the plan for this session.

        objective: one-sentence description of what you're trying to
                   accomplish.
        steps: ordered list of step descriptions. Each is a specific
               actionable item, not a vague intent.
        postconditions: list of conditions that must be true when you
                        declare the task complete. Examples: "file X
                        exists and contains Y"; "tests in module Z pass".

        Call this once at the start of any non-trivial task, before
        beginning work. If the plan is wrong mid-task, call this again
        to replace it — the harness records the rewrite.
        """
        holder.plan = Plan(
            objective=objective,
            steps=[Step(id=f"s{i}", description=d) for i, d in enumerate(steps)],
            postconditions=[Postcondition(description=d) for d in postconditions],
        )
        return f"plan created with {len(steps)} steps and {len(postconditions)} postconditions"

    @tool(side_effects={"read"})
    def plan_show() -> str:
        """Display the current plan with its step and postcondition status.

        Use this any time you want to re-orient — especially after long
        sub-tasks or compaction. The plan is durable; compaction won't
        lose it.
        """
        return holder.require().to_render()

    @tool(side_effects={"write"})
    def step_update(step_number: int, status: str, evidence: str = "",
                     notes: str = "") -> str:
        """Update a step's status.

        step_number: 1-based index from `plan_show`.
        status: one of 'pending', 'in_progress', 'done', 'blocked'.
        evidence: required for 'done'. One-sentence proof of completion
                  (reference to a tool result, a scratchpad key, etc.).
        notes: optional free text.
        """
        plan = holder.require()
        if step_number < 1 or step_number > len(plan.steps):
            return f"step_number {step_number} out of range (1..{len(plan.steps)})"
        try:
            new_status = StepStatus(status)
        except ValueError:
            return f"invalid status {status!r}; use pending/in_progress/done/blocked"
        if new_status == StepStatus.done and not evidence:
            return ("error: marking a step 'done' requires evidence. Describe "
                    "what proved the step complete (e.g., 'wrote file and "
                    "read it back; content matches').")
        s = plan.steps[step_number - 1]
        plan.steps[step_number - 1] = Step(
            id=s.id, description=s.description,
            status=new_status, evidence=evidence or None,
            notes=notes,
        )
        return f"step {step_number} → {status}"

    @tool(side_effects={"write"})
    def postcondition_verify(postcondition_number: int, evidence: str) -> str:
        """Mark a postcondition as verified.

        postcondition_number: 1-based index.
        evidence: required. Concrete proof the postcondition holds.

        This is what the harness checks before letting you declare the
        task complete. Do not verify a postcondition without evidence.
        """
        plan = holder.require()
        if postcondition_number < 1 or postcondition_number > len(plan.postconditions):
            return (f"postcondition_number {postcondition_number} out of range "
                    f"(1..{len(plan.postconditions)})")
        if not evidence:
            return "error: evidence is required to verify a postcondition"
        pc = plan.postconditions[postcondition_number - 1]
        plan.postconditions[postcondition_number - 1] = Postcondition(
            description=pc.description, satisfied=True, evidence=evidence,
        )
        return f"postcondition {postcondition_number} verified"

    return [plan_create, plan_show, step_update, postcondition_verify]

# src/harness/agent.py (plan-aware version, sketch)

async def arun(
    provider: Provider,
    catalog: ToolCatalog,
    user_message: str,
    system: str | None = None,
    # ...
    plan_holder: "PlanHolder | None" = None,
) -> str:
    # ... existing setup

    for _ in range(MAX_ITERATIONS):
        # ... existing tool selection, compaction, turn execution

        if response.is_final:
            final_text = response.text or ""
            if plan_holder and plan_holder.plan:
                plan = plan_holder.plan
                if not plan.is_ready_to_finalize():
                    # reject the finalization; ask the agent to continue
                    synthetic = (
                        "The plan is not complete. Before declaring the "
                        "task done, either mark remaining steps as done "
                        "with evidence, verify outstanding postconditions, "
                        "or mark them blocked with a reason. Current plan:\n\n"
                        + plan.to_render()
                    )
                    transcript.append(Message.from_assistant_response(response))
                    transcript.append(Message.user_text(synthetic))
                    continue
            transcript.append(Message.from_assistant_response(response))
            return final_text

        # ... existing tool dispatch

# examples/ch16_plan.py
import asyncio

from harness.agent import arun
from harness.plans.tools import PlanHolder, plan_tools
from harness.providers.anthropic import AnthropicProvider
from harness.tools.selector import ToolCatalog
from harness.tools.std import STANDARD_TOOLS


SYSTEM = """\
You are an agent that works from structured plans. For any non-trivial
task, your first action is to call plan_create with an explicit objective,
steps, and postconditions. After each major action, call step_update to
record progress with evidence. Before declaring the task complete, verify
every postcondition with postcondition_verify.

The harness will reject your final answer if the plan is not complete.
Do not claim completion prematurely. If a step cannot be completed, mark
it 'blocked' with a reason rather than skipping it.
"""


async def main() -> None:
    provider = AnthropicProvider()
    holder = PlanHolder()
    catalog = ToolCatalog(tools=STANDARD_TOOLS + plan_tools(holder))

    answer = await arun(
        provider=provider,
        catalog=catalog,
        system=SYSTEM,
        plan_holder=holder,
        user_message=(
            "For each of these three files, verify they exist, measure "
            "their size in bytes, and report the largest one: "
            "/etc/hostname, /etc/os-release, /etc/machine-id. "
            "Produce a summary with file paths, sizes, and the winner."
        ),
    )
    print("---")
    print(answer)
    print("---")
    print(holder.plan.to_render() if holder.plan else "(no plan)")


asyncio.run(main())

git add -A && git commit -m "ch16: structured plans with enforced completion"
git tag ch16-plans