from __future__ import annotations

import json
import hashlib
import sqlite3
import time
from dataclasses import asdict, dataclass
from pathlib import Path
from typing import Any

from .agent_types import (
    AgentPermissions,
    AgentRuntimeConfig,
    BudgetConfig,
    DEFAULT_MAX_TURNS,
    ModelConfig,
    ModelPricing,
    OutputSchemaConfig,
    UsageStats,
)


@dataclass(frozen=True)
class StoredSession:
    session_id: str
    messages: tuple[str, ...]
    input_tokens: int
    output_tokens: int


DEFAULT_SESSION_DIR = Path('.port_sessions')
DEFAULT_AGENT_SESSION_DIR = DEFAULT_SESSION_DIR / 'agent'
AGENT_SESSION_DB_FILENAME = 'sessions.db'


def save_session(session: StoredSession, directory: Path | None = None) -> Path:
    target_dir = directory or DEFAULT_SESSION_DIR
    target_dir.mkdir(parents=True, exist_ok=True)
    path = target_dir / f'{session.session_id}.json'
    path.write_text(json.dumps(asdict(session), indent=2))
    return path


def load_session(session_id: str, directory: Path | None = None) -> StoredSession:
    target_dir = directory or DEFAULT_SESSION_DIR
    data = json.loads((target_dir / f'{session_id}.json').read_text())
    return StoredSession(
        session_id=data['session_id'],
        messages=tuple(data['messages']),
        input_tokens=data['input_tokens'],
        output_tokens=data['output_tokens'],
    )


JSONDict = dict[str, Any]


@dataclass(frozen=True)
class StoredAgentSession:
    session_id: str
    model_config: JSONDict
    runtime_config: JSONDict
    system_prompt_parts: tuple[str, ...]
    user_context: dict[str, str]
    system_context: dict[str, str]
    messages: tuple[JSONDict, ...]
    turns: int
    tool_calls: int
    usage: JSONDict
    total_cost_usd: float
    file_history: tuple[JSONDict, ...]
    budget_state: JSONDict
    plugin_state: JSONDict
    scratchpad_directory: str | None = None
    is_training: bool = False
    session_metadata: JSONDict | None = None
    display_messages: tuple[JSONDict, ...] = ()


def save_agent_session(session: StoredAgentSession, directory: Path | None = None) -> Path:
    target_dir = directory or DEFAULT_AGENT_SESSION_DIR
    target_dir.mkdir(parents=True, exist_ok=True)
    session_dir = target_dir / session.session_id
    session_dir.mkdir(parents=True, exist_ok=True)
    path = session_dir / 'session.json'
    payload = asdict(session)
    _write_agent_session_db_payload(target_dir, session.session_id, payload, path)
    _sync_agent_display_messages(
        target_dir,
        session.session_id,
        tuple(message for message in session.display_messages if isinstance(message, dict))
        or tuple(message for message in session.messages if isinstance(message, dict)),
    )
    path.write_text(json.dumps(payload, indent=2, ensure_ascii=False), encoding='utf-8')
    return path


def load_agent_session(session_id: str, directory: Path | None = None) -> StoredAgentSession:
    target_dir = directory or DEFAULT_AGENT_SESSION_DIR
    data = _read_agent_session_db_payload(target_dir, session_id)
    if data is not None:
        return _stored_agent_session_from_payload(data)
    path = target_dir / session_id / 'session.json'
    if not path.exists():
        path = target_dir / f'{session_id}.json'
    data = json.loads(path.read_text(encoding='utf-8'))
    # Backfill SQLite on first read so future live UI reads do not depend on
    # repeatedly parsing JSON snapshots.
    _write_agent_session_db_payload(target_dir, session_id, data, path)
    return _stored_agent_session_from_payload(data)


def list_agent_sessions(
    directory: Path | None = None,
) -> list[tuple[StoredAgentSession, float]]:
    target_dir = directory or DEFAULT_AGENT_SESSION_DIR
    sessions: dict[str, tuple[StoredAgentSession, float]] = {}
    for data, updated_at in _iter_agent_session_db_payloads(target_dir):
        try:
            stored = _stored_agent_session_from_payload(data)
        except (KeyError, TypeError, ValueError):
            continue
        sessions[stored.session_id] = (stored, updated_at)

    for path in _iter_agent_session_snapshot_files(target_dir):
        session_id = _session_id_from_snapshot_path(path)
        if session_id in sessions:
            continue
        try:
            stored = load_agent_session(session_id, directory=target_dir)
        except (FileNotFoundError, json.JSONDecodeError, KeyError, TypeError, ValueError):
            continue
        try:
            updated_at = path.stat().st_mtime
        except OSError:
            updated_at = 0.0
        sessions[stored.session_id] = (stored, updated_at)
    return sorted(sessions.values(), key=lambda item: item[1], reverse=True)


