OpenSpace/openspace/services/session/restore.py

"""Session discovery and resume restoration.

OpenSpace keeps the project-scoped append-only transcript from
``session_storage.py`` as the resume source of truth.  This module is the
runtime-facing layer: lightweight listing, cross-project decisions, transcript
deserialization, cost/metadata/worktree recovery, and TodoWrite state hydration.
"""

from __future__ import annotations

import dataclasses
import json
import os
import shlex
import subprocess
from collections.abc import Sequence as SequenceABC
from dataclasses import dataclass, field
from datetime import datetime, timezone
from pathlib import Path
from typing import Any, Mapping, Sequence

from openspace.services.memory.paths import find_project_root
from openspace.services.session.recovery import (
    deserialize_for_resume as conversation_deserialize_for_resume,
)
from openspace.persistence.file_history import copy_file_history_for_resume
from openspace.services.session.storage import (
    METADATA_FILENAME,
    TRANSCRIPT_FILENAME,
    SessionLoadResult,
    SessionStorage,
    get_projects_dir,
    get_sessions_dir,
)
from openspace.tools.todo_tool import (
    TODO_WRITE_TOOL_ALIAS,
    TODO_WRITE_TOOL_NAME,
    TodoItem,
    normalize_todos,
    validate_todo_payload,
)
from openspace.utils.logging import Logger

logger = Logger.get_logger(__name__)

DEFAULT_DISCOVERY_LIMIT = 50
DEFAULT_PAGE_SIZE = 20
TAIL_READ_BYTES = 64 * 1024


@dataclass(slots=True)
class CrossProjectResumeResult:
    is_cross_project: bool
    is_same_repo_worktree: bool = False
    project_path: str | None = None
    command: str | None = None

    def to_dict(self) -> dict[str, Any]:
        payload = {
            "is_cross_project": self.is_cross_project,
            "isCrossProject": self.is_cross_project,
            "is_same_repo_worktree": self.is_same_repo_worktree,
            "isSameRepoWorktree": self.is_same_repo_worktree,
        }
        if self.project_path:
            payload["project_path"] = self.project_path
            payload["projectPath"] = self.project_path
        if self.command:
            payload["command"] = self.command
        return payload


@dataclass(slots=True)
class SessionSummary:
    session_id: str
    title: str
    first_prompt: str
    cwd: str
    project_root: str
    model: str | None
    created_at: str
    last_active_at: str
    message_count: int
    cost_usd: float | None
    tag: str | None
    mode: str
    agent_name: str | None
    agent_type: str | None
    session_dir: str
    transcript_path: str
    same_project: bool
    cross_project: bool
    cross_project_result: CrossProjectResumeResult = field(
        default_factory=lambda: CrossProjectResumeResult(False)
    )

    def to_dict(self) -> dict[str, Any]:
        return {
            "session_id": self.session_id,
            "title": self.title,
            "first_prompt": self.first_prompt,
            "preview": self.first_prompt,
            "cwd": self.cwd,
            "project_root": self.project_root,
            "project_path": self.project_root or self.cwd,
            "worktree_path": _worktree_path_from_summary(self),
            "model": self.model,
            "created_at": self.created_at,
            "updated_at": self.last_active_at,
            "last_active_at": self.last_active_at,
            "message_count": self.message_count,
            "cost": self.cost_usd,
            "cost_usd": self.cost_usd,
            "tag": self.tag,
            "mode": self.mode,
            "agent_name": self.agent_name,
            "agent_type": self.agent_type,
            "session_dir": self.session_dir,
            "transcript_path": self.transcript_path,
            "same_project": self.same_project,
            "cross_project": self.cross_project,
            "cross_project_result": self.cross_project_result.to_dict(),
        }


@dataclass(slots=True)
class SessionDiscoveryResult:
    sessions: list[SessionSummary]
    total: int
    page: int
    page_size: int
    has_more: bool

    def to_dict(self) -> dict[str, Any]:
        return {
            "sessions": [session.to_dict() for session in self.sessions],
            "total": self.total,
            "page": self.page,
            "page_size": self.page_size,
            "has_more": self.has_more,
        }


