openclaw_store.py

#!/usr/bin/env python3
"""
openclaw_store.py — unified planning + memory backend for OpenClaw agents.

Built to be driven by a small local model over many turns
of a long task, so every command:

  - prints exactly one JSON object to stdout, success or failure — never a
    Python traceback
  - uses short ids like g1 / t4 / m9 instead of uuids, so the agent can
    copy them around reliably
  - tells the agent what to do next via "hint" / "note" fields in the
    response

Quick start:
    python3 openclaw_store.py brief
    python3 openclaw_store.py help

Storage: a single SQLite file, openclaw_store.db, next to this script.
"""

import argparse
import json
import os
import sqlite3
import sys
from datetime import datetime, timezone
from pathlib import Path

DEFAULT_DB_PATH = Path(__file__).resolve().parent / "openclaw_store.db"

PREFIX = {"goal": "g", "task": "t", "mem": "m"}
TYPE_NAME = {v: k for k, v in PREFIX.items()}

TASK_STATUSES = ["pending", "active", "blocked", "completed", "failed", "cancelled"]
GOAL_STATUSES = ["active", "paused", "completed", "abandoned"]

LIST_TRUNCATE = 150   # chars, for description/value fields shown in list views
MAX_LIMIT = 200        # hard ceiling on any --limit, so a bad value can't flood context


# --------------------------------------------------------------------------
# small helpers
# --------------------------------------------------------------------------

def now_iso():
    return datetime.now(timezone.utc).isoformat(timespec="seconds")


def truncate(text, limit=LIST_TRUNCATE):
    if not text:
        return text
    text = str(text)
    return text if len(text) <= limit else text[:limit].rstrip() + "…"


def make_id(kind, rowid):
    if rowid is None:
        return None
    return f"{PREFIX[kind]}{rowid}"


def parse_id(kind, public_id):
    """'g14' -> 14, after checking it's really a goal id. The message on
    failure is written to be read by the agent, not just a human, since
    that's who actually sees it."""
    if public_id is None or not str(public_id).strip():
        raise ValueError(f"missing {kind} id")
    s = str(public_id).strip().lower()
    prefix = PREFIX[kind]
    if not s.startswith(prefix):
        other = TYPE_NAME.get(s[:1]) if s else None
        extra = f" (looks like a {other} id instead)" if other else ""
        raise ValueError(
            f"'{public_id}' is not a valid {kind} id{extra} — {kind} ids look "
            f"like '{prefix}1', '{prefix}2', etc."
        )
    digits = s[len(prefix):]
    if not digits.isdigit():
        raise ValueError(
            f"'{public_id}' is not a valid {kind} id — expected '{prefix}' "
            f"followed by digits, e.g. '{prefix}1'."
        )
    return int(digits)


def parse_id_list(kind, raw):
    """'t3,t7' -> [3, 7]. None/empty -> []."""
    if not raw:
        return []
    return [parse_id(kind, part.strip()) for part in str(raw).split(",") if part.strip()]


def clamp_limit(limit):
    try:
        limit = int(limit)
    except (TypeError, ValueError):
        limit = 50
    return max(1, min(limit, MAX_LIMIT))


def _require_text(value, name):
    if value is None or not str(value).strip():
        raise ValueError(f"{name} cannot be empty")
    return str(value).strip()


def _validate_range(value, lo, hi, name):
    if value is None:
        return
    if not (lo <= value <= hi):
        raise ValueError(f"{name} must be between {lo} and {hi} (got {value})")


class JSONArgumentParser(argparse.ArgumentParser):
    """Same as argparse.ArgumentParser, but a bad command line comes back as
    the same JSON-on-stdout shape as every other response, rather than a
    plain-text usage line on stderr that an agent harness might never show
    the model."""

    def error(self, message):
        print(json.dumps({
            "status": "error",
            "error": "bad_arguments",
            "message": message,
            "hint": "Run 'python3 openclaw_store.py help' for the full command reference.",
        }, indent=2))
        sys.exit(2)


# --------------------------------------------------------------------------
# store
# --------------------------------------------------------------------------