def agent_session_delta(
    session_id: str,
    directory: Path | None = None,
    *,
    after_message_seq: int = 0,
    limit: int = 500,
) -> dict[str, Any]:
    target_dir = directory or DEFAULT_AGENT_SESSION_DIR
    try:
        stored = load_agent_session(session_id, directory=target_dir)
    except (FileNotFoundError, json.JSONDecodeError, KeyError, TypeError, ValueError):
        stored = None
    if stored is not None:
        _sync_agent_display_messages(
            target_dir,
            session_id,
            stored.display_messages or stored.messages,
        )
    with _connect_agent_session_db(target_dir) as conn:
        latest_row = conn.execute(
            """
            select max(seq) as latest_seq
            from agent_display_messages
            where session_id = ?
            """,
            (session_id,),
        ).fetchone()
        rows = conn.execute(
            """
            select seq, message_json, updated_at
            from agent_display_messages
            where session_id = ? and seq > ?
            order by seq asc
            limit ?
            """,
            (session_id, max(0, int(after_message_seq)), max(1, int(limit))),
        ).fetchall()
    messages: list[dict[str, Any]] = []
    for row in rows:
        try:
            message = json.loads(row['message_json'])
        except (TypeError, json.JSONDecodeError):
            continue
        if isinstance(message, dict):
            messages.append(
                {
                    'seq': int(row['seq']),
                    'updated_at': float(row['updated_at'] or 0.0),
                    'message': message,
                }
            )
    latest_seq = 0
    if latest_row is not None and latest_row['latest_seq'] is not None:
        latest_seq = int(latest_row['latest_seq'])
    return {
        'session_id': session_id,
        'latest_message_seq': latest_seq,
        'messages': messages,
        'session': (
            {
                'session_id': stored.session_id,
                'turns': stored.turns,
                'tool_calls': stored.tool_calls,
                'usage': stored.usage,
                'total_cost_usd': stored.total_cost_usd,
                'model': stored.model_config.get('model'),
                'is_training': stored.is_training,
                'session_metadata': stored.session_metadata or {},
            }
            if stored is not None
            else {'session_id': session_id}
        ),
    }


def append_agent_display_message(
    session_id: str,
    message: JSONDict,
    directory: Path | None = None,
) -> dict[str, Any] | None:
    target_dir = directory or DEFAULT_AGENT_SESSION_DIR
    if not isinstance(message, dict):
        return None
    message_key = _display_message_key(message)
    message_json = json.dumps(message, ensure_ascii=False, sort_keys=True)
    now = time.time()
    with _connect_agent_session_db(target_dir) as conn:
        existing = conn.execute(
            """
            select seq, message_json
            from agent_display_messages
            where session_id = ? and message_key = ?
            order by seq asc
            limit 1
            """,
            (session_id, message_key),
        ).fetchone()
        if existing is not None:
            if existing['message_json'] != message_json:
                conn.execute(
                    """
                    update agent_display_messages
                    set role = ?,
                        updated_at = ?,
                        message_json = ?
                    where session_id = ? and seq = ?
                    """,
                    (
                        str(message.get('role') or ''),
                        now,
                        message_json,
                        session_id,
                        int(existing['seq']),
                    ),
                )
            return {
                'seq': int(existing['seq']),
                'updated_at': now,
                'message': message,
            }
        latest = conn.execute(
            """
            select max(seq) as latest_seq
            from agent_display_messages
            where session_id = ?
            """,
            (session_id,),
        ).fetchone()
        seq = int(latest['latest_seq'] or 0) + 1 if latest is not None else 1
        conn.execute(
            """
            insert into agent_display_messages (
                session_id, seq, message_key, role, updated_at, message_json
            )
            values (?, ?, ?, ?, ?, ?)
            """,
            (
                session_id,
                seq,
                message_key,
                str(message.get('role') or ''),
                now,
                message_json,
            ),
        )
    return {'seq': seq, 'updated_at': now, 'message': message}


def enqueue_session_input(
    session_id: str,
    *,
    content: str,
    directory: Path | None = None,
    run_id: str | None = None,
    kind: str = 'next_turn',
    status: str = 'pending',
    metadata: dict[str, Any] | None = None,
) -> dict[str, Any]:
    target_dir = directory or DEFAULT_AGENT_SESSION_DIR
    now = time.time()
    normalized_kind = kind if kind in {'next_turn', 'guidance'} else 'next_turn'
    with _connect_agent_session_db(target_dir) as conn:
        cursor = conn.execute(
            """
            insert into agent_input_queue (
                session_id, run_id, kind, status, content,
                created_at, updated_at, metadata_json
            )
            values (?, ?, ?, ?, ?, ?, ?, ?)
            """,
            (
                session_id,
                run_id or '',
                normalized_kind,
                status,
                content,
                now,
                now,
                json.dumps(metadata or {}, ensure_ascii=False, sort_keys=True),
            ),
        )
        item_id = int(cursor.lastrowid)
    return {
        'id': item_id,
        'session_id': session_id,
        'run_id': run_id or '',
        'kind': normalized_kind,
        'status': status,
        'content': content,
        'created_at': now,
        'updated_at': now,
        'metadata': metadata or {},
    }