@dataclass(slots=True)
class ResumeDeserializationResult:
    messages: list[dict[str, Any]]
    turn_interruption_state: dict[str, Any]
    inserted_sentinel: bool = False
    inserted_continuation: bool = False


@dataclass(slots=True)
class RestoredSession:
    session_id: str
    session_record: dict[str, Any]
    title: str | None
    mode: str | None
    metadata: dict[str, Any]
    runtime: dict[str, Any]
    messages: list[dict[str, Any]]
    cost: dict[str, Any] | None
    cost_total: float | None
    agent: dict[str, Any] | None
    standalone_agent_context: dict[str, Any] | None
    worktree: dict[str, Any] | None
    file_history_snapshots: list[dict[str, Any]]
    content_replacements: list[dict[str, Any]]
    transcript_segments: list[dict[str, Any]]
    todo_state: dict[str, list[dict[str, str]]]
    turn_interruption_state: dict[str, Any]
    session_dir: str
    transcript_path: str

    def to_dict(self) -> dict[str, Any]:
        return dataclasses.asdict(self)


class CrossProjectResumeRequired(RuntimeError):
    """Raised when a session should be resumed from its original project."""

    def __init__(self, result: CrossProjectResumeResult) -> None:
        self.result = result
        super().__init__(
            result.command
            or f"Session belongs to a different project: {result.project_path}"
        )


async def discover_sessions(
    cwd: str | Path,
    *,
    limit: int = DEFAULT_DISCOVERY_LIMIT,
    page: int = 0,
    page_size: int = DEFAULT_PAGE_SIZE,
    all_projects: bool = False,
    same_repo: bool = True,
    config_home: str | Path | None = None,
) -> SessionDiscoveryResult:
    """Discover resumable sessions without loading full transcripts."""

    current_cwd = _resolve_path(cwd)
    current_project_root = str(find_project_root(current_cwd))
    worktree_paths = _git_worktree_paths(current_project_root)
    scan_all = bool(all_projects or same_repo)

    summaries_by_id: dict[str, SessionSummary] = {}
    for session_dir in _iter_session_dirs(
        current_project_root,
        config_home=config_home,
        all_projects=scan_all,
    ):
        summary = _summary_from_session_dir(
            session_dir,
            current_project_root=current_project_root,
            current_cwd=str(current_cwd),
            worktree_paths=worktree_paths,
        )
        if summary is None:
            continue

        include = all_projects or summary.same_project
        if not include and same_repo:
            include = summary.cross_project_result.is_same_repo_worktree
        if not include:
            continue

        existing = summaries_by_id.get(summary.session_id)
        if existing is None or _sort_timestamp(summary) >= _sort_timestamp(existing):
            summaries_by_id[summary.session_id] = summary

    summaries = sorted(
        summaries_by_id.values(),
        key=lambda item: (_sort_timestamp(item), item.session_id),
        reverse=True,
    )
    normalized_limit = max(0, int(limit or 0))
    if normalized_limit:
        summaries = summaries[:normalized_limit]

    normalized_page_size = max(1, int(page_size or DEFAULT_PAGE_SIZE))
    normalized_page = max(0, int(page or 0))
    start = normalized_page * normalized_page_size
    page_items = summaries[start : start + normalized_page_size]

    return SessionDiscoveryResult(
        sessions=page_items,
        total=len(summaries),
        page=normalized_page,
        page_size=normalized_page_size,
        has_more=start + normalized_page_size < len(summaries),
    )