class OpenClawStore:
    def __init__(self, db_path):
        self.db_path = Path(db_path)
        self.db_path.parent.mkdir(parents=True, exist_ok=True)
        self.db = sqlite3.connect(str(self.db_path))
        self.db.row_factory = sqlite3.Row
        self.db.execute("PRAGMA journal_mode=WAL")
        self._create_tables()

    def close(self):
        self.db.close()

    def _create_tables(self):
        self.db.execute("""
        CREATE TABLE IF NOT EXISTS goals (
            id INTEGER PRIMARY KEY AUTOINCREMENT,
            title TEXT NOT NULL,
            description TEXT DEFAULT '',
            status TEXT NOT NULL DEFAULT 'active',
            priority INTEGER NOT NULL DEFAULT 5,
            created_at TEXT NOT NULL,
            updated_at TEXT NOT NULL
        )
        """)
        self.db.execute("""
        CREATE TABLE IF NOT EXISTS tasks (
            id INTEGER PRIMARY KEY AUTOINCREMENT,
            goal_id INTEGER,
            title TEXT NOT NULL,
            description TEXT DEFAULT '',
            status TEXT NOT NULL DEFAULT 'pending',
            priority INTEGER NOT NULL DEFAULT 5,
            depends_on TEXT DEFAULT '[]',
            notes TEXT DEFAULT '',
            created_at TEXT NOT NULL,
            updated_at TEXT NOT NULL,
            completed_at TEXT,
            FOREIGN KEY(goal_id) REFERENCES goals(id)
        )
        """)
        self.db.execute("""
        CREATE TABLE IF NOT EXISTS memories (
            id INTEGER PRIMARY KEY AUTOINCREMENT,
            category TEXT NOT NULL DEFAULT 'general',
            key TEXT NOT NULL,
            value TEXT NOT NULL,
            tags TEXT DEFAULT '[]',
            source TEXT DEFAULT 'manual',
            importance INTEGER NOT NULL DEFAULT 5,
            confidence REAL NOT NULL DEFAULT 1.0,
            related_id TEXT,
            access_count INTEGER NOT NULL DEFAULT 0,
            created_at TEXT NOT NULL,
            updated_at TEXT NOT NULL,
            last_accessed TEXT
        )
        """)
        for stmt in (
            "CREATE INDEX IF NOT EXISTS idx_goal_status ON goals(status)",
            "CREATE INDEX IF NOT EXISTS idx_task_status ON tasks(status)",
            "CREATE INDEX IF NOT EXISTS idx_task_goal ON tasks(goal_id)",
            "CREATE INDEX IF NOT EXISTS idx_task_priority ON tasks(priority)",
            "CREATE UNIQUE INDEX IF NOT EXISTS idx_mem_category_key ON memories(category, key)",
            "CREATE INDEX IF NOT EXISTS idx_mem_importance ON memories(importance)",
        ):
            self.db.execute(stmt)
        self.db.commit()

    # ---------------- GOALS ----------------

    def goal_add(self, title, description="", priority=5):
        title = _require_text(title, "title")
        _validate_range(priority, 1, 10, "priority")
        now = now_iso()
        cur = self.db.execute(
            "INSERT INTO goals (title, description, status, priority, created_at, updated_at) "
            "VALUES (?, ?, 'active', ?, ?, ?)",
            (title, description, priority, now, now),
        )
        self.db.commit()
        gid = make_id("goal", cur.lastrowid)
        return {
            "status": "success",
            "id": gid,
            "hint": f'Add tasks under it with: task add "..." --goal {gid}',
        }

    def goal_list(self, status=None, limit=50):
        limit = clamp_limit(limit)
        query = "SELECT * FROM goals"
        params = []
        if status and status != "all":
            if status not in GOAL_STATUSES:
                raise ValueError(
                    f"'{status}' is not a valid goal status — use one of: {', '.join(GOAL_STATUSES)} "
                    f"(or 'all' for everything)."
                )
            query += " WHERE status = ?"
            params.append(status)
        query += " ORDER BY priority DESC, created_at ASC"
        rows = self.db.execute(query, params).fetchall()
        total = len(rows)
        rows = rows[:limit]
        goals = []
        for r in rows:
            counts = self.db.execute(
                "SELECT COUNT(*) AS total, SUM(status='completed') AS done FROM tasks WHERE goal_id=?",
                (r["id"],),
            ).fetchone()
            goals.append({
                "id": make_id("goal", r["id"]),
                "title": r["title"],
                "description": truncate(r["description"]),
                "status": r["status"],
                "priority": r["priority"],
                "tasks_done": counts["done"] or 0,
                "tasks_total": counts["total"] or 0,
            })
        result = {"status": "success", "count": len(goals), "goals": goals}
        if total > len(goals):
            result["truncated"] = True
            result["total_matching"] = total
        return result

    def goal_get(self, public_id):
        gid = parse_id("goal", public_id)
        row = self.db.execute("SELECT * FROM goals WHERE id=?", (gid,)).fetchone()
        if not row:
            raise ValueError(f"no goal with id '{public_id}'. Run 'goal list' to see valid ids.")
        tasks = self.db.execute(
            "SELECT id, title, status, priority FROM tasks WHERE goal_id=? "
            "ORDER BY priority DESC, created_at ASC",
            (gid,),
        ).fetchall()
        return {
            "status": "success",
            "goal": {
                "id": make_id("goal", row["id"]),
                "title": row["title"],
                "description": row["description"],
                "status": row["status"],
                "priority": row["priority"],
                "created_at": row["created_at"],
                "updated_at": row["updated_at"],
                "tasks": [
                    {
                        "id": make_id("task", t["id"]),
                        "title": t["title"],
                        "status": t["status"],
                        "priority": t["priority"],
                    }
                    for t in tasks
                ],
            },
        }

    def goal_status(self, public_id, new_status):
        if new_status not in GOAL_STATUSES:
            raise ValueError(
                f"'{new_status}' is not a valid goal status — use one of: {', '.join(GOAL_STATUSES)}"
            )
        gid = parse_id("goal", public_id)
        if not self.db.execute("SELECT 1 FROM goals WHERE id=?", (gid,)).fetchone():
            raise ValueError(f"no goal with id '{public_id}'. Run 'goal list' to see valid ids.")
        self.db.execute(
            "UPDATE goals SET status=?, updated_at=? WHERE id=?", (new_status, now_iso(), gid)
        )
        self.db.commit()
        return {"status": "success", "id": public_id, "new_status": new_status}

    def goal_rm(self, public_id):
        gid = parse_id("goal", public_id)
        if not self.db.execute("SELECT 1 FROM goals WHERE id=?", (gid,)).fetchone():
            raise ValueError(f"no goal with id '{public_id}'. Run 'goal list' to see valid ids.")
        self.db.execute("UPDATE tasks SET goal_id=NULL WHERE goal_id=?", (gid,))
        self.db.execute("DELETE FROM goals WHERE id=?", (gid,))
        self.db.commit()
        return {
            "status": "success",
            "id": public_id,
            "note": "Deleted. Any tasks under it are kept but now unassigned.",
        }

    # ---------------- TASKS ----------------

    def task_add(self, title, goal_id=None, description="", priority=5, depends_on=None):
        title = _require_text(title, "title")
        _validate_range(priority, 1, 10, "priority")
        gid = None
        if goal_id:
            gid = parse_id("goal", goal_id)
            if not self.db.execute("SELECT 1 FROM goals WHERE id=?", (gid,)).fetchone():
                raise ValueError(
                    f"no goal with id '{goal_id}'. Run 'goal list' to see valid ids, "
                    f"or omit --goal for a standalone task."
                )
        dep_ids = parse_id_list("task", depends_on)
        for d in dep_ids:
            if not self.db.execute("SELECT 1 FROM tasks WHERE id=?", (d,)).fetchone():
                raise ValueError(f"depends-on task '{make_id('task', d)}' does not exist.")
        now = now_iso()
        cur = self.db.execute(
            "INSERT INTO tasks (goal_id, title, description, status, priority, depends_on, "
            "notes, created_at, updated_at, completed_at) "
            "VALUES (?, ?, ?, 'pending', ?, ?, '', ?, ?, NULL)",
            (gid, title, description, priority, json.dumps(dep_ids), now, now),
        )
        self.db.commit()
        tid = make_id("task", cur.lastrowid)
        return {
            "status": "success",
            "id": tid,
            "hint": "Run 'task next' to see if this is ready to work on.",
        }

    def task_list(self, goal_id=None, status=None, limit=50):
        limit = clamp_limit(limit)
        query = "SELECT * FROM tasks WHERE 1=1"
        params = []
        if goal_id:
            gid = parse_id("goal", goal_id)
            query += " AND goal_id = ?"
            params.append(gid)
        if status and status != "all":
            if status not in TASK_STATUSES:
                raise ValueError(
                    f"'{status}' is not a valid task status — use one of: {', '.join(TASK_STATUSES)} "
                    f"(or 'all' for everything)."
                )
            query += " AND status = ?"
            params.append(status)
        query += " ORDER BY priority DESC, created_at ASC"
        rows = self.db.execute(query, params).fetchall()
        total = len(rows)
        rows = rows[:limit]
        tasks = [self._format_task(dict(r), full=False) for r in rows]
        result = {"status": "success", "count": len(tasks), "tasks": tasks}
        if total > len(tasks):
            result["truncated"] = True
            result["total_matching"] = total
        return result

    def task_get(self, public_id):
        tid = parse_id("task", public_id)
        row = self.db.execute("SELECT * FROM tasks WHERE id=?", (tid,)).fetchone()
        if not row:
            raise ValueError(f"no task with id '{public_id}'. Run 'task list' to see valid ids.")
        return {"status": "success", "task": self._format_task(dict(row), full=True)}

    def next_task(self):
        active_row = self.db.execute(
            "SELECT * FROM tasks WHERE status='active' ORDER BY priority DESC, updated_at ASC LIMIT 1"
        ).fetchone()
        if active_row:
            return {
                "status": "success",
                "task": self._format_task(dict(active_row), full=True),
                "note": "This task is already 'active' — resume it rather than starting something new.",
            }

        rows = self.db.execute(
            "SELECT * FROM tasks WHERE status='pending' ORDER BY priority DESC, created_at ASC"
        ).fetchall()

        for row in rows:
            formatted = self._format_task(dict(row), full=True)
            if formatted["ready"]:
                return {"status": "success", "task": formatted}

        if not rows:
            return {
                "status": "success",
                "task": None,
                "message": "No pending tasks. Add one with 'task add', or check 'goal list' for what's left to plan.",
            }
        return {
            "status": "success",
            "task": None,
            "message": (
                f"{len(rows)} pending task(s) exist but all are blocked on unfinished "
                f"dependencies. Check 'task list --status pending' for details."
            ),
        }

    def task_status(self, public_id, new_status, note=None):
        if new_status not in TASK_STATUSES:
            raise ValueError(
                f"'{new_status}' is not a valid task status — use one of: {', '.join(TASK_STATUSES)}"
            )
        tid = parse_id("task", public_id)
        if not self.db.execute("SELECT 1 FROM tasks WHERE id=?", (tid,)).fetchone():
            raise ValueError(f"no task with id '{public_id}'. Run 'task list' to see valid ids.")
        now = now_iso()
        completed_at = now if new_status == "completed" else None
        self.db.execute(
            "UPDATE tasks SET status=?, updated_at=?, completed_at=? WHERE id=?",
            (new_status, now, completed_at, tid),
        )
        if note:
            self._append_note(tid, note)
        self.db.commit()
        return {"status": "success", "id": public_id, "new_status": new_status}

    def task_done(self, public_id, note=None):
        return self.task_status(public_id, "completed", note)

    def task_note(self, public_id, text):
        text = _require_text(text, "note text")
        tid = parse_id("task", public_id)
        if not self.db.execute("SELECT 1 FROM tasks WHERE id=?", (tid,)).fetchone():
            raise ValueError(f"no task with id '{public_id}'. Run 'task list' to see valid ids.")
        self._append_note(tid, text)
        self.db.commit()
        return {"status": "success", "id": public_id, "note": "appended"}

    def task_rm(self, public_id):
        tid = parse_id("task", public_id)
        if not self.db.execute("SELECT 1 FROM tasks WHERE id=?", (tid,)).fetchone():
            raise ValueError(f"no task with id '{public_id}'. Run 'task list' to see valid ids.")
        self.db.execute("DELETE FROM tasks WHERE id=?", (tid,))
        self.db.commit()
        return {"status": "success", "id": public_id, "deleted": True}

    def _append_note(self, tid, text):
        ts = now_iso()
        row = self.db.execute("SELECT notes FROM tasks WHERE id=?", (tid,)).fetchone()
        existing = (row["notes"] or "") if row else ""
        entry = f"[{ts}] {text}"
        merged = f"{existing}\n{entry}" if existing else entry
        self.db.execute("UPDATE tasks SET notes=?, updated_at=? WHERE id=?", (merged, ts, tid))

    def _format_task(self, row, full):
        dep_ids = json.loads(row["depends_on"] or "[]")
        unmet = []
        for d in dep_ids:
            dep_row = self.db.execute("SELECT status FROM tasks WHERE id=?", (d,)).fetchone()
            if not dep_row or dep_row["status"] != "completed":
                unmet.append(make_id("task", d))

        goal_title = None
        if row["goal_id"]:
            g = self.db.execute("SELECT title FROM goals WHERE id=?", (row["goal_id"],)).fetchone()
            goal_title = g["title"] if g else None

        if row["status"] == "active":
            ready = True
        elif row["status"] == "pending":
            ready = not unmet
        else:
            ready = False

        out = {
            "id": make_id("task", row["id"]),
            "goal_id": make_id("goal", row["goal_id"]) if row["goal_id"] else None,
            "goal_title": goal_title,
            "title": row["title"],
            "status": row["status"],
            "priority": row["priority"],
            "depends_on": [make_id("task", d) for d in dep_ids],
            "ready": ready,
        }
        if full:
            out["description"] = row["description"]
            out["notes"] = row["notes"]
            out["created_at"] = row["created_at"]
            out["updated_at"] = row["updated_at"]
            out["completed_at"] = row["completed_at"]
            if unmet:
                out["blocked_by"] = unmet
        else:
            out["description"] = truncate(row["description"])
        return out

    # ---------------- MEMORY ----------------

    def mem_add(self, category, key, value, tags=None, source="manual",
                importance=5, confidence=1.0, related_id=None):
        category = _require_text(category, "category")
        key = _require_text(key, "key")
        value = _require_text(value, "value")
        _validate_range(importance, 1, 10, "importance")
        _validate_range(confidence, 0.0, 1.0, "confidence")
        if related_id:
            s = str(related_id).strip().lower()
            if s.startswith("g"):
                gid = parse_id("goal", related_id)
                if not self.db.execute("SELECT 1 FROM goals WHERE id=?", (gid,)).fetchone():
                    raise ValueError(f"related id '{related_id}' does not match any goal.")
            elif s.startswith("t"):
                tid = parse_id("task", related_id)
                if not self.db.execute("SELECT 1 FROM tasks WHERE id=?", (tid,)).fetchone():
                    raise ValueError(f"related id '{related_id}' does not match any task.")
            else:
                raise ValueError(
                    f"'{related_id}' doesn't look like a goal or task id "
                    f"(expected something like 'g1' or 't3')."
                )
        now = now_iso()
        tags_json = json.dumps(tags or [])
        self.db.execute("""
            INSERT INTO memories
                (category, key, value, tags, source, importance, confidence,
                 related_id, access_count, created_at, updated_at, last_accessed)
            VALUES (?, ?, ?, ?, ?, ?, ?, ?, 0, ?, ?, NULL)
            ON CONFLICT(category, key) DO UPDATE SET
                value=excluded.value,
                tags=excluded.tags,
                source=excluded.source,
                importance=excluded.importance,
                confidence=excluded.confidence,
                related_id=excluded.related_id,
                updated_at=excluded.updated_at
        """, (category, key, value, tags_json, source, importance, confidence,
              related_id, now, now))
        self.db.commit()
        row = self.db.execute(
            "SELECT id FROM memories WHERE category=? AND key=?", (category, key)
        ).fetchone()
        mid = make_id("mem", row["id"])
        return {
            "status": "success",
            "id": mid,
            "hint": "Adding the same category+key again will update this entry instead of duplicating it.",
        }

    def mem_get(self, public_id):
        mid = parse_id("mem", public_id)
        row = self.db.execute("SELECT * FROM memories WHERE id=?", (mid,)).fetchone()
        if not row:
            raise ValueError(f"no memory with id '{public_id}'. Run 'mem list' to see valid ids.")
        now = now_iso()
        self.db.execute(
            "UPDATE memories SET access_count = access_count + 1, last_accessed = ? WHERE id = ?",
            (now, mid),
        )
        self.db.commit()
        mem = dict(row)
        mem["access_count"] = (mem["access_count"] or 0) + 1
        mem["last_accessed"] = now
        return {"status": "success", "memory": self._format_memory(mem, full=True)}

    def mem_search(self, query_text, min_importance=0, limit=20):
        limit = clamp_limit(limit)
        pattern = f"%{query_text}%"
        rows = self.db.execute("""
            SELECT * FROM memories
            WHERE (key LIKE ? OR value LIKE ? OR category LIKE ? OR tags LIKE ?)
            AND importance >= ?
            ORDER BY importance DESC, updated_at DESC
        """, (pattern, pattern, pattern, pattern, min_importance)).fetchall()
        total = len(rows)
        rows = rows[:limit]
        memories = [self._format_memory(dict(r), full=True) for r in rows]
        result = {"status": "success", "count": len(memories), "memories": memories}
        if total > len(memories):
            result["truncated"] = True
            result["total_matching"] = total
        return result

    def mem_list(self, category=None, limit=50):
        limit = clamp_limit(limit)
        query = "SELECT * FROM memories"
        params = []
        if category:
            query += " WHERE category = ?"
            params.append(category)
        query += " ORDER BY importance DESC, updated_at DESC"
        rows = self.db.execute(query, params).fetchall()
        total = len(rows)
        rows = rows[:limit]
        memories = [self._format_memory(dict(r), full=False) for r in rows]
        result = {"status": "success", "count": len(memories), "memories": memories}
        if total > len(memories):
            result["truncated"] = True
            result["total_matching"] = total
        return result

    def mem_importance(self, public_id, value):
        _validate_range(value, 1, 10, "importance")
        mid = parse_id("mem", public_id)
        if not self.db.execute("SELECT 1 FROM memories WHERE id=?", (mid,)).fetchone():
            raise ValueError(f"no memory with id '{public_id}'. Run 'mem list' to see valid ids.")
        self.db.execute(
            "UPDATE memories SET importance=?, updated_at=? WHERE id=?", (value, now_iso(), mid)
        )
        self.db.commit()
        return {"status": "success", "id": public_id, "importance": value}

    def mem_rm(self, public_id):
        mid = parse_id("mem", public_id)
        if not self.db.execute("SELECT 1 FROM memories WHERE id=?", (mid,)).fetchone():
            raise ValueError(f"no memory with id '{public_id}'. Run 'mem list' to see valid ids.")
        self.db.execute("DELETE FROM memories WHERE id=?", (mid,))
        self.db.commit()
        return {"status": "success", "id": public_id, "deleted": True}

    def _format_memory(self, row, full):
        tags = row.get("tags")
        if isinstance(tags, str):
            tags = json.loads(tags or "[]")
        out = {
            "id": make_id("mem", row["id"]),
            "category": row["category"],
            "key": row["key"],
            "importance": row["importance"],
            "tags": tags or [],
        }
        if full:
            out["value"] = row["value"]
            out["confidence"] = row["confidence"]
            out["source"] = row["source"]
            out["related_id"] = row["related_id"]
            out["access_count"] = row["access_count"]
            out["updated_at"] = row["updated_at"]
        else:
            out["value"] = truncate(row["value"], 80)
        return out

    # ---------------- ORIENTATION ----------------

    def brief(self, memory_limit=5):
        goal_rows = self.db.execute(
            "SELECT * FROM goals WHERE status='active' ORDER BY priority DESC, created_at ASC"
        ).fetchall()
        goals = []
        for r in goal_rows:
            counts = self.db.execute(
                "SELECT COUNT(*) AS total, SUM(status='completed') AS done FROM tasks WHERE goal_id=?",
                (r["id"],),
            ).fetchone()
            goals.append({
                "id": make_id("goal", r["id"]),
                "title": r["title"],
                "priority": r["priority"],
                "tasks_done": counts["done"] or 0,
                "tasks_total": counts["total"] or 0,
            })

        next_result = self.next_task()

        pending_count = self.db.execute(
            "SELECT COUNT(*) AS c FROM tasks WHERE status='pending'"
        ).fetchone()["c"]

        mem_rows = self.db.execute(
            "SELECT * FROM memories ORDER BY importance DESC, updated_at DESC LIMIT ?",
            (memory_limit,),
        ).fetchall()
        memories = [self._format_memory(dict(r), full=False) for r in mem_rows]

        return {
            "status": "success",
            "active_goals": goals,
            "pending_task_count": pending_count,
            "next_task": next_result.get("task"),
            "next_task_note": next_result.get("note") or next_result.get("message"),
            "top_memories": memories,
        }