def list_session_input_queue(
    session_id: str,
    directory: Path | None = None,
    *,
    statuses: tuple[str, ...] = ('pending',),
    limit: int = 100,
) -> list[dict[str, Any]]:
    target_dir = directory or DEFAULT_AGENT_SESSION_DIR
    placeholders = ','.join('?' for _ in statuses)
    with _connect_agent_session_db(target_dir) as conn:
        rows = conn.execute(
            f"""
            select *
            from agent_input_queue
            where session_id = ?
              and status in ({placeholders})
            order by id asc
            limit ?
            """,
            (session_id, *statuses, max(1, int(limit))),
        ).fetchall()
    return [_queue_row_to_dict(row) for row in rows]


def update_session_input_queue_item(
    session_id: str,
    item_id: int,
    directory: Path | None = None,
    *,
    content: str | None = None,
    kind: str | None = None,
    status: str | None = None,
    run_id: str | None = None,
) -> dict[str, Any] | None:
    target_dir = directory or DEFAULT_AGENT_SESSION_DIR
    assignments: list[str] = ['updated_at = ?']
    values: list[Any] = [time.time()]
    if content is not None:
        assignments.append('content = ?')
        values.append(content)
    if kind is not None:
        assignments.append('kind = ?')
        values.append(kind if kind in {'next_turn', 'guidance'} else 'next_turn')
    if status is not None:
        assignments.append('status = ?')
        values.append(status)
        if status in {'consumed', 'cancelled'}:
            assignments.append('consumed_at = coalesce(consumed_at, ?)')
            values.append(time.time())
    if run_id is not None:
        assignments.append('run_id = ?')
        values.append(run_id)
    values.extend([session_id, int(item_id)])
    with _connect_agent_session_db(target_dir) as conn:
        conn.execute(
            f"""
            update agent_input_queue
            set {', '.join(assignments)}
            where session_id = ? and id = ?
            """,
            values,
        )
        row = conn.execute(
            """
            select *
            from agent_input_queue
            where session_id = ? and id = ?
            """,
            (session_id, int(item_id)),
        ).fetchone()
    return _queue_row_to_dict(row) if row is not None else None


def cancel_session_input_queue(
    session_id: str,
    directory: Path | None = None,
    *,
    run_id: str | None = None,
) -> int:
    target_dir = directory or DEFAULT_AGENT_SESSION_DIR
    now = time.time()
    where = 'session_id = ? and status = ?'
    values: list[Any] = [session_id, 'pending']
    if run_id:
        where += ' and (run_id = ? or run_id = ?)'
        values.extend([run_id, ''])
    with _connect_agent_session_db(target_dir) as conn:
        cursor = conn.execute(
            f"""
            update agent_input_queue
            set status = 'cancelled',
                updated_at = ?,
                consumed_at = coalesce(consumed_at, ?)
            where {where}
            """,
            (now, now, *values),
        )
        return int(cursor.rowcount or 0)


def consume_session_guidance(
    session_id: str,
    run_id: str,
    directory: Path | None = None,
    *,
    limit: int = 10,
) -> list[dict[str, Any]]:
    target_dir = directory or DEFAULT_AGENT_SESSION_DIR
    now = time.time()
    with _connect_agent_session_db(target_dir) as conn:
        rows = conn.execute(
            """
            select *
            from agent_input_queue
            where session_id = ?
              and status = 'pending'
              and kind = 'guidance'
              and (run_id = ? or run_id = '')
            order by id asc
            limit ?
            """,
            (session_id, run_id, max(1, int(limit))),
        ).fetchall()
        ids = [int(row['id']) for row in rows]
        if ids:
            conn.execute(
                f"""
                update agent_input_queue
                set status = 'consumed',
                    updated_at = ?,
                    consumed_at = ?
                where id in ({','.join('?' for _ in ids)})
                """,
                (now, now, *ids),
            )
    return [_queue_row_to_dict(row) for row in rows]


def consume_next_session_input(
    session_id: str,
    directory: Path | None = None,
) -> dict[str, Any] | None:
    target_dir = directory or DEFAULT_AGENT_SESSION_DIR
    now = time.time()
    with _connect_agent_session_db(target_dir) as conn:
        row = conn.execute(
            """
            select *
            from agent_input_queue
            where session_id = ?
              and status = 'pending'
              and kind = 'next_turn'
            order by id asc
            limit 1
            """,
            (session_id,),
        ).fetchone()
        if row is None:
            return None
        cursor = conn.execute(
            """
            update agent_input_queue
            set status = 'consumed',
                updated_at = ?,
                consumed_at = ?
            where id = ? and status = 'pending'
            """,
            (now, now, int(row['id'])),
        )
        if int(cursor.rowcount or 0) <= 0:
            return None
    item = _queue_row_to_dict(row)
    item['status'] = 'consumed'
    item['updated_at'] = now
    item['consumed_at'] = now
    return item


def delete_agent_session(session_id: str, directory: Path | None = None) -> bool:
    target_dir = directory or DEFAULT_AGENT_SESSION_DIR
    deleted = False
    with _connect_agent_session_db(target_dir) as conn:
        cursor = conn.execute(
            'delete from agent_sessions where session_id = ?',
            (session_id,),
        )
        deleted = cursor.rowcount > 0
        conn.execute(
            'delete from agent_display_messages where session_id = ?',
            (session_id,),
        )
        conn.execute(
            'delete from agent_input_queue where session_id = ?',
            (session_id,),
        )
    nested = target_dir / session_id
    if nested.exists():
        import shutil

        shutil.rmtree(nested)
        deleted = True
    legacy = target_dir / f'{session_id}.json'
    if legacy.exists():
        legacy.unlink()
        deleted = True
    return deleted