async def restore_session(
    session_id: str,
    *,
    cwd: str | Path | None = None,
    fork: bool = False,
    allow_cross_project: bool = False,
    context: Any | None = None,
    config_home: str | Path | None = None,
) -> RestoredSession:
    """Restore a session from SessionStorage and hydrate runtime state."""

    current_cwd = _resolve_path(cwd or os.getcwd())
    current_project_root = str(find_project_root(current_cwd))

    if fork:
        source = SessionStorage.for_session(
            str(session_id),
            cwd=current_cwd,
            config_home=config_home,
            create=False,
        )
        loaded = source.load()
        metadata = dict(loaded.metadata)
        metadata["forked_from"] = str(session_id)
        metadata["cwd"] = str(current_cwd)
        metadata["project_root"] = current_project_root
        metadata["project_path"] = current_project_root
        storage = SessionStorage.create_new(
            cwd=current_cwd,
            model=_none_or_str(metadata.get("model")),
            config_home=config_home,
            metadata=metadata,
        )
        await storage.replace_messages(loaded.messages, metadata_patch=metadata)
        await copy_file_history_for_resume(source, storage, loaded.file_history_snapshots)
        loaded = storage.load()
        effective_session_id = storage.session_id
    else:
        storage = SessionStorage.for_session(
            str(session_id),
            cwd=current_cwd,
            config_home=config_home,
            create=False,
        )
        loaded = storage.load()
        effective_session_id = str(session_id)

    summary = _summary_from_session_dir(
        loaded.session_dir,
        current_project_root=current_project_root,
        current_cwd=str(current_cwd),
        worktree_paths=_git_worktree_paths(current_project_root),
    )
    cross_project = (
        summary.cross_project_result
        if summary is not None
        else check_cross_project_resume(
            _summary_from_loaded(loaded, current_project_root, str(current_cwd)),
            str(current_cwd),
            _git_worktree_paths(current_project_root),
        )
    )
    if (
        cross_project.is_cross_project
        and not cross_project.is_same_repo_worktree
        and not allow_cross_project
    ):
        raise CrossProjectResumeRequired(cross_project)

    deserialized = deserialize_messages_for_resume(loaded.messages)
    metadata = dict(loaded.metadata)
    metadata["session_id"] = effective_session_id
    metadata.setdefault("session_dir", str(loaded.session_dir))
    metadata.setdefault("transcript_path", str(loaded.transcript_path))

    runtime = dict(metadata.get("runtime") or {})
    runtime["session_id"] = effective_session_id
    if metadata.get("model"):
        runtime.setdefault("model", metadata.get("model"))

    cost_snapshot = _select_cost_snapshot(metadata, loaded)
    cost_total = _extract_cost_usd(cost_snapshot or runtime)
    if cost_total is not None:
        runtime["cost_usd"] = cost_total

    worktree = _restore_worktree_metadata(storage, metadata)
    worktree_data = _mapping(worktree)
    file_history = (
        _canonical_file_history_entries(metadata.get("file_history_snapshots"))
        or _canonical_file_history_entries(worktree_data.get("file_history_snapshots"))
        or _canonical_file_history_entries(loaded.file_history_snapshots)
    )
    content_replacements = (
        _dict_list(metadata.get("content_replacements"))
        or _dict_list(worktree_data.get("content_replacements"))
        or list(loaded.content_replacements)
    )
    if worktree is not None:
        if file_history and "file_history_snapshots" not in worktree:
            worktree["file_history_snapshots"] = file_history
        if content_replacements and "content_replacements" not in worktree:
            worktree["content_replacements"] = content_replacements

    todos = extract_todos_from_transcript(deserialized.messages)
    todo_key = _todo_key(context, effective_session_id)
    todo_state = {todo_key: todos}
    if context is not None:
        existing = getattr(context, "todo_state", None)
        if isinstance(existing, dict):
            existing[todo_key] = todos
        else:
            try:
                setattr(context, "todo_state", dict(todo_state))
            except Exception:
                pass

    runtime["todo_state"] = dict(todo_state)
    runtime["turn_interruption_state"] = dict(deserialized.turn_interruption_state)

    agent = _agent_payload(metadata)
    standalone_agent_context = _standalone_agent_context(metadata)
    record = _session_record(metadata, runtime, worktree, file_history, content_replacements)

    return RestoredSession(
        session_id=effective_session_id,
        session_record=record,
        title=_none_or_str(metadata.get("title")),
        mode=_none_or_str(metadata.get("mode")),
        metadata=dict(metadata.get("metadata") or {}),
        runtime=runtime,
        messages=deserialized.messages,
        cost=cost_snapshot,
        cost_total=cost_total,
        agent=agent,
        standalone_agent_context=standalone_agent_context,
        worktree=worktree,
        file_history_snapshots=file_history,
        content_replacements=content_replacements,
        transcript_segments=list(loaded.transcript_segments),
        todo_state=todo_state,
        turn_interruption_state=deserialized.turn_interruption_state,
        session_dir=str(loaded.session_dir),
        transcript_path=str(loaded.transcript_path),
    )


