Plan Memory

Plan Memory gives agents persistent task decompositions as DAGs — solving two critical failure modes: plan amnesia (re-deriving decompositions after every step) and retry loops (repeating the same failing approach indefinitely).

Architecture

Plans use a two-node architecture:

Plan container (plan type) — stores plan-level metadata: title, status, task count, completion percentage
Plan task nodes (plan_task type) — individual tasks forming a DAG via DEPENDS_ON edges

┌──────────────────────────────┐
│  Plan Container              │
│  plan_id: "plan_abc123"      │
│  status: "in_progress"       │
│  completed_tasks: 1/4        │
└──────────┬───────────────────┘
           │ plan_id = "plan_abc123"
    ┌──────┼──────────┬──────────┐
    ▼      ▼          ▼          ▼
  [t1]   [t2]       [t3]       [t4]
  done   in_prog    pending    pending
           │          │          │
           └──DEPENDS_ON──┘     │
                      └──DEPENDS_ON──┘

Usage

Creating a Plan

from smartmemory.plans import PlanManager

manager = PlanManager(memory)
result = manager.create(
    "Implement auth system",
    tasks=[
        {"content": "Design JWT schema"},
        {"content": "Implement token service", "depends_on": []},
        {"content": "Add refresh flow"},
        {"content": "Write integration tests"},
    ],
    context="User requested JWT-based authentication",
)
plan_id = result["plan_id"]       # "plan_abc123def4"
task_ids = result["task_ids"]     # [uuid1, uuid2, uuid3, uuid4]

Tracking Progress

# Complete a task — automatically recomputes plan completion %
manager.update_task(plan_id, task_ids[0], "complete")

# What can be worked on next? (pending tasks with all deps complete)
next_tasks = manager.get_next_tasks(plan_id)

# What's blocked? (tasks with incomplete dependencies)
blocked = manager.get_blocked_tasks(plan_id)

Plan Lifecycle

# Get full plan state
plan = manager.get(plan_id)
# {"plan": {..., "completed_tasks": 2, "total_tasks": 4}, "tasks": [...]}

# List active plans
active = manager.get_active()

# Complete — optionally graduate to a decision record
decision_id = manager.complete(plan_id, graduate=True)

# Or mark as failed
manager.fail(plan_id, reason="Dependencies unavailable")

Graduation

When a plan completes, it can optionally graduate to a Decision record — preserving the decomposition as a first-class decision with a DERIVED_FROM edge back to the plan. This creates a permanent record of "we decided to do X, and here's the plan that proved it works."

Task Dependencies

Tasks form a DAG (directed acyclic graph) via DEPENDS_ON edges. The dependency resolution logic:

Next tasks: pending tasks where ALL dependencies are complete
Blocked tasks: non-complete tasks where ANY dependency is incomplete
Tasks with no dependencies are always eligible as next tasks

Plan Status Values

Status	Meaning
`active`	Plan is in progress
`completed`	All work done (may have graduated to decision)
`failed`	Plan abandoned with reason

Integration with Structured Ingestion

Plan containers and tasks are both stored via the structured ingestion system:

# These are equivalent — PlanManager calls them internally
memory.ingest_structured({"content": "My plan", "plan_id": "plan_123"}, schema="plan")
memory.ingest_structured({"content": "Task 1", "plan_id": "plan_123"}, schema="plan_task")

See Ingestion Flow for details on structured ingestion strategies.

Plan Memory

On this page