def _stored_agent_session_from_payload(data: JSONDict) -> StoredAgentSession:
    messages = tuple(
        message for message in data['messages'] if isinstance(message, dict)
    )
    display_messages = tuple(
        message
        for message in data.get('display_messages', messages)
        if isinstance(message, dict)
    )
    if not display_messages:
        display_messages = messages
    session_metadata = (
        dict(data.get('session_metadata', {}))
        if isinstance(data.get('session_metadata'), dict)
        else {}
    )
    for key in ('title', 'title_source', 'title_updated_at'):
        if key in data and key not in session_metadata:
            session_metadata[key] = data[key]
    return StoredAgentSession(
        session_id=data['session_id'],
        model_config=dict(data['model_config']),
        runtime_config=dict(data['runtime_config']),
        system_prompt_parts=tuple(data['system_prompt_parts']),
        user_context=dict(data['user_context']),
        system_context=dict(data['system_context']),
        messages=messages,
        turns=int(data['turns']),
        tool_calls=int(data['tool_calls']),
        usage=dict(data.get('usage', {})),
        total_cost_usd=float(data.get('total_cost_usd', 0.0)),
        file_history=tuple(
            entry for entry in data.get('file_history', []) if isinstance(entry, dict)
        ),
        budget_state=(
            dict(data.get('budget_state', {}))
            if isinstance(data.get('budget_state'), dict)
            else {}
        ),
        plugin_state=(
            dict(data.get('plugin_state', {}))
            if isinstance(data.get('plugin_state'), dict)
            else {}
        ),
        scratchpad_directory=(
            str(data['scratchpad_directory'])
            if isinstance(data.get('scratchpad_directory'), str)
            else None
        ),
        is_training=bool(data.get('is_training', False)),
        session_metadata=session_metadata,
        display_messages=display_messages,
    )


def _write_agent_session_db_payload(
    directory: Path,
    session_id: str,
    payload: JSONDict,
    snapshot_path: Path,
) -> None:
    updated_at = time.time()
    payload_json = json.dumps(payload, ensure_ascii=False, sort_keys=True)
    with _connect_agent_session_db(directory) as conn:
        current = conn.execute(
            """
            select updated_at, snapshot_path, payload_json
            from agent_sessions
            where session_id = ?
            """,
            (session_id,),
        ).fetchone()
        if current is not None and current['payload_json'] != payload_json:
            _write_agent_session_revision(
                conn,
                session_id=session_id,
                updated_at=float(current['updated_at'] or 0.0),
                snapshot_path=str(current['snapshot_path'] or ''),
                payload_json=str(current['payload_json'] or ''),
            )
        _write_agent_session_revision(
            conn,
            session_id=session_id,
            updated_at=updated_at,
            snapshot_path=str(snapshot_path),
            payload_json=payload_json,
        )
        conn.execute(
            """
            insert into agent_sessions (
                session_id, updated_at, snapshot_path, payload_json
            )
            values (?, ?, ?, ?)
            on conflict(session_id) do update set
                updated_at=excluded.updated_at,
                snapshot_path=excluded.snapshot_path,
                payload_json=excluded.payload_json
            """,
            (session_id, updated_at, str(snapshot_path), payload_json),
        )


def _write_agent_session_revision(
    conn: sqlite3.Connection,
    *,
    session_id: str,
    updated_at: float,
    snapshot_path: str,
    payload_json: str,
) -> None:
    if not payload_json:
        return
    latest = conn.execute(
        """
        select payload_json
        from agent_session_revisions
        where session_id = ?
        order by revision_id desc
        limit 1
        """,
        (session_id,),
    ).fetchone()
    if latest is not None and latest['payload_json'] == payload_json:
        return
    conn.execute(
        """
        insert into agent_session_revisions (
            session_id, updated_at, snapshot_path, payload_json
        )
        values (?, ?, ?, ?)
        """,
        (session_id, updated_at, snapshot_path, payload_json),
    )