async def rewind_session(
    session_id: str,
    messages: Sequence[Mapping[str, Any]],
    *,
    cwd: str | Path | None = None,
    config_home: str | Path | None = None,
    model: str | None = None,
    metadata_patch: Mapping[str, Any] | None = None,
    cost: Mapping[str, Any] | None = None,
    allow_cross_project: bool = True,
    context: Any | None = None,
) -> RestoredSession:
    """Replace a session transcript and restore the resulting canonical session."""

    storage = SessionStorage.for_session(
        str(session_id),
        cwd=cwd,
        config_home=config_home,
        create=True,
    )
    patch = dict(metadata_patch or {})
    patch["session_id"] = str(session_id)
    patch["last_task_id"] = None
    patch["last_status"] = "rewound"
    runtime = dict(patch.get("runtime") or {})
    runtime.pop("active_task_id", None)
    runtime["phase"] = "rewound"
    runtime["session_id"] = str(session_id)
    patch["runtime"] = runtime
    if cost is not None:
        patch["cost"] = dict(cost)

    normalized_messages = [
        dict(message) for message in messages if isinstance(message, Mapping)
    ]
    await storage.replace_messages(
        normalized_messages,
        model=model,
        metadata_patch=patch,
    )
    return await restore_session(
        str(session_id),
        cwd=cwd,
        allow_cross_project=allow_cross_project,
        context=context,
        config_home=config_home,
    )


def deserialize_messages_for_resume(
    messages: Sequence[Mapping[str, Any]],
) -> ResumeDeserializationResult:
    """Make persisted OpenAI-shaped messages safe for resume."""

    recovered = conversation_deserialize_for_resume(messages)

    return ResumeDeserializationResult(
        messages=recovered.messages,
        turn_interruption_state=recovered.turn_interruption_state,
        inserted_sentinel=recovered.inserted_sentinel,
        inserted_continuation=recovered.inserted_continuation,
    )


def extract_todos_from_transcript(
    messages: Sequence[Mapping[str, Any]],
) -> list[dict[str, str]]:
    """Return the last valid TodoWrite payload from a transcript."""

    for message in reversed(list(messages)):
        if not isinstance(message, Mapping) or message.get("role") != "assistant":
            continue
        for payload in _iter_assistant_tool_payloads(message):
            name = _none_or_str(payload.get("name"))
            if name not in {TODO_WRITE_TOOL_NAME, TODO_WRITE_TOOL_ALIAS}:
                continue
            args = payload.get("input")
            if not isinstance(args, Mapping):
                return []
            validation_error = validate_todo_payload(args)
            if validation_error is not None:
                return []
            todos = normalize_todos(args.get("todos") or [])
            if todos and all(todo.status == "completed" for todo in todos):
                return []
            return [todo.to_dict() for todo in todos if isinstance(todo, TodoItem)]
    return []