# --------------------------------------------------------------------------
# CLI
# --------------------------------------------------------------------------

HELP_TEXT = """\
OpenClaw Store — planning + memory for long-running agent tasks
Database: __DB_PATH__

USAGE
  python3 openclaw_store.py <noun> <verb> [args] [--flags]

GOALS  (a goal groups related tasks)
  goal add "<title>" [--desc "text"] [--priority 1-10]
  goal list [--status active|paused|completed|abandoned|all]
  goal get <id>
  goal status <id> <active|paused|completed|abandoned>
  goal rm <id>                    tasks under it are kept, just unassigned

TASKS  (the actionable unit — this is what you actually execute)
  task add "<title>" [--goal <goal_id>] [--desc "text"] [--priority 1-10] [--depends <task_id>[,<task_id>...]]
  task list [--goal <goal_id>] [--status pending|active|blocked|completed|failed|cancelled]
  task get <id>
  task next                       the one task to work on right now
  task status <id> <status> [--note "what happened"]
  task done <id> [--note "..."]   shortcut for: task status <id> completed
  task note <id> "text"           append a note without changing status
  task rm <id>

MEMORY  (facts and lessons worth keeping for the rest of the run)
  mem add <category> "<key>" "<value>" [--tags a,b,c] [--importance 1-10] [--related <goal_or_task_id>]
  mem search "<query>" [--min-importance N] [--limit N]
  mem list [--category X]
  mem get <id>
  mem importance <id> <1-10>
  mem rm <id>

ORIENTATION
  brief                           snapshot: active goals + progress, next task, top memories
  help                             this text

NOTES
  - ids look like g1 (goal), t4 (task), m9 (memory). Reuse them exactly as returned.
  - every command prints ONE JSON object, either
    {"status": "success", ...} or {"status": "error", "message": "..."}.
  - "mem add" with a category+key that already exists updates that entry instead of
    creating a duplicate.
  - start (and resume) a session with: python3 openclaw_store.py brief
"""


