agentic-mobile-control/storage/local_skills.py

"""Persistent local store for agent-managed skills and cloud-skill overrides.

Storage layer per CONSTITUTION.md: zero HTTP/MCP/LLM dependencies. This store
holds the agent's own authored skills and its local overrides of cloud skills,
in a SQLite file (``tasks/local_skills.sqlite3``) physically separate from the
synced catalog store (``tasks/skills.sqlite3``) and from task metadata. The
synced store's read-only-except-sync contract is preserved: this module never
writes to ``tasks/skills.sqlite3`` and :mod:`storage.skill_catalog` never
writes here.
"""

from __future__ import annotations

import json
import sqlite3
from pathlib import Path
from typing import Any
from uuid import uuid4

from core.models import utc_now
from skills_learning.models import (
    FlowTemplateSkill,
    KnowledgeSkill,
    Skill,
    SkillMetadata,
)

# Origin tag persisted on local rows (the read-merge layer derives the
# ``origin`` field shown to the LLM from which store a skill came from, not
# from this value, but we keep it explicit for traceability).
LOCAL_SOURCE = "local"


class LocalSkillStore:
    """SQLite-backed store of the agent's local skills and cloud-skill overrides.

    Two tables:
      * ``local_skills`` — authored local skills (id, content, ...).
      * ``overrides`` — local overrides keyed by the cloud skill id they
        shadow; the row carries the overriding content.
    """

    def __init__(self, db_path: str | Path = "tasks/local_skills.sqlite3") -> None:
        self.db_path = Path(db_path)
        self.db_path.parent.mkdir(parents=True, exist_ok=True)
        self._ensure_schema()

    # ------------------------------------------------------------------
    # Local skills — read
    # ------------------------------------------------------------------

    def list_local(self) -> list[SkillMetadata]:
        with self._connect() as conn:
            rows = conn.execute(
                """
                SELECT id, name, description, kind, tags_json, version,
                       source, created_at, updated_at
                FROM local_skills
                """
            ).fetchall()
        metas = [_row_to_metadata(row) for row in rows]
        metas.sort(key=lambda m: m.name)
        return metas

    def search_local(self, query: str) -> list[SkillMetadata]:
        like = f"%{query}%"
        with self._connect() as conn:
            rows = conn.execute(
                """
                SELECT id, name, description, kind, tags_json, version,
                       source, created_at, updated_at
                FROM local_skills
                WHERE name LIKE ? OR description LIKE ? OR tags_json LIKE ?
                """,
                (like, like, like),
            ).fetchall()
        return [_row_to_metadata(row) for row in rows]

    def get_local(self, skill_id: str) -> KnowledgeSkill | FlowTemplateSkill | None:
        with self._connect() as conn:
            row = conn.execute(
                "SELECT * FROM local_skills WHERE id = ?",
                (skill_id,),
            ).fetchone()
        return _row_to_skill(row) if row is not None else None

    # ------------------------------------------------------------------
    # Local skills — write
    # ------------------------------------------------------------------

    def create_local(self, skill: Skill) -> KnowledgeSkill | FlowTemplateSkill:
        """Persist ``skill`` as a new local skill with a fresh local id."""
        stored = _with_metadata(
            skill, id=uuid4().hex, source=LOCAL_SOURCE, reassign_timestamps=True
        )
        with self._connect() as conn:
            self._upsert_skill_row(conn, "local_skills", stored)
        return stored

    def update_local(self, skill: Skill) -> KnowledgeSkill | FlowTemplateSkill:
        existing = self.get_local(skill.id)
        if existing is None:
            raise KeyError(f"unknown local skill {skill.id}")
        stored = _with_metadata(skill, source=LOCAL_SOURCE, updated_at=utc_now())
        with self._connect() as conn:
            self._upsert_skill_row(conn, "local_skills", stored)
        return stored

    def delete_local(self, skill_id: str) -> bool:
        with self._connect() as conn:
            cur = conn.execute("DELETE FROM local_skills WHERE id = ?", (skill_id,))
        return cur.rowcount > 0

    # ------------------------------------------------------------------
    # Overrides — keyed by cloud skill id
    # ------------------------------------------------------------------

    def has_override(self, cloud_skill_id: str) -> bool:
        with self._connect() as conn:
            row = conn.execute(
                "SELECT 1 FROM overrides WHERE cloud_skill_id = ?",
                (cloud_skill_id,),
            ).fetchone()
        return row is not None

    def get_override(
        self, cloud_skill_id: str
    ) -> KnowledgeSkill | FlowTemplateSkill | None:
        with self._connect() as conn:
            row = conn.execute(
                "SELECT * FROM overrides WHERE cloud_skill_id = ?",
                (cloud_skill_id,),
            ).fetchone()
        return _override_row_to_skill(row) if row is not None else None

    def list_override_cloud_ids(self) -> set[str]:
        with self._connect() as conn:
            rows = conn.execute("SELECT cloud_skill_id FROM overrides").fetchall()
        return {row["cloud_skill_id"] for row in rows}

    def list_overrides(self) -> list[KnowledgeSkill | FlowTemplateSkill]:
        with self._connect() as conn:
            rows = conn.execute("SELECT * FROM overrides").fetchall()
        return [_override_row_to_skill(row) for row in rows]

    def upsert_override(
        self, cloud_skill_id: str, skill: Skill
    ) -> KnowledgeSkill | FlowTemplateSkill:
        """Create or replace the local override shadowing ``cloud_skill_id``."""
        stored = _with_metadata(
            skill, id=cloud_skill_id, source=LOCAL_SOURCE, reassign_timestamps=True
        )
        with self._connect() as conn:
            self._upsert_override_row(conn, cloud_skill_id, stored)
        return stored

    def remove_override(self, cloud_skill_id: str) -> bool:
        with self._connect() as conn:
            cur = conn.execute(
                "DELETE FROM overrides WHERE cloud_skill_id = ?",
                (cloud_skill_id,),
            )
        return cur.rowcount > 0

    def fork_override_to_local(self, cloud_skill_id: str) -> str | None:
        """Promote an override to a standalone local skill (design D9).

        Returns the new local skill id, or ``None`` if no override existed.
        The override row is deleted; subsequent reads see a local skill.
        """
        with self._connect() as conn:
            row = conn.execute(
                "SELECT * FROM overrides WHERE cloud_skill_id = ?",
                (cloud_skill_id,),
            ).fetchone()
            if row is None:
                return None
            skill = _override_row_to_skill(row)
            local_id = uuid4().hex
            stored = _with_metadata(
                skill,
                id=local_id,
                source=LOCAL_SOURCE,
                reassign_timestamps=True,
            )
            self._upsert_skill_row(conn, "local_skills", stored)
            conn.execute(
                "DELETE FROM overrides WHERE cloud_skill_id = ?",
                (cloud_skill_id,),
            )
        return local_id

    # ------------------------------------------------------------------
    # Internals
    # ------------------------------------------------------------------

    def _upsert_skill_row(
        self, conn: sqlite3.Connection, table: str, skill: Skill
    ) -> None:
        meta = skill.metadata
        content, steps_payload, parameters_payload = _skill_payload(skill)
        conn.execute(
            f"""
            INSERT INTO {table} (
                id, name, description, kind, tags_json, version, source,
                created_at, updated_at, content, steps_json, parameters_json
            ) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)
            ON CONFLICT(id) DO UPDATE SET
                name = excluded.name,
                description = excluded.description,
                kind = excluded.kind,
                tags_json = excluded.tags_json,
                version = excluded.version,
                source = excluded.source,
                created_at = excluded.created_at,
                updated_at = excluded.updated_at,
                content = excluded.content,
                steps_json = excluded.steps_json,
                parameters_json = excluded.parameters_json
            """,
            (
                meta.id,
                meta.name,
                meta.description,
                meta.kind,
                json.dumps(list(meta.tags)),
                meta.version,
                meta.source,
                meta.created_at.isoformat(),
                meta.updated_at.isoformat(),
                content,
                json.dumps(steps_payload),
                json.dumps(parameters_payload),
            ),
        )

    def _upsert_override_row(
        self,
        conn: sqlite3.Connection,
        cloud_skill_id: str,
        skill: Skill,
    ) -> None:
        meta = skill.metadata
        content, steps_payload, parameters_payload = _skill_payload(skill)
        conn.execute(
            """
            INSERT INTO overrides (
                cloud_skill_id, name, description, kind, tags_json, version,
                source, created_at, updated_at, content, steps_json,
                parameters_json
            ) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)
            ON CONFLICT(cloud_skill_id) DO UPDATE SET
                name = excluded.name,
                description = excluded.description,
                kind = excluded.kind,
                tags_json = excluded.tags_json,
                version = excluded.version,
                source = excluded.source,
                updated_at = excluded.updated_at,
                content = excluded.content,
                steps_json = excluded.steps_json,
                parameters_json = excluded.parameters_json
            """,
            (
                cloud_skill_id,
                meta.name,
                meta.description,
                meta.kind,
                json.dumps(list(meta.tags)),
                meta.version,
                meta.source,
                meta.created_at.isoformat(),
                meta.updated_at.isoformat(),
                content,
                json.dumps(steps_payload),
                json.dumps(parameters_payload),
            ),
        )

    def _ensure_schema(self) -> None:
        with self._connect() as conn:
            conn.execute(
                """
                CREATE TABLE IF NOT EXISTS local_skills (
                    id TEXT PRIMARY KEY,
                    name TEXT NOT NULL,
                    description TEXT NOT NULL DEFAULT '',
                    kind TEXT NOT NULL,
                    tags_json TEXT NOT NULL DEFAULT '[]',
                    version INTEGER NOT NULL DEFAULT 1,
                    source TEXT NOT NULL,
                    created_at TEXT NOT NULL,
                    updated_at TEXT NOT NULL,
                    content TEXT NOT NULL DEFAULT '',
                    steps_json TEXT NOT NULL DEFAULT '[]',
                    parameters_json TEXT NOT NULL DEFAULT '{}'
                )
                """
            )
            conn.execute(
                """
                CREATE TABLE IF NOT EXISTS overrides (
                    cloud_skill_id TEXT PRIMARY KEY,
                    name TEXT NOT NULL,
                    description TEXT NOT NULL DEFAULT '',
                    kind TEXT NOT NULL,
                    tags_json TEXT NOT NULL DEFAULT '[]',
                    version INTEGER NOT NULL DEFAULT 1,
                    source TEXT NOT NULL,
                    created_at TEXT NOT NULL,
                    updated_at TEXT NOT NULL,
                    content TEXT NOT NULL DEFAULT '',
                    steps_json TEXT NOT NULL DEFAULT '[]',
                    parameters_json TEXT NOT NULL DEFAULT '{}'
                )
                """
            )
            conn.execute(
                "CREATE INDEX IF NOT EXISTS idx_local_skills_name ON local_skills(name)"
            )

    def _connect(self) -> sqlite3.Connection:
        connection = sqlite3.connect(self.db_path)
        connection.row_factory = sqlite3.Row
        return connection