def check_cross_project_resume(
    session: SessionSummary,
    current_cwd: str | Path,
    worktree_paths: Sequence[str] | None = None,
) -> CrossProjectResumeResult:
    """Return the cross-project resume decision for a session."""

    current_root = str(find_project_root(current_cwd))
    project_path = session.project_root or session.cwd
    if not project_path or _same_path(project_path, current_root) or _same_path(project_path, current_cwd):
        return CrossProjectResumeResult(False)

    normalized_worktrees = [str(Path(path).expanduser().resolve()) for path in worktree_paths or ()]
    try:
        resolved_project = str(Path(project_path).expanduser().resolve())
    except Exception:
        resolved_project = project_path
    for worktree in normalized_worktrees:
        if resolved_project == worktree or resolved_project.startswith(worktree + os.sep):
            return CrossProjectResumeResult(
                True,
                is_same_repo_worktree=True,
                project_path=project_path,
            )

    return CrossProjectResumeResult(
        True,
        is_same_repo_worktree=False,
        project_path=project_path,
        command=f"cd {shlex.quote(session.cwd or project_path)} && openspace --resume {shlex.quote(session.session_id)}",
    )


def _iter_session_dirs(
    cwd: str | Path,
    *,
    config_home: str | Path | None,
    all_projects: bool,
) -> list[Path]:
    if all_projects:
        projects_dir = get_projects_dir(config_home)
        if not projects_dir.exists():
            return []
        roots = [path / "sessions" for path in projects_dir.iterdir() if path.is_dir()]
    else:
        roots = [get_sessions_dir(cwd, config_home)]

    result: list[Path] = []
    for root in roots:
        if not root.exists():
            continue
        for candidate in root.iterdir():
            if not candidate.is_dir():
                continue
            if (candidate / TRANSCRIPT_FILENAME).exists() or (candidate / METADATA_FILENAME).exists():
                result.append(candidate)
    return result


def _summary_from_session_dir(
    session_dir: Path,
    *,
    current_project_root: str,
    current_cwd: str,
    worktree_paths: Sequence[str],
) -> SessionSummary | None:
    metadata = _read_json_object(session_dir / METADATA_FILENAME) or {}
    transcript = session_dir / TRANSCRIPT_FILENAME
    if not metadata and not transcript.exists():
        return None

    session_id = str(metadata.get("session_id") or session_dir.name)
    cwd = str(metadata.get("cwd") or metadata.get("project_path") or metadata.get("project_root") or "")
    project_root = str(metadata.get("project_root") or metadata.get("project_path") or cwd)
    first_prompt = _first_prompt(transcript)
    last_active = str(
        metadata.get("last_active_at")
        or metadata.get("updated_at")
        or _mtime_iso(transcript)
        or ""
    )
    summary = SessionSummary(
        session_id=session_id,
        title=str(metadata.get("title") or metadata.get("name") or first_prompt or session_id),
        first_prompt=first_prompt,
        cwd=cwd,
        project_root=project_root,
        model=_none_or_str(metadata.get("model") or _mapping(metadata.get("runtime")).get("model")),
        created_at=str(metadata.get("created_at") or ""),
        last_active_at=last_active,
        message_count=int(metadata.get("message_count") or _count_messages(transcript)),
        cost_usd=_extract_cost_usd(metadata.get("cost") or metadata.get("runtime")),
        tag=_none_or_str(metadata.get("tag")),
        mode=str(metadata.get("mode") or "normal"),
        agent_name=_none_or_str(metadata.get("agent_name") or _mapping(metadata.get("agent")).get("name")),
        agent_type=_none_or_str(metadata.get("agent_type") or _mapping(metadata.get("agent")).get("type")),
        session_dir=str(session_dir),
        transcript_path=str(transcript),
        same_project=_same_path(project_root, current_project_root) or _same_path(cwd, current_cwd),
        cross_project=False,
    )
    cross = check_cross_project_resume(summary, current_cwd, worktree_paths)
    summary.cross_project_result = cross
    summary.cross_project = cross.is_cross_project
    summary.same_project = not cross.is_cross_project
    return summary


