MusicWorkshop/backend/app/task_store.py

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

from .task_constants import (
  ACTIVE_TASK_STATUSES,
  SCAN_PROGRESS_LOG_LIMIT,
  STAGE_STATUS_FAILED,
  TASK_STATUS_COMPLETED,
  TASK_STATUS_FAILED,
  TASK_STATUS_PENDING,
  TASK_TYPE_INGEST,
  current_timestamp,
  create_empty_dedupe_stats,
  create_empty_match_stats,
  create_empty_organize_stats,
  create_empty_preprocess_stats,
  create_empty_repair_stats,
  create_empty_scan_stats,
  create_empty_task_stats,
  create_pending_repair_stage_states,
  create_pending_stage_states
)


class TaskConflictError(Exception):
  def __init__(self, active_task_id: str):
    super().__init__('Task already running')
    self.active_task_id = active_task_id


class TaskNotFoundError(Exception):
  pass


class TaskStore:
  def __init__(self, db_path: Path):
    self.db_path = Path(db_path)
    self.db_path.parent.mkdir(parents=True, exist_ok=True)
    self._initialize()

  def _connect(self):
    connection = sqlite3.connect(self.db_path, check_same_thread=False)
    connection.row_factory = sqlite3.Row
    connection.execute('PRAGMA foreign_keys = ON')
    return connection

  def _initialize(self):
    with self._connect() as connection:
      connection.executescript(
        '''
        CREATE TABLE IF NOT EXISTS task_runs (
          id TEXT PRIMARY KEY,
          task_type TEXT NOT NULL DEFAULT 'ingest',
          trigger_source TEXT NOT NULL,
          source_task_id TEXT,
          status TEXT NOT NULL,
          current_stage TEXT NOT NULL,
          stage_states_json TEXT NOT NULL,
          config_snapshot_json TEXT NOT NULL,
          stats_json TEXT NOT NULL,
          repair_plan_json TEXT,
          error_message TEXT,
          started_at TEXT NOT NULL,
          completed_at TEXT,
          updated_at TEXT NOT NULL
        );

        CREATE INDEX IF NOT EXISTS idx_task_runs_status_updated
        ON task_runs (status, updated_at DESC);

        CREATE TABLE IF NOT EXISTS task_items (
          id INTEGER PRIMARY KEY AUTOINCREMENT,
          task_id TEXT NOT NULL,
          original_path TEXT NOT NULL,
          relative_path TEXT NOT NULL,
          filename TEXT NOT NULL,
          extension TEXT NOT NULL,
          size_bytes INTEGER,
          modified_at TEXT,
          local_cover TEXT,
          local_lyric TEXT,
          scan_status TEXT NOT NULL,
          scan_reason TEXT,
          scan_message TEXT,
          created_at TEXT NOT NULL,
          updated_at TEXT NOT NULL,
          FOREIGN KEY (task_id) REFERENCES task_runs(id) ON DELETE CASCADE
        );

        CREATE TABLE IF NOT EXISTS task_logs (
          id INTEGER PRIMARY KEY AUTOINCREMENT,
          task_id TEXT NOT NULL,
          stage TEXT NOT NULL,
          level TEXT NOT NULL,
          event_type TEXT NOT NULL,
          message TEXT NOT NULL,
          payload_json TEXT,
          created_at TEXT NOT NULL,
          FOREIGN KEY (task_id) REFERENCES task_runs(id) ON DELETE CASCADE
        );
        '''
      )
      self._ensure_task_items_columns(connection)
      self._ensure_indexes(connection)
      connection.commit()

  def _ensure_task_items_columns(self, connection: sqlite3.Connection):
    existing_columns = {
      row['name']
      for row in connection.execute("PRAGMA table_info('task_items')").fetchall()
    }

    column_specs = {
      'parent_item_id': 'INTEGER',
      'is_active': 'INTEGER NOT NULL DEFAULT 1',
      'current_file_path': 'TEXT',
      'preprocess_status': "TEXT NOT NULL DEFAULT 'pending'",
      'preprocess_reason': 'TEXT',
      'preprocess_message': 'TEXT',
      'audio_props_json': 'TEXT',
      'original_tags_json': 'TEXT',
      'preprocess_artifacts_json': 'TEXT',
      'acoustic_fingerprint': 'TEXT',
      'fingerprint_duration_seconds': 'REAL',
      'match_status': "TEXT NOT NULL DEFAULT 'pending'",
      'match_reason': 'TEXT',
      'match_message': 'TEXT',
      'match_source': 'TEXT',
      'match_confidence': 'REAL',
      'match_is_authoritative': 'INTEGER NOT NULL DEFAULT 0',
      'matched_metadata_json': 'TEXT',
      'match_candidates_json': 'TEXT',
      'match_enrichment_json': 'TEXT',
      'dedupe_status': "TEXT NOT NULL DEFAULT 'pending'",
      'dedupe_reason': 'TEXT',
      'dedupe_message': 'TEXT',
      'dedupe_group_key': 'TEXT',
      'duplicate_of_path': 'TEXT',
      'duplicate_of_item_id': 'INTEGER',
      'dedupe_decision_json': 'TEXT',
      'organize_status': "TEXT NOT NULL DEFAULT 'pending'",
      'organize_reason': 'TEXT',
      'organize_message': 'TEXT',
      'library_relative_path': 'TEXT',
      'library_file_path': 'TEXT',
      'trash_file_path': 'TEXT',
      'organize_decision_json': 'TEXT',
      'exception_resolution_status': "TEXT NOT NULL DEFAULT 'open'",
      'exception_resolution_json': 'TEXT',
      'last_repair_task_id': 'TEXT'
    }

    task_run_columns = {
      row['name']
      for row in connection.execute("PRAGMA table_info('task_runs')").fetchall()
    }

    task_run_specs = {
      'task_type': "TEXT NOT NULL DEFAULT 'ingest'",
      'source_task_id': 'TEXT',
      'repair_plan_json': 'TEXT'
    }

    for column_name, column_sql in task_run_specs.items():
      if column_name not in task_run_columns:
        connection.execute(
          f'ALTER TABLE task_runs ADD COLUMN {column_name} {column_sql}'
        )

    for column_name, column_sql in column_specs.items():
      if column_name not in existing_columns:
        connection.execute(
          f'ALTER TABLE task_items ADD COLUMN {column_name} {column_sql}'
        )

    connection.execute(
      '''
      UPDATE task_items
      SET current_file_path = original_path
      WHERE current_file_path IS NULL
      '''
    )

  def _ensure_indexes(self, connection: sqlite3.Connection):
    connection.executescript(
      '''
      CREATE INDEX IF NOT EXISTS idx_task_items_task_created
      ON task_items (task_id, created_at ASC);

      CREATE INDEX IF NOT EXISTS idx_task_items_task_status
      ON task_items (task_id, scan_status);

      CREATE INDEX IF NOT EXISTS idx_task_items_task_preprocess
      ON task_items (task_id, preprocess_status, is_active);

      CREATE INDEX IF NOT EXISTS idx_task_items_task_match
      ON task_items (task_id, match_status, is_active);

      CREATE INDEX IF NOT EXISTS idx_task_items_task_dedupe
      ON task_items (task_id, dedupe_status, is_active);

      CREATE INDEX IF NOT EXISTS idx_task_items_task_organize
      ON task_items (task_id, organize_status, is_active);

      CREATE INDEX IF NOT EXISTS idx_task_logs_task_id
      ON task_logs (task_id, id ASC);
      '''
    )

  def fail_stale_active_tasks(self) -> list[str]:
    failed_task_ids: list[str] = []
    timestamp = current_timestamp()

    with self._connect() as connection:
      connection.isolation_level = None
      connection.execute('BEGIN IMMEDIATE')
      stale_rows = connection.execute(
        '''
        SELECT id, stage_states_json
        FROM task_runs
        WHERE status IN (?, ?)
        ''',
        ACTIVE_TASK_STATUSES
      ).fetchall()

      for row in stale_rows:
        stage_states = json.loads(row['stage_states_json'])
        repaired_stage_states = {
          stage_id: (
            STAGE_STATUS_FAILED
            if stage_status in ('pending', 'running')
            else stage_status
          )
          for stage_id, stage_status in stage_states.items()
        }
        connection.execute(
          '''
          UPDATE task_runs
          SET status = ?, stage_states_json = ?, error_message = ?, completed_at = ?, updated_at = ?
          WHERE id = ?
          ''',
          (
            TASK_STATUS_FAILED,
            json.dumps(repaired_stage_states),
            'Service restarted unexpectedly',
            timestamp,
            timestamp,
            row['id']
          )
        )
        connection.execute(
          '''
          INSERT INTO task_logs (task_id, stage, level, event_type, message, payload_json, created_at)
          VALUES (?, ?, ?, ?, ?, ?, ?)
          ''',
          (
            row['id'],
            'system',
            'error',
            'task.failed',
            'Service restarted unexpectedly',
            None,
            timestamp
          )
        )
        failed_task_ids.append(row['id'])

      connection.commit()

    return failed_task_ids

  def create_task_if_idle(
    self,
    config_snapshot: dict,
    trigger_source: str = 'manual',
    *,
    task_type: str = TASK_TYPE_INGEST,
    source_task_id: str | None = None,
    repair_plan_json: dict | None = None
  ) -> dict:
    task_id = str(uuid4())
    timestamp = current_timestamp()
    if task_type == TASK_TYPE_INGEST:
      stage_states = create_pending_stage_states()
      stats = create_empty_task_stats()
      initial_stage = 'scan'
    else:
      stage_states = create_pending_repair_stage_states()
      stats = create_empty_repair_stats()
      initial_stage = 'prepare'

    with self._connect() as connection:
      connection.isolation_level = None
      connection.execute('BEGIN IMMEDIATE')
      active_row = connection.execute(
        '''
        SELECT id
        FROM task_runs
        WHERE status IN (?, ?)
          AND task_type = ?
        ORDER BY started_at DESC
        LIMIT 1
        ''',
        (*ACTIVE_TASK_STATUSES, task_type)
      ).fetchone()

      if active_row is not None:
        connection.rollback()
        raise TaskConflictError(active_row['id'])

      connection.execute(
        '''
        INSERT INTO task_runs (
          id, task_type, trigger_source, source_task_id, status, current_stage, stage_states_json,
          config_snapshot_json, stats_json, repair_plan_json, error_message, started_at, completed_at, updated_at
        )
        VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)
        ''',
        (
          task_id,
          task_type,
          trigger_source,
          source_task_id,
          TASK_STATUS_PENDING,
          initial_stage,
          json.dumps(stage_states),
          json.dumps(config_snapshot),
          json.dumps(stats),
          json.dumps(repair_plan_json) if repair_plan_json is not None else None,
          None,
          timestamp,
          None,
          timestamp
        )
      )
      connection.commit()

    return self.get_task(task_id)

  def get_active_task(self, task_type: str = TASK_TYPE_INGEST) -> dict | None:
    with self._connect() as connection:
      row = connection.execute(
        '''
        SELECT *
        FROM task_runs
        WHERE status IN (?, ?)
          AND task_type = ?
        ORDER BY started_at DESC
        LIMIT 1
        ''',
        (*ACTIVE_TASK_STATUSES, task_type)
      ).fetchone()

    return self._parse_task(row) if row is not None else None

  def get_latest_task(self, task_type: str | None = TASK_TYPE_INGEST) -> dict | None:
    with self._connect() as connection:
      if task_type is None:
        row = connection.execute(
          '''
          SELECT *
          FROM task_runs
          ORDER BY started_at DESC
          LIMIT 1
          '''
        ).fetchone()
      else:
        row = connection.execute(
          '''
          SELECT *
          FROM task_runs
          WHERE task_type = ?
          ORDER BY started_at DESC
          LIMIT 1
          ''',
          (task_type,)
        ).fetchone()

    return self._parse_task(row) if row is not None else None

  def get_task(self, task_id: str) -> dict:
    with self._connect() as connection:
      row = connection.execute(
        'SELECT * FROM task_runs WHERE id = ?',
        (task_id,)
      ).fetchone()

    if row is None:
      raise TaskNotFoundError(task_id)

    return self._parse_task(row)

  def update_task(
    self,
    task_id: str,
    *,
    status: str | None = None,
    current_stage: str | None = None,
    stage_states: dict[str, str] | None = None,
    stats: dict | None = None,
    error_message: str | None = None,
    completed_at: str | None = None
  ) -> dict:
    assignments: list[str] = []
    values: list[object] = []

    if status is not None:
      assignments.append('status = ?')
      values.append(status)
    if current_stage is not None:
      assignments.append('current_stage = ?')
      values.append(current_stage)
    if stage_states is not None:
      assignments.append('stage_states_json = ?')
      values.append(json.dumps(stage_states))
    if stats is not None:
      assignments.append('stats_json = ?')
      values.append(json.dumps(stats))
    if error_message is not None or completed_at is not None:
      assignments.append('error_message = ?')
      values.append(error_message)
    if completed_at is not None:
      assignments.append('completed_at = ?')
      values.append(completed_at)

    assignments.append('updated_at = ?')
    values.append(current_timestamp())
    values.append(task_id)

    with self._connect() as connection:
      connection.execute(
        f'UPDATE task_runs SET {", ".join(assignments)} WHERE id = ?',
        values
      )
      connection.commit()

    return self.get_task(task_id)

  def append_log(
    self,
    task_id: str,
    stage: str,
    level: str,
    event_type: str,
    message: str,
    payload: dict | None = None
  ) -> dict:
    timestamp = current_timestamp()

    with self._connect() as connection:
      cursor = connection.execute(
        '''
        INSERT INTO task_logs (task_id, stage, level, event_type, message, payload_json, created_at)
        VALUES (?, ?, ?, ?, ?, ?, ?)
        ''',
        (
          task_id,
          stage,
          level,
          event_type,
          message,
          json.dumps(payload) if payload is not None else None,
          timestamp
        )
      )
      connection.commit()
      row = connection.execute(
        'SELECT * FROM task_logs WHERE id = ?',
        (cursor.lastrowid,)
      ).fetchone()

    return self._parse_log(row)

  def insert_task_item(
    self,
    task_id: str,
    *,
    original_path: str,
    relative_path: str,
    filename: str,
    extension: str,
    size_bytes: int | None,
    modified_at: str | None,
    local_cover: str | None,
    local_lyric: str | None,
    scan_status: str,
    scan_reason: str | None,
    scan_message: str | None,
    parent_item_id: int | None = None,
    is_active: int = 1,
    current_file_path: str | None = None,
    preprocess_status: str = 'pending',
    preprocess_reason: str | None = None,
    preprocess_message: str | None = None,
    audio_props_json: dict | None = None,
    original_tags_json: dict | None = None,
    preprocess_artifacts_json: dict | None = None,
    acoustic_fingerprint: str | None = None,
    fingerprint_duration_seconds: float | None = None,
    match_status: str = 'pending',
    match_reason: str | None = None,
    match_message: str | None = None,
    match_source: str | None = None,
    match_confidence: float | None = None,
    match_is_authoritative: int = 0,
    matched_metadata_json: dict | None = None,
    match_candidates_json: list[dict] | None = None,
    match_enrichment_json: dict | None = None,
    dedupe_status: str = 'pending',
    dedupe_reason: str | None = None,
    dedupe_message: str | None = None,
    dedupe_group_key: str | None = None,
    duplicate_of_path: str | None = None,
    duplicate_of_item_id: int | None = None,
    dedupe_decision_json: dict | None = None,
    organize_status: str = 'pending',
    organize_reason: str | None = None,
    organize_message: str | None = None,
    library_relative_path: str | None = None,
    library_file_path: str | None = None,
    trash_file_path: str | None = None,
    organize_decision_json: dict | None = None,
    exception_resolution_status: str = 'open',
    exception_resolution_json: dict | None = None,
    last_repair_task_id: str | None = None
  ) -> dict:
    timestamp = current_timestamp()
    current_file_path = current_file_path or original_path

    with self._connect() as connection:
      cursor = connection.execute(
        '''
        INSERT INTO task_items (
          task_id, parent_item_id, is_active, original_path, current_file_path, relative_path,
          filename, extension, size_bytes, modified_at, local_cover, local_lyric, scan_status,
          scan_reason, scan_message, preprocess_status, preprocess_reason, preprocess_message,
          audio_props_json, original_tags_json, preprocess_artifacts_json, acoustic_fingerprint,
          fingerprint_duration_seconds, match_status, match_reason, match_message, match_source,
          match_confidence, match_is_authoritative, matched_metadata_json, match_candidates_json,
          match_enrichment_json, dedupe_status, dedupe_reason, dedupe_message,
          dedupe_group_key, duplicate_of_path, duplicate_of_item_id, dedupe_decision_json,
          organize_status, organize_reason, organize_message, library_relative_path,
          library_file_path, trash_file_path, organize_decision_json,
          exception_resolution_status, exception_resolution_json, last_repair_task_id,
          created_at, updated_at
        )
        VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)
        ''',
        (
          task_id,
          parent_item_id,
          is_active,
          original_path,
          current_file_path,
          relative_path,
          filename,
          extension,
          size_bytes,
          modified_at,
          local_cover,
          local_lyric,
          scan_status,
          scan_reason,
          scan_message,
          preprocess_status,
          preprocess_reason,
          preprocess_message,
          json.dumps(audio_props_json) if audio_props_json is not None else None,
          json.dumps(original_tags_json) if original_tags_json is not None else None,
          json.dumps(preprocess_artifacts_json) if preprocess_artifacts_json is not None else None,
          acoustic_fingerprint,
          fingerprint_duration_seconds,
          match_status,
          match_reason,
          match_message,
          match_source,
          match_confidence,
          match_is_authoritative,
          json.dumps(matched_metadata_json) if matched_metadata_json is not None else None,
          json.dumps(match_candidates_json) if match_candidates_json is not None else None,
          json.dumps(match_enrichment_json) if match_enrichment_json is not None else None,
          dedupe_status,
          dedupe_reason,
          dedupe_message,
          dedupe_group_key,
          duplicate_of_path,
          duplicate_of_item_id,
          json.dumps(dedupe_decision_json) if dedupe_decision_json is not None else None,
          organize_status,
          organize_reason,
          organize_message,
          library_relative_path,
          library_file_path,
          trash_file_path,
          json.dumps(organize_decision_json) if organize_decision_json is not None else None,
          exception_resolution_status,
          json.dumps(exception_resolution_json) if exception_resolution_json is not None else None,
          last_repair_task_id,
          timestamp,
          timestamp
        )
      )
      connection.commit()
      row = connection.execute(
        'SELECT * FROM task_items WHERE id = ?',
        (cursor.lastrowid,)
      ).fetchone()

    return self._parse_item(row)

  def update_task_item(self, item_id: int, **fields) -> dict:
    json_fields = {
      'audio_props_json',
      'original_tags_json',
      'preprocess_artifacts_json',
      'matched_metadata_json',
      'match_candidates_json',
      'match_enrichment_json',
      'dedupe_decision_json',
      'organize_decision_json',
      'exception_resolution_json'
    }
    assignments: list[str] = []
    values: list[object] = []

    for key, value in fields.items():
      assignments.append(f'{key} = ?')
      if key in json_fields and value is not None:
        values.append(json.dumps(value))
      else:
        values.append(value)

    assignments.append('updated_at = ?')
    values.append(current_timestamp())
    values.append(item_id)

    with self._connect() as connection:
      connection.execute(
        f'UPDATE task_items SET {", ".join(assignments)} WHERE id = ?',
        values
      )
      connection.commit()
      row = connection.execute(
        'SELECT * FROM task_items WHERE id = ?',
        (item_id,)
      ).fetchone()

    if row is None:
      raise TaskNotFoundError(item_id)

    return self._parse_item(row)

  def list_preprocess_candidate_items(self, task_id: str) -> list[dict]:
    with self._connect() as connection:
      rows = connection.execute(
        '''
        SELECT *
        FROM task_items
        WHERE task_id = ?
          AND is_active = 1
          AND scan_status = 'queued'
          AND preprocess_status IN ('pending', 'running')
        ORDER BY created_at ASC, id ASC
        ''',
        (task_id,)
      ).fetchall()

    return [self._parse_item(row) for row in rows]

  def list_match_candidate_items(self, task_id: str) -> list[dict]:
    with self._connect() as connection:
      rows = connection.execute(
        '''
        SELECT *
        FROM task_items
        WHERE task_id = ?
          AND is_active = 1
          AND scan_status = 'queued'
          AND preprocess_status IN ('completed', 'warning')
          AND match_status IN ('pending', 'running')
        ORDER BY created_at ASC, id ASC
        ''',
        (task_id,)
      ).fetchall()

    return [self._parse_item(row) for row in rows]

  def list_dedupe_candidate_items(self, task_id: str) -> list[dict]:
    with self._connect() as connection:
      rows = connection.execute(
        '''
        SELECT *
        FROM task_items
        WHERE task_id = ?
          AND is_active = 1
          AND match_status IN ('matched', 'matched_fallback')
          AND dedupe_status IN ('pending', 'running')
        ORDER BY created_at ASC, id ASC
        ''',
        (task_id,)
      ).fetchall()

    return [self._parse_item(row) for row in rows]

  def list_organize_candidate_items(self, task_id: str) -> list[dict]:
    with self._connect() as connection:
      rows = connection.execute(
        '''
        SELECT *
        FROM task_items
        WHERE task_id = ?
          AND is_active = 1
          AND match_status IN ('matched', 'matched_fallback')
          AND dedupe_status IN ('unique', 'duplicate_replaced')
          AND organize_status IN ('pending', 'running')
        ORDER BY created_at ASC, id ASC
        ''',
        (task_id,)
      ).fetchall()

    return [self._parse_item(row) for row in rows]

  def list_task_items(
    self,
    task_id: str,
    scan_status: str | None,
    page: int,
    page_size: int,
    *,
    preprocess_status: str | None = None,
    match_status: str | None = None,
    dedupe_status: str | None = None,
    organize_status: str | None = None,
    active_only: bool = False
  ) -> dict:
    offset = (page - 1) * page_size
    where_clauses = ['task_id = ?']
    params: list[object] = [task_id]

    if scan_status:
      where_clauses.append('scan_status = ?')
      params.append(scan_status)
    if preprocess_status:
      where_clauses.append('preprocess_status = ?')
      params.append(preprocess_status)
    if match_status:
      where_clauses.append('match_status = ?')
      params.append(match_status)
    if dedupe_status:
      where_clauses.append('dedupe_status = ?')
      params.append(dedupe_status)
    if organize_status:
      where_clauses.append('organize_status = ?')
      params.append(organize_status)
    if active_only:
      where_clauses.append('is_active = 1')

    where_clause = f"WHERE {' AND '.join(where_clauses)}"

    with self._connect() as connection:
      total = connection.execute(
        f'SELECT COUNT(*) AS total FROM task_items {where_clause}',
        params
      ).fetchone()['total']
      rows = connection.execute(
        f'''
        SELECT *
        FROM task_items
        {where_clause}
        ORDER BY created_at ASC, id ASC
        LIMIT ? OFFSET ?
        ''',
        [*params, page_size, offset]
      ).fetchall()

    return {
      'items': [self._parse_item(row) for row in rows],
      'page': page,
      'page_size': page_size,
      'total': total
    }

  def list_all_task_items(self, task_id: str, *, active_only: bool = False) -> list[dict]:
    where_clauses = ['task_id = ?']
    params: list[object] = [task_id]
    if active_only:
      where_clauses.append('is_active = 1')

    with self._connect() as connection:
      rows = connection.execute(
        f'''
        SELECT *
        FROM task_items
        WHERE {' AND '.join(where_clauses)}
        ORDER BY created_at ASC, id ASC
        ''',
        params
      ).fetchall()

    return [self._parse_item(row) for row in rows]

  def list_task_history(self, page: int, page_size: int) -> dict:
    offset = (page - 1) * page_size

    with self._connect() as connection:
      total = connection.execute(
        '''
        SELECT COUNT(*) AS total
        FROM task_runs
        WHERE status IN (?, ?)
          AND task_type = 'ingest'
        ''',
        (TASK_STATUS_COMPLETED, TASK_STATUS_FAILED)
      ).fetchone()['total']
      rows = connection.execute(
        '''
        WITH task_item_stats AS (
          SELECT
            task_id,
            COUNT(*) AS total_items,
            SUM(CASE WHEN organize_status = 'organized' THEN 1 ELSE 0 END) AS success_items
          FROM task_items
          GROUP BY task_id
        )
        SELECT
          task_runs.id AS task_id,
          task_runs.started_at,
          task_runs.status,
          COALESCE(task_item_stats.total_items, 0) AS total_items,
          COALESCE(task_item_stats.success_items, 0) AS success_items
        FROM task_runs
        LEFT JOIN task_item_stats ON task_item_stats.task_id = task_runs.id
        WHERE task_runs.status IN (?, ?)
          AND task_runs.task_type = 'ingest'
        ORDER BY task_runs.started_at DESC, task_runs.id DESC
        LIMIT ? OFFSET ?
        ''',
        (TASK_STATUS_COMPLETED, TASK_STATUS_FAILED, page_size, offset)
      ).fetchall()

    return {
      'items': [self._parse_task_history_row(row) for row in rows],
      'page': page,
      'page_size': page_size,
      'total': total
    }

  def list_task_logs(self, task_id: str, page: int, page_size: int) -> dict:
    offset = (page - 1) * page_size

    with self._connect() as connection:
      total = connection.execute(
        'SELECT COUNT(*) AS total FROM task_logs WHERE task_id = ?',
        (task_id,)
      ).fetchone()['total']
      rows = connection.execute(
        '''
        SELECT *
        FROM task_logs
        WHERE task_id = ?
        ORDER BY id ASC
        LIMIT ? OFFSET ?
        ''',
        (task_id, page_size, offset)
      ).fetchall()

    return {
      'logs': [self._parse_log(row) for row in rows],
      'page': page,
      'page_size': page_size,
      'total': total
    }

  def list_library_provenance_items(self) -> list[dict]:
    with self._connect() as connection:
      rows = connection.execute(
        '''
        SELECT
          task_id,
          library_file_path,
          library_relative_path,
          updated_at AS organized_at,
          match_source,
          match_confidence,
          dedupe_status
        FROM task_items
        WHERE organize_status = 'organized'
        ORDER BY updated_at DESC, id DESC
        '''
      ).fetchall()

    return [
      {
        'task_id': row['task_id'],
        'library_file_path': row['library_file_path'],
        'library_relative_path': row['library_relative_path'],
        'organized_at': row['organized_at'],
        'match_source': row['match_source'],
        'match_confidence': row['match_confidence'],
        'dedupe_status': row['dedupe_status']
      }
      for row in rows
    ]

  def list_exception_source_items(self, resolution_status: str = 'open') -> list[dict]:
    if resolution_status not in {'open', 'resolved', 'ignored', 'all', 'planned'}:
      raise ValueError(f'Unsupported resolution status: {resolution_status}')

    resolution_clause = ''
    params: list[object] = []
    if resolution_status == 'open':
      resolution_clause = "AND task_items.exception_resolution_status IN ('open', 'planned')"
    elif resolution_status != 'all':
      resolution_clause = 'AND task_items.exception_resolution_status = ?'
      params.append(resolution_status)

    with self._connect() as connection:
      rows = connection.execute(
        '''
        SELECT
          task_items.*,
          task_runs.started_at AS task_started_at
        FROM task_items
        JOIN task_runs ON task_runs.id = task_items.task_id
        WHERE (
          task_items.organize_status IN ('trashed', 'failed')
          OR task_items.dedupe_status IN ('duplicate_trashed', 'failed')
          OR task_items.match_status IN ('low_score', 'failed', 'not_found')
          OR (
            task_items.preprocess_status = 'failed'
            AND task_items.preprocess_reason = 'convert_failed'
          )
          OR (
            task_items.preprocess_status = 'warning'
            AND task_items.preprocess_reason LIKE '%metadata_failed%'
          )
          OR (
            task_items.exception_resolution_status IN ('open', 'planned')
            AND task_items.exception_resolution_json IS NOT NULL
            AND (
              task_items.exception_resolution_json LIKE '%"workflow_state"%candidate_selected%'
              OR task_items.exception_resolution_json LIKE '%"workflow_state"%ready_to_ingest%'
            )
          )
        )
        ''' + resolution_clause + '''
        ORDER BY task_items.updated_at DESC, task_items.id DESC
        ''',
        params
      ).fetchall()

    return [self._parse_item_with_task_context(row) for row in rows]

  def get_exception_source_item(self, item_id: int) -> dict | None:
    with self._connect() as connection:
      row = connection.execute(
        '''
        SELECT
          task_items.*,
          task_runs.started_at AS task_started_at
        FROM task_items
        JOIN task_runs ON task_runs.id = task_items.task_id
        WHERE task_items.id = ?
        ''',
        (item_id,)
      ).fetchone()

    if row is None:
      return None

    return self._parse_item_with_task_context(row)

  def get_exception_source_items_by_ids(self, item_ids: list[int]) -> list[dict]:
    if not item_ids:
      return []

    placeholders = ', '.join('?' for _ in item_ids)
    with self._connect() as connection:
      rows = connection.execute(
        f'''
        SELECT
          task_items.*,
          task_runs.started_at AS task_started_at
        FROM task_items
        JOIN task_runs ON task_runs.id = task_items.task_id
        WHERE task_items.id IN ({placeholders})
        ''',
        item_ids
      ).fetchall()

    return [self._parse_item_with_task_context(row) for row in rows]

  def get_task_snapshot(self, task_id: str) -> dict:
    task = self.get_task(task_id)

    with self._connect() as connection:
      total_logs = connection.execute(
        'SELECT COUNT(*) AS total FROM task_logs WHERE task_id = ?',
        (task_id,)
      ).fetchone()['total']
      recent_rows = connection.execute(
        '''
        SELECT *
        FROM task_logs
        WHERE task_id = ?
        ORDER BY id DESC
        LIMIT ?
        ''',
        (task_id, SCAN_PROGRESS_LOG_LIMIT)
      ).fetchall()

    recent_logs = [self._parse_log(row) for row in reversed(recent_rows)]

    return {
      'task': task,
      'recent_logs': recent_logs,
      'recent_logs_limit': SCAN_PROGRESS_LOG_LIMIT,
      'has_more_logs': total_logs > len(recent_logs),
      'latest_log_id': recent_logs[-1]['id'] if recent_logs else None
    }

  def _parse_task(self, row: sqlite3.Row | None) -> dict | None:
    if row is None:
      return None

    raw_stats = json.loads(row['stats_json'])
    if row['task_type'] != TASK_TYPE_INGEST:
      stats = {
        'prepare': {
          **create_empty_repair_stats()['prepare'],
          **(raw_stats.get('prepare') or {})
        },
        'execute': {
          **create_empty_repair_stats()['execute'],
          **(raw_stats.get('execute') or {})
        }
      }
    elif 'scan' in raw_stats:
      stats = {
        'scan': {
          **create_empty_scan_stats(),
          **(raw_stats.get('scan') or {})
        },
        'preprocess': {
          **create_empty_preprocess_stats(),
          **(raw_stats.get('preprocess') or {})
        },
        'match': {
          **create_empty_match_stats(),
          **(raw_stats.get('match') or {})
        },
        'dedupe': {
          **create_empty_dedupe_stats(),
          **(raw_stats.get('dedupe') or {})
        },
        'organize': {
          **create_empty_organize_stats(),
          **(raw_stats.get('organize') or {})
        }
      }
    else:
      stats = {
        'scan': {
          **create_empty_scan_stats(),
          **raw_stats
        },
        'preprocess': create_empty_preprocess_stats(),
        'match': create_empty_match_stats(),
        'dedupe': create_empty_dedupe_stats(),
        'organize': create_empty_organize_stats()
      }

    return {
      'task_id': row['id'],
      'task_type': row['task_type'] or TASK_TYPE_INGEST,
      'trigger_source': row['trigger_source'],
      'source_task_id': row['source_task_id'],
      'status': row['status'],
      'current_stage': row['current_stage'],
      'stage_states': json.loads(row['stage_states_json']),
      'stats': stats,
      'repair_plan_json': json.loads(row['repair_plan_json']) if row['repair_plan_json'] else None,
      'error_message': row['error_message'],
      'started_at': row['started_at'],
      'completed_at': row['completed_at'],
      'updated_at': row['updated_at']
    }

  def _parse_item(self, row: sqlite3.Row) -> dict:
    return {
      'id': row['id'],
      'task_id': row['task_id'],
      'parent_item_id': row['parent_item_id'],
      'is_active': bool(row['is_active']),
      'original_path': row['original_path'],
      'current_file_path': row['current_file_path'] or row['original_path'],
      'relative_path': row['relative_path'],
      'filename': row['filename'],
      'extension': row['extension'],
      'size_bytes': row['size_bytes'],
      'modified_at': row['modified_at'],
      'local_cover': row['local_cover'],
      'local_lyric': row['local_lyric'],
      'scan_status': row['scan_status'],
      'scan_reason': row['scan_reason'],
      'scan_message': row['scan_message'],
      'preprocess_status': row['preprocess_status'] or 'pending',
      'preprocess_reason': row['preprocess_reason'],
      'preprocess_message': row['preprocess_message'],
      'audio_props_json': json.loads(row['audio_props_json']) if row['audio_props_json'] else None,
      'original_tags_json': json.loads(row['original_tags_json']) if row['original_tags_json'] else None,
      'preprocess_artifacts_json': (
        json.loads(row['preprocess_artifacts_json'])
        if row['preprocess_artifacts_json']
        else None
      ),
      'acoustic_fingerprint': row['acoustic_fingerprint'],
      'fingerprint_duration_seconds': row['fingerprint_duration_seconds'],
      'match_status': row['match_status'] or 'pending',
      'match_reason': row['match_reason'],
      'match_message': row['match_message'],
      'match_source': row['match_source'],
      'match_confidence': row['match_confidence'],
      'match_is_authoritative': bool(row['match_is_authoritative']),
      'matched_metadata_json': (
        json.loads(row['matched_metadata_json'])
        if row['matched_metadata_json']
        else None
      ),
      'match_candidates_json': (
        json.loads(row['match_candidates_json'])
        if row['match_candidates_json']
        else None
      ),
      'match_enrichment_json': (
        json.loads(row['match_enrichment_json'])
        if row['match_enrichment_json']
        else None
      ),
      'dedupe_status': row['dedupe_status'] or 'pending',
      'dedupe_reason': row['dedupe_reason'],
      'dedupe_message': row['dedupe_message'],
      'dedupe_group_key': row['dedupe_group_key'],
      'duplicate_of_path': row['duplicate_of_path'],
      'duplicate_of_item_id': row['duplicate_of_item_id'],
      'dedupe_decision_json': (
        json.loads(row['dedupe_decision_json'])
        if row['dedupe_decision_json']
        else None
      ),
      'organize_status': row['organize_status'] or 'pending',
      'organize_reason': row['organize_reason'],
      'organize_message': row['organize_message'],
      'library_relative_path': row['library_relative_path'],
      'library_file_path': row['library_file_path'],
      'trash_file_path': row['trash_file_path'],
      'organize_decision_json': (
        json.loads(row['organize_decision_json'])
        if row['organize_decision_json']
        else None
      ),
      'exception_resolution_status': row['exception_resolution_status'] or 'open',
      'exception_resolution_json': (
        json.loads(row['exception_resolution_json'])
        if row['exception_resolution_json']
        else None
      ),
      'last_repair_task_id': row['last_repair_task_id'],
      'created_at': row['created_at'],
      'updated_at': row['updated_at']
    }

  def _parse_item_with_task_context(self, row: sqlite3.Row) -> dict:
    item = self._parse_item(row)
    item['task_started_at'] = row['task_started_at']
    return item

  def _parse_log(self, row: sqlite3.Row) -> dict:
    return {
      'id': row['id'],
      'task_id': row['task_id'],
      'stage': row['stage'],
      'level': row['level'],
      'event_type': row['event_type'],
      'message': row['message'],
      'payload': json.loads(row['payload_json']) if row['payload_json'] else None,
      'created_at': row['created_at']
    }

  def _parse_task_history_row(self, row: sqlite3.Row) -> dict:
    total_items = row['total_items'] or 0
    success_items = row['success_items'] or 0
    exception_items = total_items - success_items

    return {
      'task_id': row['task_id'],
      'started_at': row['started_at'],
      'status': row['status'],
      'total_items': total_items,
      'success_items': success_items,
      'exception_items': exception_items,
      'report_status': (
        'success'
        if row['status'] == TASK_STATUS_COMPLETED and exception_items == 0
        else 'warning'
      )
    }