def _sync_agent_display_messages(
    directory: Path,
    session_id: str,
    display_messages: tuple[JSONDict, ...],
) -> None:
    if not display_messages:
        return
    now = time.time()
    with _connect_agent_session_db(directory) as conn:
        existing_rows = conn.execute(
            """
            select seq, message_key, message_json
            from agent_display_messages
            where session_id = ?
            """,
            (session_id,),
        ).fetchall()
        existing: dict[int, sqlite3.Row] = {
            int(row['seq']): row for row in existing_rows
        }
        existing_by_key: dict[str, sqlite3.Row] = {
            str(row['message_key'] or ''): row
            for row in existing_rows
            if str(row['message_key'] or '')
        }
        next_seq = max(existing.keys(), default=0) + 1
        seq_offset = 0
        keep_seqs: set[int] = set()
        for seq, row in existing.items():
            try:
                existing_message = json.loads(row['message_json'])
            except (TypeError, json.JSONDecodeError):
                existing_message = {}
            if isinstance(existing_message, dict) and _is_persistent_display_message(
                existing_message
            ):
                keep_seqs.add(seq)
        for index, message in enumerate(display_messages, start=1):
            if not isinstance(message, dict):
                continue
            message_json = json.dumps(message, ensure_ascii=False, sort_keys=True)
            message_key = _display_message_key(message)
            target_seq = index + seq_offset
            row = existing.get(target_seq)
            while row is not None:
                try:
                    old_target_message = json.loads(row['message_json'])
                except (TypeError, json.JSONDecodeError):
                    old_target_message = {}
                if not isinstance(old_target_message, dict):
                    old_target_message = {}
                if not _is_persistent_display_message(old_target_message):
                    break
                # Display-only messages, such as runtime guidance, are inserted
                # outside the model transcript. Later full-session syncs must
                # preserve them and shift model messages after them; otherwise
                # the UI sees an old pending prompt as a new user branch.
                seq_offset += 1
                target_seq = index + seq_offset
                row = existing.get(target_seq)
            keyed_row = existing_by_key.get(message_key)
            if keyed_row is not None and keyed_row['message_json'] == message_json:
                keep_seqs.add(int(keyed_row['seq']))
                continue
            if keyed_row is not None:
                try:
                    keyed_old_message = json.loads(keyed_row['message_json'])
                except (TypeError, json.JSONDecodeError):
                    keyed_old_message = {}
                if not isinstance(keyed_old_message, dict):
                    keyed_old_message = {}
                if _is_replaceable_display_message(keyed_old_message):
                    conn.execute(
                        """
                        update agent_display_messages
                        set role = ?,
                            updated_at = ?,
                            message_json = ?
                        where session_id = ? and seq = ?
                        """,
                        (
                            str(message.get('role') or ''),
                            now,
                            message_json,
                            session_id,
                            int(keyed_row['seq']),
                        ),
                    )
                    updated_row = {
                        'seq': int(keyed_row['seq']),
                        'message_key': message_key,
                        'message_json': message_json,
                    }
                    existing[int(keyed_row['seq'])] = updated_row  # type: ignore[assignment]
                    existing_by_key[message_key] = updated_row  # type: ignore[assignment]
                    keep_seqs.add(int(keyed_row['seq']))
                continue
            if row is None:
                conn.execute(
                    """
                    insert into agent_display_messages (
                        session_id, seq, message_key, role, updated_at, message_json
                    )
                    values (?, ?, ?, ?, ?, ?)
                    """,
                    (
                        session_id,
                        target_seq,
                        message_key,
                        str(message.get('role') or ''),
                        now,
                        message_json,
                    ),
                )
                inserted_row = {
                    'seq': target_seq,
                    'message_key': message_key,
                    'message_json': message_json,
                }
                existing[target_seq] = inserted_row  # type: ignore[assignment]
                existing_by_key[message_key] = inserted_row  # type: ignore[assignment]
                next_seq = max(next_seq, target_seq + 1)
                keep_seqs.add(target_seq)
                continue
            if row['message_json'] == message_json:
                keep_seqs.add(target_seq)
                continue
            try:
                old_message = json.loads(row['message_json'])
            except (TypeError, json.JSONDecodeError):
                old_message = {}
            if not isinstance(old_message, dict):
                old_message = {}
            old_key = str(row['message_key'] or '')
            should_update = (
                old_key == message_key
                or _is_replaceable_display_message(old_message)
                or _same_role_and_content(old_message, message)
            )
            if not should_update:
                append_seq = next_seq
                next_seq += 1
                conn.execute(
                    """
                    insert into agent_display_messages (
                        session_id, seq, message_key, role, updated_at, message_json
                    )
                    values (?, ?, ?, ?, ?, ?)
                    """,
                    (
                        session_id,
                        append_seq,
                        message_key,
                        str(message.get('role') or ''),
                        now,
                        message_json,
                    ),
                )
                appended_row = {
                    'seq': append_seq,
                    'message_key': message_key,
                    'message_json': message_json,
                }
                existing[append_seq] = appended_row  # type: ignore[assignment]
                existing_by_key[message_key] = appended_row  # type: ignore[assignment]
                keep_seqs.add(append_seq)
                continue
            conn.execute(
                """
                update agent_display_messages
                set message_key = ?,
                    role = ?,
                    updated_at = ?,
                    message_json = ?
                where session_id = ? and seq = ?
                """,
                (
                    message_key,
                    str(message.get('role') or ''),
                    now,
                    message_json,
                    session_id,
                    target_seq,
                ),
            )
            updated_row = {
                'seq': target_seq,
                'message_key': message_key,
                'message_json': message_json,
            }
            existing[target_seq] = updated_row  # type: ignore[assignment]
            existing_by_key[message_key] = updated_row  # type: ignore[assignment]
            keep_seqs.add(target_seq)
        stale_seqs = [
            seq
            for seq, row in existing.items()
            if seq not in keep_seqs
            and not _is_persistent_display_row(row)
        ]
        if stale_seqs:
            conn.execute(
                f"""
                delete from agent_display_messages
                where session_id = ?
                  and seq in ({','.join('?' for _ in stale_seqs)})
                """,
                (session_id, *stale_seqs),
            )