def _summary_from_loaded(
    loaded: SessionLoadResult,
    current_project_root: str,
    current_cwd: str,
) -> SessionSummary:
    metadata = dict(loaded.metadata)
    return SessionSummary(
        session_id=loaded.session_id,
        title=str(metadata.get("title") or loaded.session_id),
        first_prompt="",
        cwd=str(metadata.get("cwd") or current_cwd),
        project_root=str(metadata.get("project_root") or metadata.get("project_path") or ""),
        model=_none_or_str(metadata.get("model")),
        created_at=str(metadata.get("created_at") or ""),
        last_active_at=str(metadata.get("last_active_at") or metadata.get("updated_at") or ""),
        message_count=len(loaded.messages),
        cost_usd=_extract_cost_usd(metadata.get("cost") or metadata.get("runtime")),
        tag=_none_or_str(metadata.get("tag")),
        mode=str(metadata.get("mode") or "normal"),
        agent_name=_none_or_str(metadata.get("agent_name")),
        agent_type=_none_or_str(metadata.get("agent_type")),
        session_dir=str(loaded.session_dir),
        transcript_path=str(loaded.transcript_path),
        same_project=_same_path(metadata.get("project_root"), current_project_root),
        cross_project=not _same_path(metadata.get("project_root"), current_project_root),
    )


def _select_cost_snapshot(
    metadata: Mapping[str, Any],
    loaded: SessionLoadResult,
) -> dict[str, Any] | None:
    cost = metadata.get("cost")
    if isinstance(cost, Mapping):
        return dict(cost)
    runtime = metadata.get("runtime")
    if isinstance(runtime, Mapping) and isinstance(runtime.get("cost"), Mapping):
        return dict(runtime["cost"])
    if not loaded.usage:
        return None
    return {
        "usage": list(loaded.usage),
        "total_cost": _extract_cost_usd(runtime) or 0.0,
    }


def _restore_worktree_metadata(
    storage: SessionStorage,
    metadata: dict[str, Any],
) -> dict[str, Any] | None:
    worktree = metadata.get("worktree")
    if not isinstance(worktree, Mapping):
        return None
    restored = dict(worktree)
    worktree_path = (
        restored.get("worktree_path")
        or restored.get("worktreePath")
        or restored.get("workspace_dir")
        or restored.get("path")
    )
    if isinstance(worktree_path, str) and worktree_path:
        if Path(worktree_path).expanduser().exists():
            restored.setdefault("workspace_dir", worktree_path)
        else:
            logger.warning("Resume worktree path no longer exists: %s", worktree_path)
            try:
                storage.update_metadata({"worktree": None}, append=True)
            except Exception:
                logger.debug("Failed to clear stale worktree metadata", exc_info=True)
            return None
    return restored or None


def _session_record(
    metadata: Mapping[str, Any],
    runtime: Mapping[str, Any],
    worktree: Mapping[str, Any] | None,
    file_history: list[dict[str, Any]],
    content_replacements: list[dict[str, Any]],
) -> dict[str, Any]:
    project_path = str(metadata.get("project_root") or metadata.get("project_path") or metadata.get("cwd") or "")
    workspace_dir = (
        _mapping(worktree).get("workspace_dir")
        or _mapping(worktree).get("worktree_path")
        or metadata.get("cwd")
        or project_path
    )
    record = dict(metadata)
    record.update(
        {
            "session_id": metadata.get("session_id"),
            "project_path": project_path,
            "worktree_path": _mapping(worktree).get("worktree_path") or workspace_dir,
            "workspace_dir": workspace_dir,
            "runtime": dict(runtime),
            "worktree": dict(worktree or {}),
            "file_history_snapshots": list(file_history),
            "content_replacements": list(content_replacements),
        }
    )
    return record


def _iter_assistant_tool_payloads(message: Mapping[str, Any]) -> list[dict[str, Any]]:
    payloads: list[dict[str, Any]] = []
    tool_calls = message.get("tool_calls")
    if isinstance(tool_calls, SequenceABC) and not isinstance(tool_calls, (str, bytes)):
        for call in reversed(list(tool_calls)):
            if not isinstance(call, Mapping):
                continue
            name = _tool_call_name(call)
            args = _tool_call_arguments(call)
            payloads.append({"name": name, "input": args})

    content = message.get("content")
    if isinstance(content, SequenceABC) and not isinstance(content, (str, bytes)):
        for block in reversed(list(content)):
            if not isinstance(block, Mapping) or block.get("type") != "tool_use":
                continue
            payloads.append(
                {
                    "name": _none_or_str(block.get("name")),
                    "input": block.get("input") if isinstance(block.get("input"), Mapping) else {},
                }
            )
    return payloads


