mini-openclaw-v2/backend/services/assistant_store.py

"""Transactional JSON storage. Connections are closed, including error paths."""
import json
import sqlite3
from contextlib import contextmanager
from datetime import datetime, timezone
from hashlib import sha256
from pathlib import Path
from uuid import uuid4


def now():
    return datetime.now(timezone.utc).isoformat()


def uid(prefix):
    return prefix + "-" + uuid4().hex


def digest(value):
    return sha256(json.dumps(value, ensure_ascii=False, sort_keys=True, separators=(",", ":"), allow_nan=False).encode()).hexdigest()


@contextmanager
def transaction(path):
    path.parent.mkdir(parents=True, exist_ok=True)
    db = sqlite3.connect(path, timeout=15)
    db.row_factory = sqlite3.Row
    db.execute("PRAGMA busy_timeout=15000")
    try:
        with db:
            yield db
    finally:
        db.close()


def initialize(path):
    with transaction(path) as db:
        db.executescript("""
        CREATE TABLE IF NOT EXISTS assistant_objects (
            kind TEXT NOT NULL, id TEXT NOT NULL, project_id TEXT NOT NULL,
            revision INTEGER NOT NULL, body TEXT NOT NULL, PRIMARY KEY(kind,id));
        CREATE INDEX IF NOT EXISTS assistant_scope ON assistant_objects(kind,project_id);
        CREATE TABLE IF NOT EXISTS assistant_events (
            seq INTEGER PRIMARY KEY AUTOINCREMENT, project_id TEXT NOT NULL,
            run_id TEXT, body TEXT NOT NULL);
        CREATE TABLE IF NOT EXISTS assistant_idempotency (
            actor TEXT, operation TEXT, key TEXT, request_hash TEXT, object_id TEXT,
            PRIMARY KEY(actor,operation,key));
        CREATE TABLE IF NOT EXISTS assistant_jobs (
            id TEXT PRIMARY KEY, project_id TEXT, state TEXT, owner TEXT,
            lease_until REAL NOT NULL DEFAULT 0, attempts INTEGER NOT NULL DEFAULT 0);
        CREATE TABLE IF NOT EXISTS assistant_meta (key TEXT PRIMARY KEY, value TEXT);
        """)
        schema = Path(__file__).resolve().parents[1] / "fixtures" / "xingbang_demo" / "business_flow_schema.sql"
        if schema.exists():
            db.executescript(schema.read_text(encoding="utf-8"))


def get(db, kind, object_id):
    row = db.execute("SELECT body FROM assistant_objects WHERE kind=? AND id=?", (kind, object_id)).fetchone()
    if row is None:
        raise KeyError(f"{kind} 不存在")
    return json.loads(row[0])


def listing(db, kind, project_id=None):
    if project_id is None:
        rows = db.execute("SELECT body FROM assistant_objects WHERE kind=? ORDER BY rowid DESC", (kind,))
    else:
        rows = db.execute("SELECT body FROM assistant_objects WHERE kind=? AND project_id=? ORDER BY rowid DESC", (kind, project_id))
    return [json.loads(row[0]) for row in rows]


def put(db, kind, body, expected=None, immutable=False):
    old = db.execute("SELECT revision FROM assistant_objects WHERE kind=? AND id=?", (kind, body["id"])).fetchone()
    if immutable and old:
        raise ValueError("发布版本和历史快照不可覆盖")
    if expected is not None and (not old or old[0] != expected):
        raise Conflict("内容已更新，请刷新后重试")
    body = dict(body, revision=(old[0] + 1 if old else 1))
    db.execute("INSERT INTO assistant_objects VALUES (?,?,?,?,?) ON CONFLICT(kind,id) DO UPDATE SET revision=excluded.revision,body=excluded.body",
               (kind, body["id"], body.get("project_id", ""), body["revision"], json.dumps(body, ensure_ascii=False, allow_nan=False)))
    return body


def event(db, project_id, action, actor, detail=None, run_id=None):
    body = {"time": now(), "action": action, "actor": actor, "detail": detail or {}}
    db.execute("INSERT INTO assistant_events(project_id,run_id,body) VALUES (?,?,?)", (project_id, run_id, json.dumps(body, ensure_ascii=False)))


def time_event(db, project_id, phase_code, phase_name, role, category, actor, source_type, source_id,
               status="running", started_at=None, ended_at=None, detail=None, event_id=None):
    """Create an immutable phase interval; callers close it with finish_time_event."""
    event_id = event_id or uid("time")
    db.execute(
        "INSERT INTO business_time_events VALUES (?,?,?,?,?,?,?,?,?,?,?,?,?)",
        (event_id, project_id, phase_code, phase_name, role, category, status,
         started_at or now(), ended_at, actor, source_type, source_id,
         json.dumps(detail or {}, ensure_ascii=False)),
    )
    return event_id


def finish_time_event(db, source_type, source_id, status="completed", ended_at=None):
    db.execute(
        "UPDATE business_time_events SET status=?,ended_at=? WHERE source_type=? AND source_id=? AND status='running'",
        (status, ended_at or now(), source_type, source_id),
    )


def time_listing(db, project_ids=None):
    query = "SELECT * FROM business_time_events"
    params = []
    if project_ids is not None:
        if not project_ids:
            return []
        query += " WHERE project_id IN (" + ",".join("?" for _ in project_ids) + ")"
        params.extend(project_ids)
    query += " ORDER BY started_at"
    result = []
    for row in db.execute(query, params):
        item = dict(row)
        item["detail"] = json.loads(item["detail"] or "{}")
        result.append(item)
    return result


class Conflict(ValueError):
    pass