def build_parser():
    parser = JSONArgumentParser(
        prog="openclaw_store.py",
        description="OpenClaw planning + memory store. Run 'help' for the full guide.",
    )
    noun = parser.add_subparsers(dest="noun")

    # ---- goal ----
    goal_p = noun.add_parser("goal", help="manage goals")
    goal_v = goal_p.add_subparsers(dest="verb", required=True)

    g_add = goal_v.add_parser("add", help="create a goal")
    g_add.add_argument("title")
    g_add.add_argument("--desc", default="", dest="description")
    g_add.add_argument("--priority", type=int, default=5)

    g_list = goal_v.add_parser("list", help="list goals")
    g_list.add_argument("--status", default=None, choices=GOAL_STATUSES + ["all"])
    g_list.add_argument("--limit", type=int, default=50)

    g_get = goal_v.add_parser("get", help="show one goal in full, with its tasks")
    g_get.add_argument("id")

    g_status = goal_v.add_parser("status", help="change a goal's status")
    g_status.add_argument("id")
    g_status.add_argument("value", choices=GOAL_STATUSES)

    g_rm = goal_v.add_parser("rm", help="delete a goal (tasks are kept, unassigned)")
    g_rm.add_argument("id")

    # ---- task ----
    task_p = noun.add_parser("task", help="manage tasks")
    task_v = task_p.add_subparsers(dest="verb", required=True)

    t_add = task_v.add_parser("add", help="create a task")
    t_add.add_argument("title")
    t_add.add_argument("--goal", default=None, dest="goal_id")
    t_add.add_argument("--desc", default="", dest="description")
    t_add.add_argument("--priority", type=int, default=5)
    t_add.add_argument("--depends", default=None, dest="depends_on")

    t_list = task_v.add_parser("list", help="list tasks")
    t_list.add_argument("--goal", default=None, dest="goal_id")
    t_list.add_argument("--status", default=None, choices=TASK_STATUSES + ["all"])
    t_list.add_argument("--limit", type=int, default=50)

    t_get = task_v.add_parser("get", help="show one task in full")
    t_get.add_argument("id")

    task_v.add_parser("next", help="the single task to work on right now")

    t_status = task_v.add_parser("status", help="change a task's status")
    t_status.add_argument("id")
    t_status.add_argument("value", choices=TASK_STATUSES)
    t_status.add_argument("--note", default=None)

    t_done = task_v.add_parser("done", help="shortcut for: status <id> completed")
    t_done.add_argument("id")
    t_done.add_argument("--note", default=None)

    t_note = task_v.add_parser("note", help="append a note to a task")
    t_note.add_argument("id")
    t_note.add_argument("text")

    t_rm = task_v.add_parser("rm", help="delete a task")
    t_rm.add_argument("id")

    # ---- mem ----
    mem_p = noun.add_parser("mem", help="manage memory")
    mem_v = mem_p.add_subparsers(dest="verb", required=True)

    m_add = mem_v.add_parser("add", help="store or update a memory")
    m_add.add_argument("category")
    m_add.add_argument("key")
    m_add.add_argument("value")
    m_add.add_argument("--tags", default=None)
    m_add.add_argument("--source", default="manual")
    m_add.add_argument("--importance", type=int, default=5)
    m_add.add_argument("--confidence", type=float, default=1.0)
    m_add.add_argument("--related", default=None, dest="related_id")

    m_search = mem_v.add_parser("search", help="search memory")
    m_search.add_argument("query")
    m_search.add_argument("--min-importance", type=int, default=0, dest="min_importance")
    m_search.add_argument("--limit", type=int, default=20)

    m_list = mem_v.add_parser("list", help="list memory")
    m_list.add_argument("--category", default=None)
    m_list.add_argument("--limit", type=int, default=50)

    m_get = mem_v.add_parser("get", help="show one memory in full")
    m_get.add_argument("id")

    m_importance = mem_v.add_parser("importance", help="change a memory's importance")
    m_importance.add_argument("id")
    m_importance.add_argument("value", type=int)

    m_rm = mem_v.add_parser("rm", help="delete a memory")
    m_rm.add_argument("id")

    # ---- orientation ----
    noun.add_parser("brief", help="snapshot of active goals, next task, top memories")
    noun.add_parser("help", help="show the full command reference")

    return parser