def _tool_call_name(call: Mapping[str, Any]) -> str | None:
    function = call.get("function")
    if isinstance(function, Mapping):
        return _none_or_str(function.get("name"))
    return _none_or_str(call.get("name"))


def _tool_call_arguments(call: Mapping[str, Any]) -> dict[str, Any]:
    function = call.get("function")
    raw: Any
    if isinstance(function, Mapping):
        raw = function.get("arguments")
    else:
        raw = call.get("arguments") or call.get("input")
    if isinstance(raw, Mapping):
        return dict(raw)
    if isinstance(raw, str):
        try:
            parsed = json.loads(raw)
        except json.JSONDecodeError:
            return {}
        return dict(parsed) if isinstance(parsed, Mapping) else {}
    return {}


def _todo_key(context: Any | None, session_id: str) -> str:
    if context is not None:
        agent_id = getattr(context, "agent_id", None)
        if agent_id:
            return str(agent_id)
        context_session = getattr(context, "session_id", None)
        if context_session:
            return str(context_session)
    return str(session_id or "primary")


def _agent_payload(metadata: Mapping[str, Any]) -> dict[str, Any] | None:
    agent = metadata.get("agent")
    if isinstance(agent, Mapping) and agent:
        return dict(agent)
    agent_type = _none_or_str(metadata.get("agent_type"))
    agent_name = _none_or_str(metadata.get("agent_name"))
    if not agent_type and not agent_name:
        return None
    return {"type": agent_type, "name": agent_name}


def _standalone_agent_context(metadata: Mapping[str, Any]) -> dict[str, Any] | None:
    context = metadata.get("standalone_agent_context")
    if isinstance(context, Mapping) and context:
        return dict(context)
    name = _none_or_str(metadata.get("agent_name"))
    color = _none_or_str(metadata.get("agent_color"))
    if not name and not color:
        return None
    payload: dict[str, Any] = {"name": name or ""}
    if color and color != "default":
        payload["color"] = color
    return payload


def _git_worktree_paths(project_root: str) -> list[str]:
    root = Path(project_root).expanduser()
    if not root.exists():
        return [str(root)]
    try:
        proc = subprocess.run(
            ["git", "-C", str(root), "worktree", "list", "--porcelain"],
            check=False,
            stdout=subprocess.PIPE,
            stderr=subprocess.DEVNULL,
            text=True,
            timeout=2,
        )
    except Exception:
        return [str(root.resolve())]
    if proc.returncode != 0:
        return [str(root.resolve())]
    paths: list[str] = []
    for line in proc.stdout.splitlines():
        if line.startswith("worktree "):
            paths.append(str(Path(line[len("worktree ") :]).expanduser().resolve()))
    return paths or [str(root.resolve())]


def _first_prompt(path: Path) -> str:
    for message in _iter_transcript_messages(path, max_lines=300):
        if message.get("role") != "user":
            continue
        content = message.get("content")
        text = _content_text(content)
        if text:
            return text[:240]
    return ""


def _iter_transcript_messages(path: Path, *, max_lines: int | None = None) -> list[dict[str, Any]]:
    if not path.exists():
        return []
    messages: list[dict[str, Any]] = []
    try:
        with path.open("r", encoding="utf-8") as handle:
            for index, line in enumerate(handle):
                if max_lines is not None and index >= max_lines:
                    break
                try:
                    raw = json.loads(line)
                except json.JSONDecodeError:
                    continue
                message = raw.get("message") if isinstance(raw, Mapping) else None
                if isinstance(message, Mapping):
                    messages.append(dict(message))
                elif isinstance(raw, Mapping) and "role" in raw:
                    messages.append(dict(raw))
    except OSError:
        return []
    return messages