# ----------------------------------------------------------------------
# Helpers
# ----------------------------------------------------------------------


def _skill_payload(
    skill: Skill,
) -> tuple[str, list[dict[str, Any]], dict[str, dict[str, Any]]]:
    if isinstance(skill, KnowledgeSkill):
        return skill.content, [], {}
    if isinstance(skill, FlowTemplateSkill):
        return (
            "",
            [step.to_dict() for step in skill.steps],
            {name: dict(schema) for name, schema in skill.parameters.items()},
        )
    return "", [], {}


def _with_metadata(
    skill: Skill,
    *,
    id: str | None = None,
    source: str | None = None,
    updated_at: Any = None,
    reassign_timestamps: bool = False,
) -> KnowledgeSkill | FlowTemplateSkill:
    changes: dict[str, Any] = {}
    if id is not None:
        changes["id"] = id
    if source is not None:
        changes["source"] = source
    now = utc_now()
    if reassign_timestamps:
        changes["created_at"] = now
        changes["updated_at"] = now
    elif updated_at is not None:
        changes["updated_at"] = updated_at
    meta = skill.metadata
    new_meta = meta
    for key, value in changes.items():
        new_meta = _replace_meta(new_meta, key, value)
    if isinstance(skill, KnowledgeSkill):
        return KnowledgeSkill(metadata=new_meta, content=skill.content)
    return FlowTemplateSkill(
        metadata=new_meta,
        steps=list(skill.steps),
        parameters={name: dict(schema) for name, schema in skill.parameters.items()},
    )


