"""V1 EMDS Workbench metadata persistence."""

from __future__ import annotations

import hashlib
import json
import sqlite3
import uuid
from dataclasses import dataclass
from datetime import datetime
from pathlib import Path
from typing import Any, Dict, List, Optional


@dataclass
class WorkspaceMetadata:
    workspace_id: str
    name: str
    description: str
    qgis_project_file: str
    input_layers: List[str]
    created_at: str
    updated_at: str
    metadata: Dict[str, Any]

    @property
    def project_id(self) -> str:
        return self.workspace_id

    @property
    def project_file(self) -> str:
        return self.qgis_project_file


@dataclass
class StudyMetadata:
    study_id: str
    workspace_id: str
    name: str
    comments: str
    output_layer_name: str
    reference_layer_name: str
    created_at: str
    updated_at: str
    metadata: Dict[str, Any]

    @property
    def spatial_reference_id(self) -> str:
        return self.study_id

    @property
    def project_id(self) -> str:
        return self.workspace_id


@dataclass
class AnalysisMetadata:
    analysis_id: str
    study_id: str
    workspace_id: str
    name: str
    description: str
    created_at: str
    updated_at: str
    metadata: Dict[str, Any]

    @property
    def provenance_id(self) -> str:
        return self.analysis_id

    @property
    def spatial_reference_id(self) -> str:
        return self.study_id

    @property
    def project_id(self) -> str:
        return self.workspace_id


@dataclass
class TaskMetadata:
    task_id: str
    analysis_id: str
    workspace_id: str
    study_id: str
    parent_task_id: str
    name: str
    engine_type: str
    model_path: str
    input_tables: List[str]
    result_tables: List[str]
    created_at: str
    updated_at: str
    metadata: Dict[str, Any]
    model_snapshot_id: str = ""

    @property
    def provenance_id(self) -> str:
        return self.analysis_id

    @property
    def project_id(self) -> str:
        return self.workspace_id

    @property
    def spatial_reference_id(self) -> str:
        return self.study_id


@dataclass
class ResultMetadata:
    result_id: str
    task_id: str
    analysis_id: str
    workspace_id: str
    study_id: str
    layer_name: str
    table_name: str
    created_at: str
    metadata: Dict[str, Any]

    @property
    def results_id(self) -> str:
        return self.result_id

    @property
    def workflow_id(self) -> str:
        return self.task_id

    @property
    def provenance_id(self) -> str:
        return self.analysis_id

    @property
    def project_id(self) -> str:
        return self.workspace_id

    @property
    def spatial_reference_id(self) -> str:
        return self.study_id


@dataclass
class WorkflowDefinitionMetadata:
    workflow_definition_id: str
    project_id: str
    name: str
    description: str
    payload: Dict[str, Any]
    created_at: str
    updated_at: str

    @property
    def workspace_id(self) -> str:
        return self.project_id


@dataclass
class ModelSnapshotMetadata:
    """Immutable record of a model as used by an analysis (blobs fetched separately)."""

    id: str
    model_kind: str
    display_name: str
    source_path: str
    source_format: str
    content_sha256: str
    byte_size: int
    schema_version: str
    authoring_tool: str
    authoring_tool_version: str
    engine_version: str
    model_summary: Dict[str, Any]
    registry_id: str
    registry_version: str
    file_mtime: str
    created_at: str
    created_by: str
    canonical_sha256: str = ""


@dataclass
class WorkflowDefinitionRevisionMetadata:
    id: str
    workflow_definition_id: str
    revision_number: int
    payload: Dict[str, Any]
    payload_sha256: str
    change_summary: str
    created_at: str
    created_by: str


@dataclass
class WorkflowRunMetadata:
    id: str
    workspace_id: str
    study_id: str
    analysis_id: str
    workflow_definition_id: str
    workflow_definition_revision_id: str
    run_kind: str
    scenario_id: str
    parent_run_id: str
    label: str
    notes: str
    status: str
    start_time: str
    end_time: str
    duration_ms: int
    emds_version: str
    gis_environment: str
    gis_version: str
    created_by: str
    created_at: str


@dataclass
class WorkflowStepRunMetadata:
    id: str
    run_id: str
    workflow_step_id: str
    task_id: str
    parent_step_run_id: str
    step_index: int
    name: str
    engine_type: str
    engine_version: str
    model_snapshot_id: str
    parameters: Dict[str, Any]
    field_mappings: List[Any]
    input_refs: List[Any]
    output_summary: Dict[str, Any]
    result_id: str
    status: str
    start_time: str
    end_time: str
    duration_ms: int
    rows_processed: int
    error_message: str
    created_at: str


@dataclass
class ResultLineageMetadata:
    id: str
    result_id: str
    step_run_id: str
    run_id: str
    model_snapshot_id: str
    upstream_result_id: str
    derived_from_result_id: str
    created_at: str


def _default_metadata_db_path() -> Path:
    from .app_paths import ensure_workspace

    return ensure_workspace().metadata_db_path