def _count_messages(path: Path) -> int:
    if not path.exists():
        return 0
    count = 0
    try:
        with path.open("r", encoding="utf-8") as handle:
            for line in handle:
                try:
                    raw = json.loads(line)
                except json.JSONDecodeError:
                    continue
                if isinstance(raw, Mapping) and (
                    raw.get("type") == "message" or "role" in raw
                ):
                    count += 1
    except OSError:
        return 0
    return count


def _content_text(content: Any) -> str:
    if isinstance(content, str):
        return content.strip()
    if isinstance(content, SequenceABC) and not isinstance(content, (str, bytes)):
        parts: list[str] = []
        for block in content:
            if isinstance(block, str):
                parts.append(block)
            elif isinstance(block, Mapping):
                if isinstance(block.get("text"), str):
                    parts.append(block["text"])
                elif isinstance(block.get("content"), str):
                    parts.append(block["content"])
        return "\n".join(part.strip() for part in parts if part and part.strip())
    return ""


def _read_json_object(path: Path) -> dict[str, Any] | None:
    try:
        raw = json.loads(path.read_text(encoding="utf-8"))
    except (OSError, json.JSONDecodeError):
        return None
    return dict(raw) if isinstance(raw, Mapping) else None


def _mtime_iso(path: Path) -> str:
    try:
        return datetime.fromtimestamp(path.stat().st_mtime, timezone.utc).isoformat()
    except OSError:
        return ""


def _sort_timestamp(summary: SessionSummary) -> float:
    try:
        return datetime.fromisoformat(summary.last_active_at).timestamp()
    except (TypeError, ValueError):
        try:
            return Path(summary.session_dir).stat().st_mtime
        except OSError:
            return 0.0


def _extract_cost_usd(raw: Any) -> float | None:
    if not isinstance(raw, Mapping):
        return None
    for key in ("total_cost", "totalCostUSD", "cost_usd", "cost"):
        value = raw.get(key)
        if isinstance(value, (int, float)):
            return float(value)
    return None


def _same_path(left: Any, right: Any) -> bool:
    if not left or not right:
        return False
    try:
        return Path(str(left)).expanduser().resolve() == Path(str(right)).expanduser().resolve()
    except Exception:
        return str(left) == str(right)


def _resolve_path(path: str | Path) -> Path:
    return Path(path).expanduser().resolve()


def _mapping(value: Any) -> dict[str, Any]:
    return dict(value) if isinstance(value, Mapping) else {}


def _dict_list(value: Any) -> list[dict[str, Any]]:
    if not isinstance(value, SequenceABC) or isinstance(value, (str, bytes)):
        return []
    return [dict(item) for item in value if isinstance(item, Mapping)]


def _canonical_file_history_entries(value: Any) -> list[dict[str, Any]]:
    return [
        item
        for item in _dict_list(value)
        if _is_canonical_file_history_snapshot(item)
    ]


def _is_canonical_file_history_snapshot(value: Mapping[str, Any]) -> bool:
    raw = value.get("snapshot") if isinstance(value.get("snapshot"), Mapping) else value
    if not isinstance(raw, Mapping):
        return False
    backups = raw.get("tracked_file_backups")
    return isinstance(backups, Mapping)


def _none_or_str(value: Any) -> str | None:
    if value is None:
        return None
    text = str(value)
    return text if text else None


def _worktree_path_from_summary(summary: SessionSummary) -> str:
    return summary.cwd or summary.project_root


__all__ = [
    "CrossProjectResumeRequired",
    "CrossProjectResumeResult",
    "RestoredSession",
    "ResumeDeserializationResult",
    "SessionDiscoveryResult",
    "SessionSummary",
    "check_cross_project_resume",
    "deserialize_messages_for_resume",
    "discover_sessions",
    "extract_todos_from_transcript",
    "restore_session",
]