def _replace_meta(meta: SkillMetadata, key: str, value: Any) -> SkillMetadata:
    from dataclasses import replace

    return replace(meta, **{key: value})


def _row_to_metadata(row: sqlite3.Row) -> SkillMetadata:
    return SkillMetadata.from_dict(
        {
            "id": row["id"],
            "name": row["name"],
            "description": row["description"],
            "kind": row["kind"],
            "tags": json.loads(row["tags_json"] or "[]"),
            "version": row["version"],
            "source": row["source"],
            "created_at": row["created_at"],
            "updated_at": row["updated_at"],
        }
    )


def _row_to_skill(row: sqlite3.Row) -> KnowledgeSkill | FlowTemplateSkill:
    return _skill_from_row(row, id_key="id")


def _override_row_to_skill(
    row: sqlite3.Row,
) -> KnowledgeSkill | FlowTemplateSkill:
    return _skill_from_row(row, id_key="cloud_skill_id")


def _skill_from_row(
    row: sqlite3.Row, *, id_key: str
) -> KnowledgeSkill | FlowTemplateSkill:
    data = {
        "id": row[id_key],
        "name": row["name"],
        "description": row["description"],
        "kind": row["kind"],
        "tags": json.loads(row["tags_json"] or "[]"),
        "version": row["version"],
        "source": row["source"],
        "created_at": row["created_at"],
        "updated_at": row["updated_at"],
        "content": row["content"],
        "steps": json.loads(row["steps_json"] or "[]"),
        "parameters": json.loads(row["parameters_json"] or "{}"),
    }
    if row["kind"] == "knowledge":
        return KnowledgeSkill.from_dict(data)
    return FlowTemplateSkill.from_dict(data)