def _display_message_key(message: JSONDict) -> str:
    message_id = message.get('message_id')
    if isinstance(message_id, str) and message_id:
        return f'id:{message_id}'
    tool_call_id = message.get('tool_call_id')
    if isinstance(tool_call_id, str) and tool_call_id:
        return f'tool:{tool_call_id}'
    role = str(message.get('role') or '')
    content = '' if message.get('content') is None else str(message.get('content') or '')
    digest = hashlib.sha1(f'{role}\0{content}'.encode('utf-8')).hexdigest()
    return f'hash:{digest}'


def _is_replaceable_display_message(message: JSONDict) -> bool:
    metadata = message.get('metadata')
    if not isinstance(metadata, dict):
        return False
    if metadata.get('placeholder') is True:
        return True
    return metadata.get('kind') in {'run_status'}


def _is_persistent_display_message(message: JSONDict) -> bool:
    metadata = message.get('metadata')
    return (
        isinstance(metadata, dict)
        and metadata.get('kind') in {'runtime_guidance_display'}
    )


def _is_persistent_display_row(row: sqlite3.Row | dict[str, Any]) -> bool:
    try:
        message = json.loads(row['message_json'])
    except (TypeError, json.JSONDecodeError, KeyError):
        return False
    return isinstance(message, dict) and _is_persistent_display_message(message)


def _same_role_and_content(left: JSONDict, right: JSONDict) -> bool:
    return (
        str(left.get('role') or '') == str(right.get('role') or '')
        and str(left.get('content') or '') == str(right.get('content') or '')
    )


def _queue_row_to_dict(row: sqlite3.Row) -> dict[str, Any]:
    try:
        metadata = json.loads(row['metadata_json'] or '{}')
    except (TypeError, json.JSONDecodeError):
        metadata = {}
    return {
        'id': int(row['id']),
        'session_id': str(row['session_id'] or ''),
        'run_id': str(row['run_id'] or ''),
        'kind': str(row['kind'] or ''),
        'status': str(row['status'] or ''),
        'content': str(row['content'] or ''),
        'created_at': float(row['created_at'] or 0.0),
        'updated_at': float(row['updated_at'] or 0.0),
        'consumed_at': (
            float(row['consumed_at'])
            if row['consumed_at'] is not None
            else None
        ),
        'metadata': metadata if isinstance(metadata, dict) else {},
    }


def _read_agent_session_db_payload(
    directory: Path,
    session_id: str,
) -> JSONDict | None:
    with _connect_agent_session_db(directory) as conn:
        row = conn.execute(
            'select payload_json from agent_sessions where session_id = ?',
            (session_id,),
        ).fetchone()
    if row is None:
        return None
    try:
        payload = json.loads(row['payload_json'])
    except (TypeError, json.JSONDecodeError):
        return None
    return payload if isinstance(payload, dict) else None


def _iter_agent_session_db_payloads(directory: Path) -> list[tuple[JSONDict, float]]:
    with _connect_agent_session_db(directory) as conn:
        rows = conn.execute(
            """
            select payload_json, updated_at
            from agent_sessions
            order by updated_at desc
            """
        ).fetchall()
    result: list[tuple[JSONDict, float]] = []
    for row in rows:
        try:
            payload = json.loads(row['payload_json'])
        except (TypeError, json.JSONDecodeError):
            continue
        if isinstance(payload, dict):
            result.append((payload, float(row['updated_at'] or 0.0)))
    return result


def _connect_agent_session_db(directory: Path) -> sqlite3.Connection:
    directory.mkdir(parents=True, exist_ok=True)
    conn = sqlite3.connect(directory.parent / AGENT_SESSION_DB_FILENAME, timeout=10)
    conn.row_factory = sqlite3.Row
    conn.execute('pragma journal_mode=wal')
    conn.execute('pragma busy_timeout=5000')
    conn.execute(
        """
        create table if not exists agent_sessions (
            session_id text primary key,
            updated_at real not null,
            snapshot_path text default '',
            payload_json text not null
        )
        """
    )
    conn.execute(
        """
        create index if not exists idx_agent_sessions_updated_at
        on agent_sessions(updated_at)
        """
    )
    conn.execute(
        """
        create table if not exists agent_session_revisions (
            revision_id integer primary key autoincrement,
            session_id text not null,
            updated_at real not null,
            snapshot_path text default '',
            payload_json text not null
        )
        """
    )
    conn.execute(
        """
        create index if not exists idx_agent_session_revisions_session
        on agent_session_revisions(session_id, revision_id)
        """
    )
    conn.execute(
        """
        create table if not exists agent_display_messages (
            session_id text not null,
            seq integer not null,
            message_key text not null,
            role text default '',
            updated_at real not null,
            message_json text not null,
            primary key(session_id, seq)
        )
        """
    )
    conn.execute(
        """
        create index if not exists idx_agent_display_messages_session
        on agent_display_messages(session_id, seq)
        """
    )
    conn.execute(
        """
        create table if not exists agent_input_queue (
            id integer primary key autoincrement,
            session_id text not null,
            run_id text default '',
            kind text not null,
            status text not null,
            content text not null,
            created_at real not null,
            updated_at real not null,
            consumed_at real,
            metadata_json text default '{}'
        )
        """
    )
    conn.execute(
        """
        create index if not exists idx_agent_input_queue_session
        on agent_input_queue(session_id, status, id)
        """
    )
    conn.execute(
        """
        create index if not exists idx_agent_input_queue_run
        on agent_input_queue(session_id, run_id, status, id)
        """
    )
    return conn