def dispatch(store, args):
    if args.noun == "goal":
        if args.verb == "add":
            return store.goal_add(args.title, args.description, args.priority)
        if args.verb == "list":
            return store.goal_list(args.status, args.limit)
        if args.verb == "get":
            return store.goal_get(args.id)
        if args.verb == "status":
            return store.goal_status(args.id, args.value)
        if args.verb == "rm":
            return store.goal_rm(args.id)

    if args.noun == "task":
        if args.verb == "add":
            return store.task_add(
                args.title, args.goal_id, args.description, args.priority, args.depends_on
            )
        if args.verb == "list":
            return store.task_list(args.goal_id, args.status, args.limit)
        if args.verb == "get":
            return store.task_get(args.id)
        if args.verb == "next":
            return store.next_task()
        if args.verb == "status":
            return store.task_status(args.id, args.value, args.note)
        if args.verb == "done":
            return store.task_done(args.id, args.note)
        if args.verb == "note":
            return store.task_note(args.id, args.text)
        if args.verb == "rm":
            return store.task_rm(args.id)

    if args.noun == "mem":
        if args.verb == "add":
            tags = [t.strip() for t in args.tags.split(",") if t.strip()] if args.tags else []
            return store.mem_add(
                args.category, args.key, args.value, tags, args.source,
                args.importance, args.confidence, args.related_id,
            )
        if args.verb == "search":
            return store.mem_search(args.query, args.min_importance, args.limit)
        if args.verb == "list":
            return store.mem_list(args.category, args.limit)
        if args.verb == "get":
            return store.mem_get(args.id)
        if args.verb == "importance":
            return store.mem_importance(args.id, args.value)
        if args.verb == "rm":
            return store.mem_rm(args.id)

    if args.noun == "brief":
        return store.brief()

    raise ValueError("unknown command. Run 'help' for the command reference.")


def main():
    parser = build_parser()
    args = parser.parse_args()

    db_path = os.environ.get("OPENCLAW_DB") or str(DEFAULT_DB_PATH)

    if args.noun is None or args.noun == "help":
        print(HELP_TEXT.replace("__DB_PATH__", str(db_path)))
        return

    store = None
    try:
        store = OpenClawStore(db_path)
        result = dispatch(store, args)
        exit_code = 0 if result.get("status") == "success" else 1
    except ValueError as e:
        result = {"status": "error", "error": "invalid_input", "message": str(e)}
        exit_code = 1
    except sqlite3.Error as e:
        result = {"status": "error", "error": "database_error", "message": str(e)}
        exit_code = 1
    except Exception as e:  # last-resort net: the agent should never see a raw traceback
        result = {"status": "error", "error": "internal_error", "message": str(e)}
        exit_code = 1
    finally:
        if store is not None:
            store.close()

    print(json.dumps(result, indent=2))
    sys.exit(exit_code)


if __name__ == "__main__":
    main()