class WorkbenchMetadataManager:
    """Manage the clean v1 Workbench schema."""

    DEFAULT_DB_NAME = "emdsInfo.sqlite"
    PROJECT_DB_NAME = DEFAULT_DB_NAME

    def __init__(self, db_path: str = None):
        self.db_path = Path(db_path) if db_path else _default_metadata_db_path()
        self.db_path.parent.mkdir(parents=True, exist_ok=True)
        self.last_save_error: Optional[str] = None
        self._init_database()
        self._ensure_columns()

    @classmethod
    def for_project_directory(cls, project_dir: Path | str) -> "WorkbenchMetadataManager":
        return cls(str(Path(project_dir) / cls.PROJECT_DB_NAME))

    def _connect(self) -> sqlite3.Connection:
        # Autocommit (isolation_level=None) so a failed statement never leaves an
        # open transaction holding a lock that would wedge the metadata DB for the
        # rest of the QGIS session; busy_timeout rides out brief cross-connection
        # contention. Methods that need a multi-statement transaction still drive
        # it explicitly with BEGIN/COMMIT/ROLLBACK.
        conn = sqlite3.connect(self.db_path, isolation_level=None)
        conn.execute("PRAGMA busy_timeout = 5000")
        conn.execute("PRAGMA foreign_keys = ON")
        conn.row_factory = sqlite3.Row
        return conn

    @staticmethod
    def _now() -> str:
        return datetime.now().isoformat()

    @staticmethod
    def _decode(value: Optional[str], default):
        if not value:
            return default
        try:
            return json.loads(value)
        except Exception:
            return default

    @staticmethod
    def _sha256_bytes(data: bytes) -> str:
        return hashlib.sha256(data or b"").hexdigest()

    @staticmethod
    def _sha256_text(text: str) -> str:
        return hashlib.sha256((text or "").encode("utf-8")).hexdigest()

    @staticmethod
    def _canonical_json(obj: Any) -> str:
        """Stable JSON for hashing/dedupe (sorted keys, compact separators)."""
        return json.dumps(obj or {}, sort_keys=True, separators=(",", ":"), default=str)

    def _init_database(self) -> None:
        conn = self._connect()
        cur = conn.cursor()
        # Phase 3.5c: `project_file` is the canonical column. The
        # legacy `qgis_project_file` column stays in the schema for
        # the grace cycle so old rows still load via the reader
        # fallback, but it's nullable (new INSERTs leave it NULL) and
        # carries no UNIQUE constraint. The canonical column carries
        # the uniqueness instead. Existing DBs with the old NOT NULL
        # UNIQUE shape on qgis_project_file are tolerated — CREATE
        # TABLE IF NOT EXISTS leaves them alone and _ensure_columns
        # backfills `project_file` so new INSERTs succeed.
        cur.execute(
            """CREATE TABLE IF NOT EXISTS workspaces (
                workspace_id TEXT PRIMARY KEY, name TEXT NOT NULL, description TEXT,
                qgis_project_file TEXT, project_file TEXT NOT NULL UNIQUE,
                input_layers TEXT NOT NULL,
                created_at TEXT NOT NULL, updated_at TEXT NOT NULL, metadata TEXT)"""
        )
        cur.execute(
            """CREATE TABLE IF NOT EXISTS studies (
                study_id TEXT PRIMARY KEY, workspace_id TEXT NOT NULL, name TEXT NOT NULL,
                comments TEXT, output_layer_name TEXT, reference_layer_name TEXT,
                created_at TEXT NOT NULL, updated_at TEXT NOT NULL, metadata TEXT,
                FOREIGN KEY(workspace_id) REFERENCES workspaces(workspace_id) ON DELETE CASCADE)"""
        )
        cur.execute(
            """CREATE TABLE IF NOT EXISTS analyses (
                analysis_id TEXT PRIMARY KEY, study_id TEXT NOT NULL, workspace_id TEXT NOT NULL,
                name TEXT NOT NULL, description TEXT, created_at TEXT NOT NULL,
                updated_at TEXT NOT NULL, metadata TEXT,
                FOREIGN KEY(study_id) REFERENCES studies(study_id) ON DELETE CASCADE,
                FOREIGN KEY(workspace_id) REFERENCES workspaces(workspace_id) ON DELETE CASCADE)"""
        )
        cur.execute(
            """CREATE TABLE IF NOT EXISTS tasks (
                task_id TEXT PRIMARY KEY, analysis_id TEXT NOT NULL, workspace_id TEXT NOT NULL,
                study_id TEXT NOT NULL, parent_task_id TEXT, name TEXT NOT NULL,
                engine_type TEXT NOT NULL, model_path TEXT, model_snapshot_id TEXT, input_tables TEXT,
                result_tables TEXT, created_at TEXT NOT NULL, updated_at TEXT NOT NULL,
                metadata TEXT, FOREIGN KEY(analysis_id) REFERENCES analyses(analysis_id) ON DELETE CASCADE,
                FOREIGN KEY(parent_task_id) REFERENCES tasks(task_id) ON DELETE CASCADE)"""
        )
        cur.execute(
            """CREATE TABLE IF NOT EXISTS results (
                result_id TEXT PRIMARY KEY, task_id TEXT NOT NULL, analysis_id TEXT NOT NULL,
                workspace_id TEXT NOT NULL, study_id TEXT NOT NULL, layer_name TEXT, table_name TEXT,
                created_at TEXT NOT NULL, metadata TEXT,
                FOREIGN KEY(task_id) REFERENCES tasks(task_id) ON DELETE CASCADE)"""
        )
        cur.execute("CREATE INDEX IF NOT EXISTS idx_studies_workspace ON studies(workspace_id)")
        cur.execute("CREATE INDEX IF NOT EXISTS idx_analyses_study ON analyses(study_id)")
        cur.execute("CREATE INDEX IF NOT EXISTS idx_tasks_analysis ON tasks(analysis_id)")
        cur.execute(
            """CREATE TABLE IF NOT EXISTS field_mapping_presets (
                preset_id TEXT PRIMARY KEY, workspace_id TEXT NOT NULL, engine_type TEXT NOT NULL,
                model_identifier TEXT NOT NULL, input_table_name TEXT NOT NULL, payload TEXT NOT NULL,
                created_at TEXT NOT NULL, updated_at TEXT NOT NULL,
                UNIQUE(workspace_id, engine_type, model_identifier, input_table_name))"""
        )
        cur.execute(
            """CREATE TABLE IF NOT EXISTS workflow_definitions (
                workflow_definition_id TEXT PRIMARY KEY, workspace_id TEXT NOT NULL,
                name TEXT NOT NULL, description TEXT, payload TEXT NOT NULL,
                created_at TEXT NOT NULL, updated_at TEXT NOT NULL,
                UNIQUE(workspace_id, name),
                FOREIGN KEY(workspace_id) REFERENCES workspaces(workspace_id) ON DELETE CASCADE)"""
        )

        # ---- Provenance & model-storage tables (contract: docs/provenance/00) ----
        # Greenfield base schema (no migrations). FKs are intentionally omitted
        # on these tables so write-order (e.g. a run row before its result row)
        # never trips PRAGMA foreign_keys; lineage integrity is enforced in code.
        cur.execute(
            """CREATE TABLE IF NOT EXISTS model_snapshots (
                id TEXT PRIMARY KEY, model_kind TEXT NOT NULL, display_name TEXT,
                source_path TEXT, source_format TEXT, content_sha256 TEXT NOT NULL,
                content_blob BLOB NOT NULL, canonical_blob BLOB, canonical_sha256 TEXT,
                byte_size INTEGER, schema_version TEXT, authoring_tool TEXT,
                authoring_tool_version TEXT, engine_version TEXT, model_summary_json TEXT,
                registry_id TEXT, registry_version TEXT, file_mtime TEXT,
                created_at TEXT NOT NULL, created_by TEXT)"""
        )
        cur.execute("CREATE INDEX IF NOT EXISTS idx_model_snapshots_sha ON model_snapshots(content_sha256)")
        cur.execute("CREATE INDEX IF NOT EXISTS idx_model_snapshots_registry ON model_snapshots(registry_id, registry_version)")
        cur.execute(
            """CREATE TABLE IF NOT EXISTS workflow_definition_revisions (
                id TEXT PRIMARY KEY, workflow_definition_id TEXT NOT NULL,
                revision_number INTEGER NOT NULL, payload TEXT NOT NULL,
                payload_sha256 TEXT NOT NULL, change_summary TEXT,
                created_at TEXT NOT NULL, created_by TEXT,
                UNIQUE(workflow_definition_id, revision_number))"""
        )
        cur.execute(
            """CREATE TABLE IF NOT EXISTS workflow_runs (
                id TEXT PRIMARY KEY, workspace_id TEXT, study_id TEXT, analysis_id TEXT,
                workflow_definition_id TEXT, workflow_definition_revision_id TEXT,
                run_kind TEXT NOT NULL DEFAULT 'baseline', scenario_id TEXT, parent_run_id TEXT,
                label TEXT, notes TEXT, status TEXT, start_time TEXT, end_time TEXT,
                duration_ms INTEGER DEFAULT 0, emds_version TEXT, gis_environment TEXT,
                gis_version TEXT, created_by TEXT, created_at TEXT NOT NULL)"""
        )
        cur.execute("CREATE INDEX IF NOT EXISTS idx_workflow_runs_analysis ON workflow_runs(analysis_id)")
        cur.execute("CREATE INDEX IF NOT EXISTS idx_workflow_runs_scenario ON workflow_runs(scenario_id)")
        cur.execute(
            """CREATE TABLE IF NOT EXISTS workflow_step_runs (
                id TEXT PRIMARY KEY, run_id TEXT NOT NULL, workflow_step_id TEXT, task_id TEXT,
                parent_step_run_id TEXT, step_index INTEGER, name TEXT, engine_type TEXT,
                engine_version TEXT, model_snapshot_id TEXT, parameters_json TEXT,
                field_mappings_json TEXT, input_refs_json TEXT, output_summary_json TEXT,
                result_id TEXT, status TEXT, start_time TEXT, end_time TEXT,
                duration_ms INTEGER DEFAULT 0, rows_processed INTEGER DEFAULT 0,
                error_message TEXT, created_at TEXT NOT NULL)"""
        )
        cur.execute("CREATE INDEX IF NOT EXISTS idx_step_runs_run ON workflow_step_runs(run_id)")
        cur.execute("CREATE INDEX IF NOT EXISTS idx_step_runs_task ON workflow_step_runs(task_id)")
        cur.execute("CREATE INDEX IF NOT EXISTS idx_step_runs_snapshot ON workflow_step_runs(model_snapshot_id)")
        cur.execute(
            """CREATE TABLE IF NOT EXISTS result_lineage (
                id TEXT PRIMARY KEY, result_id TEXT NOT NULL, step_run_id TEXT, run_id TEXT,
                model_snapshot_id TEXT, upstream_result_id TEXT, derived_from_result_id TEXT,
                created_at TEXT NOT NULL)"""
        )
        cur.execute("CREATE INDEX IF NOT EXISTS idx_result_lineage_result ON result_lineage(result_id)")

        conn.commit()
        conn.close()

    def _ensure_columns(self) -> None:
        """Self-heal database schema to support cross-platform parity and avoid migration/insert errors."""
        try:
            conn = self._connect()
            cur = conn.cursor()

            # 1. workspaces table — Phase 3.5c canonical alignment.
            cur.execute("PRAGMA table_info(workspaces)")
            ws_info = list(cur.fetchall())
            ws_cols = {row["name"].lower() for row in ws_info}

            # Step 1a: add missing columns first (cheap ALTER TABLE).
            # The rebuild below needs both columns present on the source
            # table when it copies rows; backfill project_file from the
            # legacy column for any rows that pre-date the rename.
            if "project_file" not in ws_cols:
                cur.execute("ALTER TABLE workspaces ADD COLUMN project_file TEXT")
                conn.commit()
                if "qgis_project_file" in ws_cols:
                    cur.execute("UPDATE workspaces SET project_file = qgis_project_file WHERE project_file IS NULL")
                    conn.commit()
                ws_cols.add("project_file")
            if "qgis_project_file" not in ws_cols:
                # Re-add the legacy column so the reader fallback still
                # works for rows that arrived via legacy paths. Stays
                # nullable; never written from new INSERTs.
                cur.execute("ALTER TABLE workspaces ADD COLUMN qgis_project_file TEXT")
                conn.commit()
                ws_cols.add("qgis_project_file")

            # Step 1b: detect legacy NOT NULL UNIQUE constraint on
            # qgis_project_file. PRAGMA table_info returns
            # (cid, name, type, notnull, dflt_value, pk). When the
            # legacy column carries notnull=1 we must rebuild the
            # table to relax the constraint — SQLite can't ALTER it
            # away in place. The rebuild also enforces NOT NULL UNIQUE
            # on the canonical project_file column.
            cur.execute("PRAGMA table_info(workspaces)")
            ws_info = list(cur.fetchall())
            legacy_qpf_notnull = any(
                row["name"].lower() == "qgis_project_file" and row["notnull"]
                for row in ws_info
            )
            if legacy_qpf_notnull:
                self._rebuild_workspaces_table_canonical(conn, cur)

            # 2. tasks table
            cur.execute("PRAGMA table_info(tasks)")
            t_cols = {row["name"].lower() for row in cur.fetchall()}
            if "output_layer_name" not in t_cols:
                cur.execute("ALTER TABLE tasks ADD COLUMN output_layer_name TEXT")
                conn.commit()
            if "workflow_definition_id" not in t_cols:
                cur.execute("ALTER TABLE tasks ADD COLUMN workflow_definition_id TEXT")
                conn.commit()
            if "model_snapshot_id" not in t_cols:
                cur.execute("ALTER TABLE tasks ADD COLUMN model_snapshot_id TEXT")
                conn.commit()

            # 3. workflow_definitions table
            cur.execute("PRAGMA table_info(workflow_definitions)")
            wd_cols = {row["name"].lower() for row in cur.fetchall()}
            if "failure_mode" not in wd_cols:
                cur.execute("ALTER TABLE workflow_definitions ADD COLUMN failure_mode TEXT")
                conn.commit()

            conn.close()
        except Exception as exc:
            print(f"EMDS schema self-heal warning: {exc}")

    @staticmethod
    def _rebuild_workspaces_table_canonical(conn, cur) -> None:
        """Rebuild the workspaces table in Phase 3.5c canonical shape.

        Used when an existing DB has the legacy `qgis_project_file
        TEXT NOT NULL UNIQUE` constraint that blocks new INSERTs from
        writing NULL into that column. SQLite can't ALTER away a
        NOT NULL constraint in place, so we copy → drop → rename
        inside a transaction. Idempotent and safe to call on an
        already-canonical DB (no-op via the caller's notnull check).

        Migrates the file path from `qgis_project_file` (legacy) into
        `project_file` (canonical) for every row that had one.
        """
        cur.execute("BEGIN")
        try:
            cur.execute(
                """CREATE TABLE workspaces_new (
                    workspace_id TEXT PRIMARY KEY,
                    name TEXT NOT NULL,
                    description TEXT,
                    qgis_project_file TEXT,
                    project_file TEXT NOT NULL UNIQUE,
                    input_layers TEXT NOT NULL,
                    created_at TEXT NOT NULL,
                    updated_at TEXT NOT NULL,
                    metadata TEXT
                )"""
            )
            # Pull the path from whichever legacy column held it
            # (project_file may already exist with a backfilled value
            # from a prior self-heal run).
            cur.execute(
                """INSERT INTO workspaces_new
                   (workspace_id, name, description,
                    qgis_project_file, project_file, input_layers,
                    created_at, updated_at, metadata)
                   SELECT
                       workspace_id, name, description,
                       NULL,
                       COALESCE(
                           NULLIF(IFNULL(project_file, ''), ''),
                           qgis_project_file
                       ),
                       input_layers,
                       created_at, updated_at, metadata
                   FROM workspaces"""
            )
            cur.execute("DROP TABLE workspaces")
            cur.execute("ALTER TABLE workspaces_new RENAME TO workspaces")
            conn.commit()
        except Exception:
            conn.rollback()
            raise

    def save_workspace(self, workspace_id: str, name: str, description: str, qgis_project_file: str, input_layers: List[str], metadata: Dict[str, Any] = None) -> bool:
        """Persist a workspace row.

        Phase 3.5c: `project_file` is the canonical column (matches the
        ArcGIS Pro `Workspace.cs:9,16` shape — product-neutral by
        content type). The legacy `qgis_project_file` column stays in
        the schema for the grace cycle so old rows still load via the
        reader fallback, but new INSERTs leave it NULL — the canonical
        column carries the path going forward.

        Parameter name kept as `qgis_project_file` for back-compat with
        existing call sites; semantically it's "the workspace's project
        file" (a `.qgz` here, a `.aprx` in ArcGIS Pro).
        """
        self.last_save_error = None
        try:
            existing = self.get_workspace(workspace_id)
            now = self._now()
            payload = dict(metadata or {})
            # Mirror the canonical column into the metadata JSON so
            # callers reading the JSON blob get the path too.
            payload.setdefault("project_file", qgis_project_file)
            conn = self._connect()
            conn.cursor().execute(
                """INSERT INTO workspaces
                (workspace_id, name, description, qgis_project_file, project_file, input_layers, created_at, updated_at, metadata)
                VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?)
                ON CONFLICT(workspace_id) DO UPDATE SET
                    name=excluded.name, description=excluded.description,
                    project_file=excluded.project_file, qgis_project_file=NULL, input_layers=excluded.input_layers,
                    updated_at=excluded.updated_at, metadata=excluded.metadata""",
                (workspace_id, name, description, None, qgis_project_file, json.dumps(input_layers or []),
                 existing.created_at if existing else now, now, json.dumps(payload)),
            )
            conn.commit()
            conn.close()
            return True
        except Exception as exc:
            self.last_save_error = f"{type(exc).__name__}: {exc}"
            print(f"Error saving workspace metadata: {exc}")
            return False

    def _workspace_from_row(self, row) -> WorkspaceMetadata:
        """Reader for the workspaces table.

        Phase 3.5c: prefer the canonical `project_file` column; fall
        back to legacy `qgis_project_file` for rows written before the
        canonical switch. Both branches stay populated through the
        grace cycle.
        """
        keys = row.keys() if hasattr(row, "keys") else []
        workspace_pf = ""
        if "project_file" in keys and row["project_file"]:
            workspace_pf = row["project_file"]
        elif "qgis_project_file" in keys and row["qgis_project_file"]:
            workspace_pf = row["qgis_project_file"]

        return WorkspaceMetadata(row["workspace_id"], row["name"] or "", row["description"] or "",
                                 workspace_pf, self._decode(row["input_layers"], []),
                                 row["created_at"] or "", row["updated_at"] or "", self._decode(row["metadata"], {}))

    def get_workspace(self, workspace_id: str) -> Optional[WorkspaceMetadata]:
        try:
            conn = self._connect()
            row = conn.cursor().execute("SELECT * FROM workspaces WHERE workspace_id = ?", (workspace_id,)).fetchone()
            conn.close()
            return self._workspace_from_row(row) if row else None
        except Exception as exc:
            print(f"Error retrieving workspace: {exc}")
            return None

    def get_workspace_by_qgis_project_file(self, qgis_project_file: str) -> Optional[WorkspaceMetadata]:
        try:
            conn = self._connect()
            row = conn.cursor().execute(
                "SELECT * FROM workspaces WHERE qgis_project_file = ? OR project_file = ?",
                (qgis_project_file, qgis_project_file)
            ).fetchone()
            conn.close()
            return self._workspace_from_row(row) if row else None
        except Exception as exc:
            print(f"Error retrieving workspace by QGIS project file: {exc}")
            return None

    def get_all_workspaces(self) -> List[WorkspaceMetadata]:
        try:
            conn = self._connect()
            rows = conn.cursor().execute("SELECT * FROM workspaces ORDER BY updated_at DESC, created_at DESC").fetchall()
            conn.close()
            return [self._workspace_from_row(row) for row in rows]
        except Exception as exc:
            print(f"Error retrieving workspaces: {exc}")
            return []

    def update_workspace(self, workspace_id: str, name: str = None, description: str = None, metadata: Dict[str, Any] = None) -> bool:
        existing = self.get_workspace(workspace_id)
        if existing is None:
            return False
        merged = dict(existing.metadata or {})
        if metadata:
            merged.update(metadata)
        return self.save_workspace(workspace_id, name or existing.name,
                                   existing.description if description is None else description,
                                   existing.qgis_project_file, existing.input_layers, merged)

    def save_study(self, study_id: str, workspace_id: str, name: str, comments: str, output_layer_name: str, reference_layer_name: str = "", metadata: Dict[str, Any] = None) -> bool:
        try:
            existing = self.get_study(study_id)
            now = self._now()
            conn = self._connect()
            conn.cursor().execute(
                """INSERT INTO studies
                (study_id, workspace_id, name, comments, output_layer_name, reference_layer_name, created_at, updated_at, metadata)
                VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?)
                ON CONFLICT(study_id) DO UPDATE SET
                    workspace_id=excluded.workspace_id, name=excluded.name, comments=excluded.comments,
                    output_layer_name=excluded.output_layer_name, reference_layer_name=excluded.reference_layer_name,
                    updated_at=excluded.updated_at, metadata=excluded.metadata""",
                (study_id, workspace_id, name, comments, output_layer_name, reference_layer_name,
                 existing.created_at if existing else now, now, json.dumps(metadata or {})),
            )
            conn.commit()
            conn.close()
            return True
        except Exception as exc:
            print(f"Error saving study metadata: {exc}")
            return False

    def _study_from_row(self, row) -> StudyMetadata:
        return StudyMetadata(row["study_id"], row["workspace_id"], row["name"] or "", row["comments"] or "",
                             row["output_layer_name"] or "", row["reference_layer_name"] or "",
                             row["created_at"] or "", row["updated_at"] or "", self._decode(row["metadata"], {}))

    def get_study(self, study_id: str) -> Optional[StudyMetadata]:
        try:
            conn = self._connect()
            row = conn.cursor().execute("SELECT * FROM studies WHERE study_id = ?", (study_id,)).fetchone()
            conn.close()
            return self._study_from_row(row) if row else None
        except Exception as exc:
            print(f"Error retrieving study: {exc}")
            return None

    def get_studies_for_workspace(self, workspace_id: str) -> List[StudyMetadata]:
        try:
            conn = self._connect()
            rows = conn.cursor().execute("SELECT * FROM studies WHERE workspace_id = ? ORDER BY created_at ASC", (workspace_id,)).fetchall()
            conn.close()
            return [self._study_from_row(row) for row in rows]
        except Exception as exc:
            print(f"Error retrieving studies: {exc}")
            return []

    def save_analysis(self, analysis_id: str, study_id: str, workspace_id: str, name: str, description: str = "", metadata: Dict[str, Any] = None) -> bool:
        try:
            existing = self.get_analysis(analysis_id)
            now = self._now()
            conn = self._connect()
            conn.cursor().execute(
                """INSERT INTO analyses
                (analysis_id, study_id, workspace_id, name, description, created_at, updated_at, metadata)
                VALUES (?, ?, ?, ?, ?, ?, ?, ?)
                ON CONFLICT(analysis_id) DO UPDATE SET
                    study_id=excluded.study_id, workspace_id=excluded.workspace_id, name=excluded.name,
                    description=excluded.description, updated_at=excluded.updated_at, metadata=excluded.metadata""",
                (analysis_id, study_id, workspace_id, name, description, existing.created_at if existing else now, now, json.dumps(metadata or {})),
            )
            conn.commit()
            conn.close()
            return True
        except Exception as exc:
            print(f"Error saving analysis metadata: {exc}")
            return False

    def _analysis_from_row(self, row) -> AnalysisMetadata:
        return AnalysisMetadata(row["analysis_id"], row["study_id"], row["workspace_id"], row["name"] or "",
                                row["description"] or "", row["created_at"] or "", row["updated_at"] or "",
                                self._decode(row["metadata"], {}))

    def get_analysis(self, analysis_id: str) -> Optional[AnalysisMetadata]:
        try:
            conn = self._connect()
            row = conn.cursor().execute("SELECT * FROM analyses WHERE analysis_id = ?", (analysis_id,)).fetchone()
            conn.close()
            return self._analysis_from_row(row) if row else None
        except Exception as exc:
            print(f"Error retrieving analysis: {exc}")
            return None

    def get_analyses_for_study(self, study_id: str) -> List[AnalysisMetadata]:
        try:
            conn = self._connect()
            rows = conn.cursor().execute("SELECT * FROM analyses WHERE study_id = ? ORDER BY created_at ASC", (study_id,)).fetchall()
            conn.close()
            return [self._analysis_from_row(row) for row in rows]
        except Exception as exc:
            print(f"Error retrieving analyses: {exc}")
            return []

    def save_task(self, task_id: str, analysis_id: str, workspace_id: str, study_id: str, name: str, engine_type: str, model_path: str = "", input_tables: List[str] = None, result_tables: List[str] = None, parent_task_id: str = "", metadata: Dict[str, Any] = None, model_snapshot_id: str = None) -> bool:
        try:
            existing = self.get_task(task_id)
            now = self._now()
            # Preserve an existing snapshot id when the caller doesn't pass one
            # (snapshot linkage is set by the capture layer, not every save).
            if model_snapshot_id is None:
                model_snapshot_id = existing.model_snapshot_id if existing else ""
            conn = self._connect()
            conn.cursor().execute(
                """INSERT INTO tasks
                (task_id, analysis_id, workspace_id, study_id, parent_task_id, name, engine_type, model_path,
                 model_snapshot_id, input_tables, result_tables, created_at, updated_at, metadata)
                VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)
                ON CONFLICT(task_id) DO UPDATE SET
                    analysis_id=excluded.analysis_id, workspace_id=excluded.workspace_id, study_id=excluded.study_id,
                    parent_task_id=excluded.parent_task_id, name=excluded.name, engine_type=excluded.engine_type,
                    model_path=excluded.model_path, model_snapshot_id=excluded.model_snapshot_id,
                    input_tables=excluded.input_tables,
                    result_tables=excluded.result_tables, updated_at=excluded.updated_at, metadata=excluded.metadata""",
                (task_id, analysis_id, workspace_id, study_id, parent_task_id or None, name, engine_type, model_path,
                 model_snapshot_id or None, json.dumps(input_tables or []), json.dumps(result_tables or []),
                 existing.created_at if existing else now, now, json.dumps(metadata or {})),
            )
            conn.commit()
            conn.close()
            return True
        except Exception as exc:
            print(f"Error saving task metadata: {exc}")
            return False

    def _task_from_row(self, row) -> TaskMetadata:
        keys = row.keys() if hasattr(row, "keys") else []
        snapshot_id = row["model_snapshot_id"] if "model_snapshot_id" in keys and row["model_snapshot_id"] else ""
        return TaskMetadata(row["task_id"], row["analysis_id"], row["workspace_id"], row["study_id"],
                            row["parent_task_id"] or "", row["name"] or "", row["engine_type"] or "",
                            row["model_path"] or "", self._decode(row["input_tables"], []),
                            self._decode(row["result_tables"], []), row["created_at"] or "",
                            row["updated_at"] or "", self._decode(row["metadata"], {}), snapshot_id)

    def get_task(self, task_id: str) -> Optional[TaskMetadata]:
        try:
            conn = self._connect()
            row = conn.cursor().execute("SELECT * FROM tasks WHERE task_id = ?", (task_id,)).fetchone()
            conn.close()
            return self._task_from_row(row) if row else None
        except Exception as exc:
            print(f"Error retrieving task: {exc}")
            return None

    def get_tasks_for_analysis(self, analysis_id: str) -> List[TaskMetadata]:
        try:
            conn = self._connect()
            rows = conn.cursor().execute("SELECT * FROM tasks WHERE analysis_id = ? ORDER BY created_at ASC", (analysis_id,)).fetchall()
            conn.close()
            return [self._task_from_row(row) for row in rows]
        except Exception as exc:
            print(f"Error retrieving tasks: {exc}")
            return []

    def set_task_model_snapshot(self, task_id: str, model_snapshot_id: str) -> bool:
        """Point a task at the model snapshot used for its latest run (provenance)."""
        try:
            conn = self._connect()
            conn.cursor().execute(
                "UPDATE tasks SET model_snapshot_id = ?, updated_at = ? WHERE task_id = ?",
                (model_snapshot_id or None, self._now(), task_id),
            )
            conn.commit()
            conn.close()
            return True
        except Exception as exc:
            print(f"Error updating task model snapshot: {exc}")
            return False

    def save_result(
        self,
        result_id: str,
        task_id: str,
        analysis_id: str,
        workspace_id: str,
        study_id: str,
        layer_name: str = "",
        table_name: str = "",
        metadata: Dict[str, Any] = None,
    ) -> bool:
        try:
            existing = self.get_result(result_id)
            created_at = existing.created_at if existing else self._now()
            conn = self._connect()
            conn.cursor().execute(
                """INSERT INTO results
                (result_id, task_id, analysis_id, workspace_id, study_id, layer_name, table_name, created_at, metadata)
                VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?)
                ON CONFLICT(result_id) DO UPDATE SET
                    task_id=excluded.task_id, analysis_id=excluded.analysis_id,
                    workspace_id=excluded.workspace_id, study_id=excluded.study_id,
                    layer_name=excluded.layer_name, table_name=excluded.table_name,
                    metadata=excluded.metadata""",
                (
                    result_id,
                    task_id,
                    analysis_id,
                    workspace_id,
                    study_id,
                    layer_name,
                    table_name,
                    created_at,
                    json.dumps(metadata or {}),
                ),
            )
            conn.commit()
            conn.close()
            return True
        except Exception as exc:
            print(f"Error saving result metadata: {exc}")
            return False

    def _result_from_row(self, row) -> ResultMetadata:
        return ResultMetadata(
            row["result_id"],
            row["task_id"],
            row["analysis_id"],
            row["workspace_id"],
            row["study_id"],
            row["layer_name"] or "",
            row["table_name"] or "",
            row["created_at"] or "",
            self._decode(row["metadata"], {}),
        )

    def get_result(self, result_id: str) -> Optional[ResultMetadata]:
        try:
            conn = self._connect()
            row = conn.cursor().execute("SELECT * FROM results WHERE result_id = ?", (result_id,)).fetchone()
            conn.close()
            return self._result_from_row(row) if row else None
        except Exception as exc:
            print(f"Error retrieving result: {exc}")
            return None

    def get_results_for_task(self, task_id: str) -> List[ResultMetadata]:
        try:
            conn = self._connect()
            rows = conn.cursor().execute("SELECT * FROM results WHERE task_id = ? ORDER BY created_at ASC", (task_id,)).fetchall()
            conn.close()
            return [self._result_from_row(row) for row in rows]
        except Exception as exc:
            print(f"Error retrieving results for task: {exc}")
            return []

    def get_results_for_analysis(self, analysis_id: str) -> List[ResultMetadata]:
        try:
            conn = self._connect()
            rows = conn.cursor().execute("SELECT * FROM results WHERE analysis_id = ? ORDER BY created_at ASC", (analysis_id,)).fetchall()
            conn.close()
            return [self._result_from_row(row) for row in rows]
        except Exception as exc:
            print(f"Error retrieving results for analysis: {exc}")
            return []

    def delete_workspace(self, workspace_id: str) -> bool:
        return self._delete_by_id("workspaces", "workspace_id", workspace_id)

    def delete_study(self, study_id: str) -> bool:
        return self._delete_by_id("studies", "study_id", study_id)

    def delete_analysis(self, analysis_id: str) -> bool:
        return self._delete_by_id("analyses", "analysis_id", analysis_id)

    def delete_task(self, task_id: str) -> bool:
        return self._delete_by_id("tasks", "task_id", task_id)

    def delete_result(self, result_id: str) -> bool:
        return self._delete_by_id("results", "result_id", result_id)

    def _delete_by_id(self, table_name: str, id_column: str, value: str) -> bool:
        try:
            conn = self._connect()
            conn.cursor().execute(f"DELETE FROM {table_name} WHERE {id_column} = ?", (value,))
            conn.commit()
            conn.close()
            return True
        except Exception as exc:
            print(f"Error deleting metadata from {table_name}: {exc}")
            return False

    def save_field_mapping_preset(self, *, workspace_id: str = "", project_id: str = "", engine_type: str, model_identifier: str, input_table_name: str, payload: Dict[str, Any]) -> bool:
        workspace_id = workspace_id or project_id
        if not all([workspace_id, engine_type, model_identifier, input_table_name]):
            return False
        try:
            existing = self.get_field_mapping_preset(workspace_id=workspace_id, engine_type=engine_type, model_identifier=model_identifier, input_table_name=input_table_name)
            now = self._now()
            conn = self._connect()
            conn.cursor().execute(
                """INSERT INTO field_mapping_presets
                (preset_id, workspace_id, engine_type, model_identifier, input_table_name, payload, created_at, updated_at)
                VALUES (?, ?, ?, ?, ?, ?, ?, ?)
                ON CONFLICT(workspace_id, engine_type, model_identifier, input_table_name) DO UPDATE SET
                    payload=excluded.payload, updated_at=excluded.updated_at""",
                (existing.get("preset_id") if existing else str(uuid.uuid4()), workspace_id, engine_type,
                 str(model_identifier).strip(), str(input_table_name).strip(), json.dumps(payload or {}),
                 existing.get("created_at") if existing else now, now),
            )
            conn.commit()
            conn.close()
            return True
        except Exception as exc:
            print(f"Error saving field mapping preset: {exc}")
            return False

    def get_field_mapping_preset(self, *, workspace_id: str = "", project_id: str = "", engine_type: str, model_identifier: str, input_table_name: str) -> Dict[str, Any]:
        workspace_id = workspace_id or project_id
        try:
            conn = self._connect()
            row = conn.cursor().execute(
                """SELECT * FROM field_mapping_presets
                WHERE workspace_id = ? AND engine_type = ? AND model_identifier = ? AND input_table_name = ?""",
                (workspace_id, engine_type, model_identifier, input_table_name),
            ).fetchone()
            conn.close()
            if not row:
                return {}
            return {"preset_id": row["preset_id"], "workspace_id": row["workspace_id"], "project_id": row["workspace_id"],
                    "engine_type": row["engine_type"], "model_identifier": row["model_identifier"],
                    "input_table_name": row["input_table_name"], "payload": self._decode(row["payload"], {}),
                    "created_at": row["created_at"], "updated_at": row["updated_at"]}
        except Exception as exc:
            print(f"Error retrieving field mapping preset: {exc}")
            return {}

    def save_workflow_definition(self, project_id: str, workflow_definition_id: str, name: str, description: str = "", payload: Dict[str, Any] = None) -> bool:
        try:
            existing = self.get_workflow_definition(project_id, workflow_definition_id)
            now = self._now()
            conn = self._connect()
            conn.cursor().execute(
                """INSERT INTO workflow_definitions
                (workflow_definition_id, workspace_id, name, description, payload, created_at, updated_at)
                VALUES (?, ?, ?, ?, ?, ?, ?)
                ON CONFLICT(workflow_definition_id) DO UPDATE SET
                    workspace_id=excluded.workspace_id, name=excluded.name, description=excluded.description,
                    payload=excluded.payload, updated_at=excluded.updated_at""",
                (workflow_definition_id, project_id, name, description, json.dumps(payload or {}),
                 existing.created_at if existing else now, now),
            )
            conn.commit()
            conn.close()
            return True
        except Exception as exc:
            print(f"Error saving workflow definition metadata: {exc}")
            return False

    def get_workflow_definition(self, project_id: str, workflow_definition_id: str) -> Optional[WorkflowDefinitionMetadata]:
        try:
            conn = self._connect()
            row = conn.cursor().execute(
                """SELECT workflow_definition_id, workspace_id, name, description, payload, created_at, updated_at
                FROM workflow_definitions WHERE workspace_id = ? AND workflow_definition_id = ?""",
                (project_id, workflow_definition_id),
            ).fetchone()
            conn.close()
            if not row:
                return None
            return WorkflowDefinitionMetadata(row["workflow_definition_id"], row["workspace_id"], row["name"] or "",
                                              row["description"] or "", self._decode(row["payload"], {}),
                                              row["created_at"], row["updated_at"])
        except Exception as exc:
            print(f"Error retrieving workflow definition metadata: {exc}")
            return None

    def list_workflow_definitions_for_project(self, project_id: str) -> List[WorkflowDefinitionMetadata]:
        try:
            conn = self._connect()
            rows = conn.cursor().execute(
                """SELECT workflow_definition_id, workspace_id, name, description, payload, created_at, updated_at
                FROM workflow_definitions WHERE workspace_id = ?
                ORDER BY updated_at DESC, name COLLATE NOCASE ASC""",
                (project_id,),
            ).fetchall()
            conn.close()
            return [WorkflowDefinitionMetadata(row["workflow_definition_id"], row["workspace_id"], row["name"] or "",
                                               row["description"] or "", self._decode(row["payload"], {}),
                                               row["created_at"], row["updated_at"]) for row in rows]
        except Exception as exc:
            print(f"Error listing workflow definition metadata: {exc}")
            return []

    # ------------------------------------------------------------------
    # Provenance & model-storage API (docs/provenance/00)
    # ------------------------------------------------------------------

    def save_model_snapshot(
        self,
        *,
        model_kind: str,
        content_blob: bytes,
        snapshot_id: str = "",
        display_name: str = "",
        source_path: str = "",
        source_format: str = "",
        schema_version: str = "",
        authoring_tool: str = "",
        authoring_tool_version: str = "",
        engine_version: str = "",
        model_summary: Dict[str, Any] = None,
        canonical_blob: Optional[bytes] = None,
        registry_id: str = "",
        registry_version: str = "",
        file_mtime: str = "",
        created_by: str = "",
        reuse_by_sha: bool = True,
    ) -> Optional[str]:
        """Insert an immutable model snapshot; returns its id (or None on error).

        When ``reuse_by_sha`` is set (default), an existing snapshot with the
        same ``content_sha256`` is reused instead of inserting a duplicate.
        """
        try:
            content_blob = content_blob or b""
            content_sha = self._sha256_bytes(content_blob)
            if reuse_by_sha:
                existing = self.get_model_snapshot_by_sha(content_sha)
                if existing is not None:
                    return existing.id
            snapshot_id = snapshot_id or str(uuid.uuid4())
            canonical_sha = self._sha256_bytes(canonical_blob) if canonical_blob else ""
            conn = self._connect()
            conn.execute(
                """INSERT INTO model_snapshots
                (id, model_kind, display_name, source_path, source_format, content_sha256,
                 content_blob, canonical_blob, canonical_sha256, byte_size, schema_version,
                 authoring_tool, authoring_tool_version, engine_version, model_summary_json,
                 registry_id, registry_version, file_mtime, created_at, created_by)
                VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)""",
                (
                    snapshot_id, model_kind, display_name, source_path, source_format, content_sha,
                    sqlite3.Binary(content_blob),
                    sqlite3.Binary(canonical_blob) if canonical_blob else None,
                    canonical_sha, len(content_blob), schema_version, authoring_tool,
                    authoring_tool_version, engine_version, json.dumps(model_summary or {}),
                    registry_id, registry_version, file_mtime, self._now(), created_by,
                ),
            )
            conn.commit()
            conn.close()
            return snapshot_id
        except Exception as exc:
            print(f"Error saving model snapshot: {exc}")
            return None

    def _model_snapshot_from_row(self, row) -> ModelSnapshotMetadata:
        return ModelSnapshotMetadata(
            row["id"], row["model_kind"] or "", row["display_name"] or "", row["source_path"] or "",
            row["source_format"] or "", row["content_sha256"] or "", row["byte_size"] or 0,
            row["schema_version"] or "", row["authoring_tool"] or "", row["authoring_tool_version"] or "",
            row["engine_version"] or "", self._decode(row["model_summary_json"], {}), row["registry_id"] or "",
            row["registry_version"] or "", row["file_mtime"] or "", row["created_at"] or "",
            row["created_by"] or "", row["canonical_sha256"] or "",
        )

    def get_model_snapshot(self, snapshot_id: str) -> Optional[ModelSnapshotMetadata]:
        try:
            conn = self._connect()
            row = conn.execute(
                """SELECT id, model_kind, display_name, source_path, source_format, content_sha256,
                   canonical_sha256, byte_size, schema_version, authoring_tool, authoring_tool_version,
                   engine_version, model_summary_json, registry_id, registry_version, file_mtime,
                   created_at, created_by FROM model_snapshots WHERE id = ?""",
                (snapshot_id,),
            ).fetchone()
            conn.close()
            return self._model_snapshot_from_row(row) if row else None
        except Exception as exc:
            print(f"Error retrieving model snapshot: {exc}")
            return None

    def get_model_snapshot_by_sha(self, content_sha256: str) -> Optional[ModelSnapshotMetadata]:
        try:
            conn = self._connect()
            row = conn.execute(
                """SELECT id, model_kind, display_name, source_path, source_format, content_sha256,
                   canonical_sha256, byte_size, schema_version, authoring_tool, authoring_tool_version,
                   engine_version, model_summary_json, registry_id, registry_version, file_mtime,
                   created_at, created_by FROM model_snapshots WHERE content_sha256 = ? LIMIT 1""",
                (content_sha256,),
            ).fetchone()
            conn.close()
            return self._model_snapshot_from_row(row) if row else None
        except Exception as exc:
            print(f"Error retrieving model snapshot by sha: {exc}")
            return None

    def get_model_snapshot_blob(self, snapshot_id: str, *, canonical: bool = False) -> Optional[bytes]:
        column = "canonical_blob" if canonical else "content_blob"
        try:
            conn = self._connect()
            row = conn.execute(f"SELECT {column} AS blob FROM model_snapshots WHERE id = ?", (snapshot_id,)).fetchone()
            conn.close()
            if not row or row["blob"] is None:
                return None
            return bytes(row["blob"])
        except Exception as exc:
            print(f"Error retrieving model snapshot blob: {exc}")
            return None

    def list_model_snapshots(self, model_kind: str = "") -> List[ModelSnapshotMetadata]:
        try:
            conn = self._connect()
            base = (
                """SELECT id, model_kind, display_name, source_path, source_format, content_sha256,
                   canonical_sha256, byte_size, schema_version, authoring_tool, authoring_tool_version,
                   engine_version, model_summary_json, registry_id, registry_version, file_mtime,
                   created_at, created_by FROM model_snapshots"""
            )
            if model_kind:
                rows = conn.execute(base + " WHERE model_kind = ? ORDER BY created_at DESC", (model_kind,)).fetchall()
            else:
                rows = conn.execute(base + " ORDER BY created_at DESC").fetchall()
            conn.close()
            return [self._model_snapshot_from_row(row) for row in rows]
        except Exception as exc:
            print(f"Error listing model snapshots: {exc}")
            return []

    # ---- workflow_definition_revisions ----

    def add_workflow_definition_revision(
        self,
        workflow_definition_id: str,
        payload: Dict[str, Any],
        *,
        change_summary: str = "",
        created_by: str = "",
    ) -> Optional[WorkflowDefinitionRevisionMetadata]:
        """Append a revision, deduped on payload_sha256.

        A no-op save (payload identical to the latest revision) returns the
        latest revision unchanged instead of creating a new row.
        """
        try:
            payload = payload or {}
            payload_sha = self._sha256_text(self._canonical_json(payload))
            latest = self.get_latest_workflow_definition_revision(workflow_definition_id)
            if latest is not None and latest.payload_sha256 == payload_sha:
                return latest
            revision_number = (latest.revision_number + 1) if latest else 1
            revision_id = str(uuid.uuid4())
            now = self._now()
            conn = self._connect()
            conn.execute(
                """INSERT INTO workflow_definition_revisions
                (id, workflow_definition_id, revision_number, payload, payload_sha256,
                 change_summary, created_at, created_by)
                VALUES (?, ?, ?, ?, ?, ?, ?, ?)""",
                (revision_id, workflow_definition_id, revision_number, json.dumps(payload),
                 payload_sha, change_summary, now, created_by),
            )
            conn.commit()
            conn.close()
            return WorkflowDefinitionRevisionMetadata(
                revision_id, workflow_definition_id, revision_number, payload, payload_sha,
                change_summary, now, created_by,
            )
        except Exception as exc:
            print(f"Error adding workflow definition revision: {exc}")
            return None

    def _revision_from_row(self, row) -> WorkflowDefinitionRevisionMetadata:
        return WorkflowDefinitionRevisionMetadata(
            row["id"], row["workflow_definition_id"], row["revision_number"] or 0,
            self._decode(row["payload"], {}), row["payload_sha256"] or "",
            row["change_summary"] or "", row["created_at"] or "", row["created_by"] or "",
        )

    def get_latest_workflow_definition_revision(self, workflow_definition_id: str) -> Optional[WorkflowDefinitionRevisionMetadata]:
        try:
            conn = self._connect()
            row = conn.execute(
                """SELECT * FROM workflow_definition_revisions
                WHERE workflow_definition_id = ? ORDER BY revision_number DESC LIMIT 1""",
                (workflow_definition_id,),
            ).fetchone()
            conn.close()
            return self._revision_from_row(row) if row else None
        except Exception as exc:
            print(f"Error retrieving latest workflow definition revision: {exc}")
            return None

    def list_workflow_definition_revisions(self, workflow_definition_id: str) -> List[WorkflowDefinitionRevisionMetadata]:
        try:
            conn = self._connect()
            rows = conn.execute(
                """SELECT * FROM workflow_definition_revisions
                WHERE workflow_definition_id = ? ORDER BY revision_number ASC""",
                (workflow_definition_id,),
            ).fetchall()
            conn.close()
            return [self._revision_from_row(row) for row in rows]
        except Exception as exc:
            print(f"Error listing workflow definition revisions: {exc}")
            return []

    # ---- workflow_runs ----

    def save_workflow_run(
        self,
        *,
        run_id: str = "",
        workspace_id: str = "",
        study_id: str = "",
        analysis_id: str = "",
        workflow_definition_id: str = "",
        workflow_definition_revision_id: str = "",
        run_kind: str = "baseline",
        scenario_id: str = "",
        parent_run_id: str = "",
        label: str = "",
        notes: str = "",
        status: str = "",
        start_time: str = "",
        end_time: str = "",
        duration_ms: int = 0,
        emds_version: str = "",
        gis_environment: str = "QGIS",
        gis_version: str = "",
        created_by: str = "",
    ) -> Optional[str]:
        """Upsert a run header; returns the run id (or None on error)."""
        try:
            run_id = run_id or str(uuid.uuid4())
            existing = self.get_workflow_run(run_id)
            created_at = existing.created_at if existing else self._now()
            conn = self._connect()
            conn.execute(
                """INSERT INTO workflow_runs
                (id, workspace_id, study_id, analysis_id, workflow_definition_id,
                 workflow_definition_revision_id, run_kind, scenario_id, parent_run_id, label, notes,
                 status, start_time, end_time, duration_ms, emds_version, gis_environment, gis_version,
                 created_by, created_at)
                VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)
                ON CONFLICT(id) DO UPDATE SET
                    workspace_id=excluded.workspace_id, study_id=excluded.study_id,
                    analysis_id=excluded.analysis_id, workflow_definition_id=excluded.workflow_definition_id,
                    workflow_definition_revision_id=excluded.workflow_definition_revision_id,
                    run_kind=excluded.run_kind, scenario_id=excluded.scenario_id,
                    parent_run_id=excluded.parent_run_id, label=excluded.label, notes=excluded.notes,
                    status=excluded.status, start_time=excluded.start_time, end_time=excluded.end_time,
                    duration_ms=excluded.duration_ms, emds_version=excluded.emds_version,
                    gis_environment=excluded.gis_environment, gis_version=excluded.gis_version,
                    created_by=excluded.created_by""",
                (run_id, workspace_id or None, study_id or None, analysis_id or None, workflow_definition_id or None,
                 workflow_definition_revision_id or None, run_kind, scenario_id or None, parent_run_id or None, label, notes,
                 status, start_time, end_time, duration_ms, emds_version, gis_environment, gis_version,
                 created_by, created_at),
            )
            conn.commit()
            conn.close()
            return run_id
        except Exception as exc:
            print(f"Error saving workflow run: {exc}")
            return None

    def _workflow_run_from_row(self, row) -> WorkflowRunMetadata:
        return WorkflowRunMetadata(
            row["id"], row["workspace_id"] or "", row["study_id"] or "", row["analysis_id"] or "",
            row["workflow_definition_id"] or "", row["workflow_definition_revision_id"] or "",
            row["run_kind"] or "baseline", row["scenario_id"] or "", row["parent_run_id"] or "",
            row["label"] or "", row["notes"] or "", row["status"] or "", row["start_time"] or "",
            row["end_time"] or "", row["duration_ms"] or 0, row["emds_version"] or "",
            row["gis_environment"] or "", row["gis_version"] or "", row["created_by"] or "",
            row["created_at"] or "",
        )

    def get_workflow_run(self, run_id: str) -> Optional[WorkflowRunMetadata]:
        try:
            conn = self._connect()
            row = conn.execute("SELECT * FROM workflow_runs WHERE id = ?", (run_id,)).fetchone()
            conn.close()
            return self._workflow_run_from_row(row) if row else None
        except Exception as exc:
            print(f"Error retrieving workflow run: {exc}")
            return None

    def get_workflow_runs_for_analysis(self, analysis_id: str) -> List[WorkflowRunMetadata]:
        try:
            conn = self._connect()
            rows = conn.execute(
                "SELECT * FROM workflow_runs WHERE analysis_id = ? ORDER BY created_at DESC", (analysis_id,)
            ).fetchall()
            conn.close()
            return [self._workflow_run_from_row(row) for row in rows]
        except Exception as exc:
            print(f"Error retrieving workflow runs for analysis: {exc}")
            return []

    def get_workflow_runs_for_scenario(self, scenario_id: str) -> List[WorkflowRunMetadata]:
        try:
            conn = self._connect()
            rows = conn.execute(
                "SELECT * FROM workflow_runs WHERE scenario_id = ? ORDER BY created_at ASC", (scenario_id,)
            ).fetchall()
            conn.close()
            return [self._workflow_run_from_row(row) for row in rows]
        except Exception as exc:
            print(f"Error retrieving workflow runs for scenario: {exc}")
            return []

    # ---- workflow_step_runs ----

    def save_workflow_step_run(
        self,
        *,
        run_id: str,
        step_run_id: str = "",
        workflow_step_id: str = "",
        task_id: str = "",
        parent_step_run_id: str = "",
        step_index: int = 0,
        name: str = "",
        engine_type: str = "",
        engine_version: str = "",
        model_snapshot_id: str = "",
        parameters: Dict[str, Any] = None,
        field_mappings: List[Any] = None,
        input_refs: List[Any] = None,
        output_summary: Dict[str, Any] = None,
        result_id: str = "",
        status: str = "",
        start_time: str = "",
        end_time: str = "",
        duration_ms: int = 0,
        rows_processed: int = 0,
        error_message: str = "",
    ) -> Optional[str]:
        """Insert a per-step provenance manifest; returns the step-run id."""
        try:
            step_run_id = step_run_id or str(uuid.uuid4())
            conn = self._connect()
            conn.execute(
                """INSERT INTO workflow_step_runs
                (id, run_id, workflow_step_id, task_id, parent_step_run_id, step_index, name,
                 engine_type, engine_version, model_snapshot_id, parameters_json, field_mappings_json,
                 input_refs_json, output_summary_json, result_id, status, start_time, end_time,
                 duration_ms, rows_processed, error_message, created_at)
                VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)""",
                (step_run_id, run_id, workflow_step_id, task_id or None, parent_step_run_id or None, step_index, name,
                 engine_type, engine_version, model_snapshot_id or None, json.dumps(parameters or {}),
                 json.dumps(field_mappings or []), json.dumps(input_refs or []),
                 json.dumps(output_summary or {}), result_id or None, status, start_time, end_time,
                 duration_ms, rows_processed, error_message, self._now()),
            )
            conn.commit()
            conn.close()
            return step_run_id
        except Exception as exc:
            print(f"Error saving workflow step run: {exc}")
            return None

    def _step_run_from_row(self, row) -> WorkflowStepRunMetadata:
        return WorkflowStepRunMetadata(
            row["id"], row["run_id"], row["workflow_step_id"] or "", row["task_id"] or "",
            row["parent_step_run_id"] or "", row["step_index"] or 0, row["name"] or "",
            row["engine_type"] or "", row["engine_version"] or "", row["model_snapshot_id"] or "",
            self._decode(row["parameters_json"], {}), self._decode(row["field_mappings_json"], []),
            self._decode(row["input_refs_json"], []), self._decode(row["output_summary_json"], {}),
            row["result_id"] or "", row["status"] or "", row["start_time"] or "", row["end_time"] or "",
            row["duration_ms"] or 0, row["rows_processed"] or 0, row["error_message"] or "",
            row["created_at"] or "",
        )

    def get_workflow_step_run(self, step_run_id: str) -> Optional[WorkflowStepRunMetadata]:
        try:
            conn = self._connect()
            row = conn.execute("SELECT * FROM workflow_step_runs WHERE id = ?", (step_run_id,)).fetchone()
            conn.close()
            return self._step_run_from_row(row) if row else None
        except Exception as exc:
            print(f"Error retrieving workflow step run: {exc}")
            return None

    def get_step_runs_for_run(self, run_id: str) -> List[WorkflowStepRunMetadata]:
        try:
            conn = self._connect()
            rows = conn.execute(
                "SELECT * FROM workflow_step_runs WHERE run_id = ? ORDER BY step_index ASC, created_at ASC", (run_id,)
            ).fetchall()
            conn.close()
            return [self._step_run_from_row(row) for row in rows]
        except Exception as exc:
            print(f"Error retrieving step runs for run: {exc}")
            return []

    def get_step_runs_for_task(self, task_id: str) -> List[WorkflowStepRunMetadata]:
        """All step-runs for a task, newest first (drives 'last run' / stale checks)."""
        try:
            conn = self._connect()
            rows = conn.execute(
                "SELECT * FROM workflow_step_runs WHERE task_id = ? ORDER BY created_at DESC", (task_id,)
            ).fetchall()
            conn.close()
            return [self._step_run_from_row(row) for row in rows]
        except Exception as exc:
            print(f"Error retrieving step runs for task: {exc}")
            return []

    def get_step_run_for_result(self, result_id: str) -> Optional[WorkflowStepRunMetadata]:
        try:
            conn = self._connect()
            row = conn.execute(
                "SELECT * FROM workflow_step_runs WHERE result_id = ? ORDER BY created_at DESC LIMIT 1", (result_id,)
            ).fetchone()
            conn.close()
            return self._step_run_from_row(row) if row else None
        except Exception as exc:
            print(f"Error retrieving step run for result: {exc}")
            return None

    # ---- result_lineage ----

    def save_result_lineage(
        self,
        *,
        result_id: str,
        lineage_id: str = "",
        step_run_id: str = "",
        run_id: str = "",
        model_snapshot_id: str = "",
        upstream_result_id: str = "",
        derived_from_result_id: str = "",
    ) -> Optional[str]:
        try:
            lineage_id = lineage_id or str(uuid.uuid4())
            conn = self._connect()
            conn.execute(
                """INSERT INTO result_lineage
                (id, result_id, step_run_id, run_id, model_snapshot_id, upstream_result_id,
                 derived_from_result_id, created_at)
                VALUES (?, ?, ?, ?, ?, ?, ?, ?)""",
                (lineage_id, result_id, step_run_id or None, run_id or None, model_snapshot_id or None, upstream_result_id or None,
                 derived_from_result_id or None, self._now()),
            )
            conn.commit()
            conn.close()
            return lineage_id
        except Exception as exc:
            print(f"Error saving result lineage: {exc}")
            return None

    def _lineage_from_row(self, row) -> ResultLineageMetadata:
        return ResultLineageMetadata(
            row["id"], row["result_id"], row["step_run_id"] or "", row["run_id"] or "",
            row["model_snapshot_id"] or "", row["upstream_result_id"] or "",
            row["derived_from_result_id"] or "", row["created_at"] or "",
        )

    def get_result_lineage(self, result_id: str) -> List[ResultLineageMetadata]:
        try:
            conn = self._connect()
            rows = conn.execute(
                "SELECT * FROM result_lineage WHERE result_id = ? ORDER BY created_at ASC", (result_id,)
            ).fetchall()
            conn.close()
            return [self._lineage_from_row(row) for row in rows]
        except Exception as exc:
            print(f"Error retrieving result lineage: {exc}")
            return []

    # Compatibility method names used by existing dock code during the rename.
    def save_project(self, project_id: str, name: str, description: str, project_file: str, input_layers: List[str], metadata: Dict[str, Any] = None) -> bool:
        return self.save_workspace(project_id, name, description, project_file, input_layers, metadata)

    def get_project(self, project_id: str) -> Optional[WorkspaceMetadata]:
        return self.get_workspace(project_id)

    def get_project_by_file(self, project_file: str) -> Optional[WorkspaceMetadata]:
        return self.get_workspace_by_qgis_project_file(project_file)

    def get_all_projects(self) -> List[WorkspaceMetadata]:
        return self.get_all_workspaces()

    def update_project(self, project_id: str, name: str = None, description: str = None, metadata: Dict[str, Any] = None) -> bool:
        return self.update_workspace(project_id, name=name, description=description, metadata=metadata)

    def delete_project(self, project_id: str) -> bool:
        return self.delete_workspace(project_id)

    def project_exists(self, project_id: str) -> bool:
        return self.get_workspace(project_id) is not None

    def get_project_count(self) -> int:
        return len(self.get_all_workspaces())

    def save_spatial_reference(self, spatial_reference_id: str, project_id: str, name: str, comments: str, output_layer_name: str, reference_layer_name: str = "", metadata: Dict[str, Any] = None) -> bool:
        return self.save_study(spatial_reference_id, project_id, name, comments, output_layer_name, reference_layer_name, metadata)

    def get_spatial_reference(self, spatial_reference_id: str) -> Optional[StudyMetadata]:
        return self.get_study(spatial_reference_id)

    def get_spatial_references_for_project(self, project_id: str) -> List[StudyMetadata]:
        return self.get_studies_for_workspace(project_id)

    def delete_spatial_reference(self, spatial_reference_id: str) -> bool:
        return self.delete_study(spatial_reference_id)

    def save_provenance(self, provenance_id: str, spatial_reference_id: str, project_id: str, name: str, description: str = "", metadata: Dict[str, Any] = None) -> bool:
        return self.save_analysis(provenance_id, spatial_reference_id, project_id, name, description, metadata)

    def get_provenance(self, provenance_id: str) -> Optional[AnalysisMetadata]:
        return self.get_analysis(provenance_id)

    def get_provenances_for_spatial_reference(self, spatial_reference_id: str) -> List[AnalysisMetadata]:
        return self.get_analyses_for_study(spatial_reference_id)

    def delete_provenance(self, provenance_id: str) -> bool:
        return self.delete_analysis(provenance_id)

    def save_task_detail(self, task_id: str, provenance_id: str, project_id: str, spatial_reference_id: str, name: str, engine_type: str, model_path: str = "", input_tables: List[str] = None, result_tables: List[str] = None, parent_task_id: str = "", metadata: Dict[str, Any] = None) -> bool:
        return self.save_task(task_id, provenance_id, project_id, spatial_reference_id, name, engine_type, model_path, input_tables, result_tables, parent_task_id, metadata)

    def get_task_detail(self, task_id: str) -> Optional[TaskMetadata]:
        return self.get_task(task_id)

    def get_task_details_for_provenance(self, provenance_id: str) -> List[TaskMetadata]:
        return self.get_tasks_for_analysis(provenance_id)

    def delete_task_detail(self, task_id: str) -> bool:
        return self.delete_task(task_id)

    def save_results(self, results_id: str, workflow_id: str, provenance_id: str, project_id: str, spatial_reference_id: str, layer_name: str = "", table_name: str = "", metadata: Dict[str, Any] = None) -> bool:
        return self.save_result(results_id, workflow_id, provenance_id, project_id, spatial_reference_id, layer_name, table_name, metadata)

    def get_results(self, results_id: str) -> Optional[ResultMetadata]:
        return self.get_result(results_id)

    def get_results_for_workflow(self, workflow_id: str) -> List[ResultMetadata]:
        return self.get_results_for_task(workflow_id)