def _iter_agent_session_snapshot_files(directory: Path) -> list[Path]:
    if not directory.exists():
        return []
    by_session_id: dict[str, Path] = {}
    for path in directory.glob('*.json'):
        by_session_id[_session_id_from_snapshot_path(path)] = path
    for path in directory.glob('*/session.json'):
        by_session_id[_session_id_from_snapshot_path(path)] = path
    return list(by_session_id.values())


def _session_id_from_snapshot_path(path: Path) -> str:
    if path.name == 'session.json':
        return path.parent.name
    return path.stem


def serialize_model_config(model_config: ModelConfig) -> JSONDict:
    return {
        'model': model_config.model,
        'base_url': model_config.base_url,
        'api_key': model_config.api_key,
        'temperature': model_config.temperature,
        'timeout_seconds': model_config.timeout_seconds,
        'pricing': {
            'input_cost_per_million_tokens_usd': model_config.pricing.input_cost_per_million_tokens_usd,
            'output_cost_per_million_tokens_usd': model_config.pricing.output_cost_per_million_tokens_usd,
            'cache_creation_input_cost_per_million_tokens_usd': model_config.pricing.cache_creation_input_cost_per_million_tokens_usd,
            'cache_read_input_cost_per_million_tokens_usd': model_config.pricing.cache_read_input_cost_per_million_tokens_usd,
        },
    }


def deserialize_model_config(payload: JSONDict) -> ModelConfig:
    return ModelConfig(
        model=str(payload['model']),
        base_url=str(payload.get('base_url', 'http://127.0.0.1:8000/v1')),
        api_key=str(payload.get('api_key', 'local-token')),
        temperature=float(payload.get('temperature', 0.0)),
        timeout_seconds=float(payload.get('timeout_seconds', 120.0)),
        pricing=_deserialize_pricing(payload.get('pricing')),
    )


def serialize_runtime_config(runtime_config: AgentRuntimeConfig) -> JSONDict:
    return {
        'cwd': str(runtime_config.cwd),
        'max_turns': runtime_config.max_turns,
        'command_timeout_seconds': runtime_config.command_timeout_seconds,
        'max_output_chars': runtime_config.max_output_chars,
        'stream_model_responses': runtime_config.stream_model_responses,
        'auto_snip_threshold_tokens': runtime_config.auto_snip_threshold_tokens,
        'auto_compact_threshold_tokens': runtime_config.auto_compact_threshold_tokens,
        'compact_preserve_messages': runtime_config.compact_preserve_messages,
        'permissions': {
            'allow_file_write': runtime_config.permissions.allow_file_write,
            'allow_shell_commands': runtime_config.permissions.allow_shell_commands,
            'allow_destructive_shell_commands': runtime_config.permissions.allow_destructive_shell_commands,
        },
        'additional_working_directories': [str(path) for path in runtime_config.additional_working_directories],
        'disable_claude_md_discovery': runtime_config.disable_claude_md_discovery,
        'budget_config': {
            'max_total_tokens': runtime_config.budget_config.max_total_tokens,
            'max_input_tokens': runtime_config.budget_config.max_input_tokens,
            'max_output_tokens': runtime_config.budget_config.max_output_tokens,
            'max_reasoning_tokens': runtime_config.budget_config.max_reasoning_tokens,
            'max_total_cost_usd': runtime_config.budget_config.max_total_cost_usd,
            'max_tool_calls': runtime_config.budget_config.max_tool_calls,
            'max_delegated_tasks': runtime_config.budget_config.max_delegated_tasks,
            'max_model_calls': runtime_config.budget_config.max_model_calls,
            'max_session_turns': runtime_config.budget_config.max_session_turns,
        },
        'output_schema': (
            {
                'name': runtime_config.output_schema.name,
                'schema': runtime_config.output_schema.schema,
                'strict': runtime_config.output_schema.strict,
            }
            if runtime_config.output_schema is not None
            else None
        ),
        'session_directory': str(runtime_config.session_directory),
        'scratchpad_root': str(runtime_config.scratchpad_root),
        'python_env_dir': (
            str(runtime_config.python_env_dir)
            if runtime_config.python_env_dir is not None
            else None
        ),
        'runtime_user': runtime_config.runtime_user,
        'enabled_skill_names': (
            list(runtime_config.enabled_skill_names)
            if runtime_config.enabled_skill_names is not None
            else None
        ),
    }


def deserialize_runtime_config(payload: JSONDict) -> AgentRuntimeConfig:
    permissions_payload = payload.get('permissions')
    if not isinstance(permissions_payload, dict):
        permissions_payload = {}
    budget_payload = payload.get('budget_config')
    if not isinstance(budget_payload, dict):
        budget_payload = {}
    output_schema_payload = payload.get('output_schema')
    enabled_skill_names_payload = payload.get('enabled_skill_names')
    enabled_skill_names = (
        tuple(
            str(name)
            for name in enabled_skill_names_payload
            if str(name).strip()
        )
        if isinstance(enabled_skill_names_payload, list)
        else None
    )
    return AgentRuntimeConfig(
        cwd=Path(str(payload['cwd'])).resolve(),
        max_turns=int(payload.get('max_turns', DEFAULT_MAX_TURNS)),
        command_timeout_seconds=float(payload.get('command_timeout_seconds', 300.0)),
        max_output_chars=int(payload.get('max_output_chars', 50000)),
        stream_model_responses=bool(payload.get('stream_model_responses', False)),
        auto_snip_threshold_tokens=_optional_int(payload.get('auto_snip_threshold_tokens')),
        auto_compact_threshold_tokens=_optional_int(payload.get('auto_compact_threshold_tokens')),
        compact_preserve_messages=int(payload.get('compact_preserve_messages', 4)),
        permissions=AgentPermissions(
            allow_file_write=bool(permissions_payload.get('allow_file_write', False)),
            allow_shell_commands=bool(permissions_payload.get('allow_shell_commands', False)),
            allow_destructive_shell_commands=bool(permissions_payload.get('allow_destructive_shell_commands', False)),
        ),
        additional_working_directories=tuple(
            Path(str(path)).resolve()
            for path in payload.get('additional_working_directories', [])
        ),
        disable_claude_md_discovery=bool(payload.get('disable_claude_md_discovery', False)),
        budget_config=BudgetConfig(
            max_total_tokens=_optional_int(budget_payload.get('max_total_tokens')),
            max_input_tokens=_optional_int(budget_payload.get('max_input_tokens')),
            max_output_tokens=_optional_int(budget_payload.get('max_output_tokens')),
            max_reasoning_tokens=_optional_int(budget_payload.get('max_reasoning_tokens')),
            max_total_cost_usd=_optional_float(budget_payload.get('max_total_cost_usd')),
            max_tool_calls=_optional_int(budget_payload.get('max_tool_calls')),
            max_delegated_tasks=_optional_int(budget_payload.get('max_delegated_tasks')),
            max_model_calls=_optional_int(budget_payload.get('max_model_calls')),
            max_session_turns=_optional_int(budget_payload.get('max_session_turns')),
        ),
        output_schema=_deserialize_output_schema(output_schema_payload),
        session_directory=Path(str(payload.get('session_directory', DEFAULT_AGENT_SESSION_DIR))).resolve(),
        scratchpad_root=Path(str(payload.get('scratchpad_root', DEFAULT_SESSION_DIR / 'scratchpad'))).resolve(),
        python_env_dir=(
            Path(str(payload['python_env_dir'])).resolve()
            if payload.get('python_env_dir') is not None
            else None
        ),
        runtime_user=(
            str(payload['runtime_user'])
            if payload.get('runtime_user') is not None
            else None
        ),
        enabled_skill_names=enabled_skill_names,
    )


def usage_from_payload(payload: JSONDict | None) -> UsageStats:
    if not isinstance(payload, dict):
        return UsageStats()
    return UsageStats(
        input_tokens=_optional_int(payload.get('input_tokens')) or 0,
        output_tokens=_optional_int(payload.get('output_tokens')) or 0,
        cache_creation_input_tokens=_optional_int(payload.get('cache_creation_input_tokens')) or 0,
        cache_read_input_tokens=_optional_int(payload.get('cache_read_input_tokens')) or 0,
        reasoning_tokens=_optional_int(payload.get('reasoning_tokens')) or 0,
    )


def _deserialize_pricing(payload: Any) -> ModelPricing:
    if not isinstance(payload, dict):
        return ModelPricing()
    return ModelPricing(
        input_cost_per_million_tokens_usd=_optional_float(payload.get('input_cost_per_million_tokens_usd')) or 0.0,
        output_cost_per_million_tokens_usd=_optional_float(payload.get('output_cost_per_million_tokens_usd')) or 0.0,
        cache_creation_input_cost_per_million_tokens_usd=(
            _optional_float(payload.get('cache_creation_input_cost_per_million_tokens_usd'))
            or 0.0
        ),
        cache_read_input_cost_per_million_tokens_usd=(
            _optional_float(payload.get('cache_read_input_cost_per_million_tokens_usd'))
            or 0.0
        ),
    )


def _deserialize_output_schema(payload: Any) -> OutputSchemaConfig | None:
    if not isinstance(payload, dict):
        return None
    schema = payload.get('schema')
    if not isinstance(schema, dict):
        return None
    name = payload.get('name')
    if not isinstance(name, str) or not name:
        return None
    return OutputSchemaConfig(
        name=name,
        schema=dict(schema),
        strict=bool(payload.get('strict', False)),
    )


def _optional_int(value: Any) -> int | None:
    if value is None or isinstance(value, bool):
        return None
    if isinstance(value, int):
        return value
    try:
        return int(value)
    except (TypeError, ValueError):
        return None


def _optional_float(value: Any) -> float | None:
    if value is None or isinstance(value, bool):
        return None
    if isinstance(value, (int, float)):
        return float(value)
    try:
        return float(value)
    except (TypeError, ValueError):
        return None
