Implement `/resume` command for CLI (#233)

* Add /resume command to CLI

* Refine /resume feature

* Refine /resume implementation

* Address review comments

---------

Co-authored-by: lewtun <lewis.c.tunstall@gmail.com>

agent/core/agent_loop.py

@@ -7,6 +7,7 @@ import json
 import logging
 import time
 from dataclasses import dataclass, field
+from pathlib import Path
 from typing import Any
 
 from litellm import (
@@ -29,7 +30,7 @@ from agent.core.doom_loop import check_for_doom_loop
 from agent.core.hub_artifacts import start_session_artifact_collection_task
 from agent.core.llm_params import _resolve_llm_params
 from agent.core.prompt_caching import with_prompt_caching
-from agent.core.session import Event, OpType, Session
+from agent.core.session import DEFAULT_SESSION_LOG_DIR, Event, OpType, Session
 from agent.core.tools import ToolRouter
 from agent.tools.jobs_tool import CPU_FLAVORS
 from agent.tools.sandbox_tool import DEFAULT_CPU_SANDBOX_HARDWARE
@@ -1667,6 +1668,20 @@ class Handlers:
             logger.warning("Undo: no user message found to remove")
         await session.send_event(Event(event_type="undo_complete"))
 
+    @staticmethod
+    async def resume(session: Session, path: str) -> None:
+        """Reload context from a saved session log into the active session."""
+        from agent.core.session_resume import restore_session_from_log
+
+        try:
+            result = restore_session_from_log(session, Path(path))
+        except Exception as e:
+            await session.send_event(
+                Event(event_type="error", data={"error": f"Resume failed: {e}"})
+            )
+            return
+        await session.send_event(Event(event_type="resume_complete", data=result))
+
     @staticmethod
     async def exec_approval(session: Session, approvals: list[dict]) -> None:
         """Handle batch job execution approval"""
@@ -1953,6 +1968,16 @@ async def process_submission(session: Session, submission) -> bool:
         await Handlers.undo(session)
         return True
 
+    if op.op_type == OpType.RESUME:
+        path = op.data.get("path") if op.data else None
+        if path:
+            await Handlers.resume(session, path)
+        else:
+            await session.send_event(
+                Event(event_type="error", data={"error": "Resume requires a path"})
+            )
+        return True
+
     if op.op_type == OpType.EXEC_APPROVAL:
         approvals = op.data.get("approvals", []) if op.data else []
         await Handlers.exec_approval(session, approvals)
@@ -2007,7 +2032,7 @@ async def submission_loop(
     # to publish to the user's HF dataset gets a fresh attempt on next run.
     if config and config.save_sessions:
         Session.retry_failed_uploads_detached(
-            directory="session_logs",
+            directory=str(DEFAULT_SESSION_LOG_DIR),
             repo_id=config.session_dataset_repo,
             personal_repo_id=session._personal_trace_repo_id(),
         )

agent/core/session.py

@@ -21,6 +21,8 @@ logger = logging.getLogger(__name__)
 _DEFAULT_MAX_TOKENS = 200_000
 _TURN_COMPLETE_NOTIFICATION_CHARS = 39000
 
+DEFAULT_SESSION_LOG_DIR = Path("session_logs")
+
 
 def _get_max_tokens_safe(model_name: str) -> int:
     """Return the max input-context tokens for a model.
@@ -60,6 +62,7 @@ class OpType(Enum):
     INTERRUPT = "interrupt"
     UNDO = "undo"
     COMPACT = "compact"
+    RESUME = "resume"
     SHUTDOWN = "shutdown"
 
 
@@ -418,7 +421,7 @@ class Session:
 
     def save_trajectory_local(
         self,
-        directory: str = "session_logs",
+        directory: str = str(DEFAULT_SESSION_LOG_DIR),
         upload_status: str = "pending",
         dataset_url: Optional[str] = None,
     ) -> Optional[str]:
@@ -613,7 +616,7 @@ class Session:
 
     @staticmethod
     def retry_failed_uploads_detached(
-        directory: str = "session_logs",
+        directory: str = str(DEFAULT_SESSION_LOG_DIR),
         repo_id: Optional[str] = None,
         *,
         personal_repo_id: Optional[str] = None,

agent/core/session_resume.py

@@ -0,0 +1,287 @@
+"""Reload a previously saved session log into the active CLI session."""
+
+from __future__ import annotations
+
+import json
+import logging
+import re
+from dataclasses import dataclass
+from datetime import datetime
+from pathlib import Path
+from typing import Any
+
+from litellm import Message
+
+from agent.core.model_switcher import is_valid_model_id
+from agent.core.session import DEFAULT_SESSION_LOG_DIR
+
+logger = logging.getLogger(__name__)
+
+_REDACTED_MARKER = re.compile(r"\[REDACTED_[A-Z_]+\]")
+
+
+@dataclass
+class SessionLogEntry:
+    """Metadata for a locally saved session log."""
+
+    path: Path
+    session_id: str
+    session_start_time: str | None
+    session_end_time: str | None
+    model_name: str | None
+    message_count: int
+    preview: str
+    mtime: float
+
+
+def _message_preview(content: Any, max_chars: int = 72) -> str:
+    """Return a one-line preview for string or OpenAI-style block content."""
+    if isinstance(content, str):
+        text = content
+    elif isinstance(content, list):
+        parts: list[str] = []
+        for block in content:
+            if isinstance(block, dict):
+                value = block.get("text") or block.get("content")
+                if isinstance(value, str):
+                    parts.append(value)
+            elif isinstance(block, str):
+                parts.append(block)
+        text = " ".join(parts)
+    else:
+        text = ""
+    text = " ".join(text.split())
+    if len(text) > max_chars:
+        return text[: max_chars - 1].rstrip() + "…"
+    return text
+
+
+def _first_user_preview(messages: list[Any]) -> str:
+    for raw in messages:
+        if isinstance(raw, dict) and raw.get("role") == "user":
+            preview = _message_preview(raw.get("content"))
+            if preview:
+                return preview
+    return "(no user prompt preview)"
+
+
+def list_session_logs(
+    directory: Path = DEFAULT_SESSION_LOG_DIR,
+) -> list[SessionLogEntry]:
+    """Return readable session logs under ``directory``, newest first."""
+    if not directory.exists():
+        return []
+
+    entries: list[SessionLogEntry] = []
+    for path in directory.glob("*.json"):
+        try:
+            with open(path) as f:
+                data = json.load(f)
+        except Exception:
+            continue
+
+        messages = data.get("messages") or []
+        if not isinstance(messages, list):
+            continue
+
+        session_id = data.get("session_id")
+        if not isinstance(session_id, str) or not session_id:
+            session_id = path.stem
+
+        stat = path.stat()
+        entries.append(
+            SessionLogEntry(
+                path=path,
+                session_id=session_id,
+                session_start_time=data.get("session_start_time"),
+                session_end_time=data.get("session_end_time"),
+                model_name=data.get("model_name"),
+                message_count=len(messages),
+                preview=_first_user_preview(messages),
+                mtime=stat.st_mtime,
+            )
+        )
+
+    entries.sort(key=lambda item: item.mtime, reverse=True)
+    return entries
+
+
+def format_session_log_entry(index: int, entry: SessionLogEntry) -> str:
+    timestamp = entry.session_end_time or entry.session_start_time
+    label = "unknown time"
+    if isinstance(timestamp, str) and timestamp:
+        try:
+            label = datetime.fromisoformat(timestamp).strftime("%Y-%m-%d %H:%M")
+        except ValueError:
+            label = timestamp[:16]
+    short_id = entry.session_id[:8]
+    model = entry.model_name or "unknown model"
+    return (
+        f"{index:>2}. {label}  {short_id}  "
+        f"{entry.message_count} msgs  {model}\n"
+        f"    {entry.preview}"
+    )
+
+
+def resolve_session_log_arg(
+    arg: str,
+    entries: list[SessionLogEntry],
+    directory: Path = DEFAULT_SESSION_LOG_DIR,
+) -> Path | None:
+    """Resolve ``/resume <arg>`` as index, path, filename, or session id prefix."""
+    value = arg.strip()
+    if not value:
+        return None
+
+    if value.isdigit():
+        idx = int(value)
+        if 1 <= idx <= len(entries):
+            return entries[idx - 1].path
+
+    candidate = Path(value).expanduser()
+    candidates = [candidate]
+    if not candidate.is_absolute():
+        candidates.append(directory / candidate)
+        if candidate.suffix != ".json":
+            candidates.append(directory / f"{value}.json")
+
+    for path in candidates:
+        if path.exists() and path.is_file():
+            return path
+
+    matches = [
+        entry.path
+        for entry in entries
+        if entry.session_id.startswith(value) or entry.path.name.startswith(value)
+    ]
+    if len(matches) == 1:
+        return matches[0]
+    return None
+
+
+def _turn_count_from_messages(messages: list[Any]) -> int:
+    return sum(
+        1 for raw in messages if isinstance(raw, dict) and raw.get("role") == "user"
+    )
+
+
+def _has_redacted_content(messages: list[Any]) -> bool:
+    """Whether any message body contains a ``[REDACTED_*]`` marker."""
+    for raw in messages:
+        if not isinstance(raw, dict):
+            continue
+        content = raw.get("content")
+        if isinstance(content, str) and _REDACTED_MARKER.search(content):
+            return True
+        if isinstance(content, list):
+            for block in content:
+                if isinstance(block, dict):
+                    text = block.get("text") or block.get("content")
+                    if isinstance(text, str) and _REDACTED_MARKER.search(text):
+                        return True
+    return False
+
+
+def restore_session_from_log(session: Any, path: Path) -> dict[str, Any]:
+    """Replace the active session context with messages from ``path``.
+
+    Continues the saved session (reusing its id and on-disk save path) when
+    the log's ``user_id`` matches the current session, and forks otherwise:
+    the caller's session id stays put and future heartbeat saves go to a
+    fresh file rather than overwriting the source log.
+
+    Returns metadata for the ``resume_complete`` event.
+    """
+    with open(path) as f:
+        data = json.load(f)
+
+    raw_messages = data.get("messages")
+    if not isinstance(raw_messages, list):
+        raise ValueError("Selected log does not contain a messages array")
+
+    restored_messages: list[Message] = []
+    dropped_count = 0
+    for raw in raw_messages:
+        if not isinstance(raw, dict) or raw.get("role") == "system":
+            continue
+        try:
+            restored_messages.append(Message.model_validate(raw))
+        except Exception as e:
+            dropped_count += 1
+            logger.warning("Dropping malformed message from %s: %s", path, e)
+
+    if not restored_messages:
+        raise ValueError("Selected log has no restorable non-system messages")
+
+    cm = session.context_manager
+    system_msg = cm.items[0] if cm.items and cm.items[0].role == "system" else None
+    cm.items = ([system_msg] if system_msg else []) + restored_messages
+
+    # Validate the saved model id before switching. ``update_model`` doesn't
+    # check availability; an unrecognised id silently sticks and the next LLM
+    # call fails with a cryptic routing error. Logs from a different
+    # deployment, an older catalog, or a removed model land here.
+    saved_model = data.get("model_name")
+    invalid_saved_model: str | None = None
+    if isinstance(saved_model, str) and saved_model:
+        if is_valid_model_id(saved_model):
+            session.update_model(saved_model)
+        else:
+            invalid_saved_model = saved_model
+            logger.warning(
+                "Saved log model %r failed format validation; keeping %r",
+                saved_model,
+                session.config.model_name,
+            )
+
+    cm._recompute_usage(session.config.model_name)
+
+    saved_session_id = data.get("session_id")
+    saved_user_id = data.get("user_id")
+    is_continuation = saved_user_id == session.user_id
+
+    if is_continuation:
+        if isinstance(saved_session_id, str) and saved_session_id:
+            session.session_id = saved_session_id
+        session.session_start_time = (
+            data.get("session_start_time") or session.session_start_time
+        )
+
+    # Always fork the on-disk save path. The source log is treated as an
+    # immutable snapshot: ``logged_events`` is reset to a single
+    # ``resumed_from`` marker below for cost accounting, so reusing the
+    # source path would let the next heartbeat save destroy the original
+    # ``llm_call``/event history on disk. The next save will pick a fresh
+    # filename instead.
+    session._local_save_path = None
+
+    saved_event_count = (
+        len(data.get("events", [])) if isinstance(data.get("events"), list) else 0
+    )
+    session.logged_events = [
+        {
+            "timestamp": datetime.now().isoformat(),
+            "event_type": "resumed_from",
+            "data": {
+                "path": str(path),
+                "original_session_id": (
+                    saved_session_id if isinstance(saved_session_id, str) else None
+                ),
+                "original_event_count": saved_event_count,
+                "forked": not is_continuation,
+            },
+        }
+    ]
+    session.turn_count = _turn_count_from_messages(raw_messages)
+    session.last_auto_save_turn = session.turn_count
+    session.pending_approval = None
+
+    return {
+        "path": str(path),
+        "restored_count": len(restored_messages),
+        "dropped_count": dropped_count,
+        "model_name": session.config.model_name,
+        "invalid_saved_model": invalid_saved_model,
+        "forked": not is_continuation,
+        "had_redacted_content": _has_redacted_content(raw_messages),
+    }

agent/main.py

@@ -9,6 +9,7 @@ Supports two modes:
 import argparse
 import asyncio
 import json
+import logging
 import os
 import signal
 import sys
@@ -55,6 +56,7 @@ litellm.drop_params = True
 litellm.suppress_debug_info = True
 
 CLI_CONFIG_PATH = Path(__file__).parent.parent / "configs" / "cli_agent_config.json"
+logger = logging.getLogger(__name__)
 
 
 def _is_scheduled_hf_job_tool(tool_info: dict[str, Any]) -> bool:
@@ -368,6 +370,46 @@ async def event_listener(
             elif event.event_type == "undo_complete":
                 console.print("[dim]Undone.[/dim]")
                 turn_complete_event.set()
+            elif event.event_type == "resume_complete":
+                data = event.data or {}
+                path = data.get("path", "?")
+                count = data.get("restored_count", 0)
+                dropped = int(data.get("dropped_count", 0) or 0)
+                model = data.get("model_name", "?")
+                invalid_model = data.get("invalid_saved_model")
+                forked = bool(data.get("forked", False))
+                redacted = bool(data.get("had_redacted_content", False))
+                verb = "Forked from" if forked else "Resumed"
+                console.print(
+                    f"[green]{verb}[/green] {path} "
+                    f"([cyan]{count}[/cyan] messages, "
+                    f"model [cyan]{model}[/cyan])."
+                )
+                if dropped:
+                    console.print(
+                        f"[yellow]Warning:[/yellow] dropped {dropped} "
+                        "malformed message(s) while restoring — surrounding "
+                        "tool-call alignment may be off."
+                    )
+                if invalid_model:
+                    console.print(
+                        f"[yellow]Warning:[/yellow] saved model id "
+                        f"[cyan]{invalid_model}[/cyan] failed validation; "
+                        f"kept current model [cyan]{model}[/cyan]."
+                    )
+                if forked:
+                    console.print(
+                        "[dim]Saved log belongs to a different user — kept "
+                        "current session id; future saves go to a fresh file.[/dim]"
+                    )
+                if redacted:
+                    console.print(
+                        "[yellow]Note:[/yellow] tokens/secrets in restored "
+                        "messages were scrubbed at save time. Your live tokens "
+                        "are used for this session; [REDACTED_*] markers in "
+                        "past messages are not re-injected."
+                    )
+                turn_complete_event.set()
             elif event.event_type == "tool_log":
                 tool = event.data.get("tool", "") if event.data else ""
                 log = event.data.get("log", "") if event.data else ""
@@ -739,12 +781,69 @@ async def get_user_input(prompt_session: PromptSession) -> str:
 # Slash commands are defined in terminal_display
 
 
+async def _resume_picker(
+    arg: str,
+    prompt_session: PromptSession | None,
+) -> Path | None:
+    """Resolve a session log path via ``arg`` or interactive selection.
+
+    Returns ``None`` if the user cancels, no logs exist, or the argument
+    matches nothing — already prints the explanation in those cases.
+    """
+    from agent.core.session_resume import (
+        format_session_log_entry,
+        list_session_logs,
+        resolve_session_log_arg,
+    )
+    from agent.core.session import DEFAULT_SESSION_LOG_DIR
+
+    console = get_console()
+    directory = DEFAULT_SESSION_LOG_DIR
+    entries = list_session_logs(directory)
+    if not entries:
+        console.print(f"[yellow]No session logs found in ./{directory}.[/yellow]")
+        return None
+
+    if arg:
+        selected = resolve_session_log_arg(arg, entries, directory)
+        if selected is None:
+            console.print(f"[bold red]No matching session log:[/bold red] {arg}")
+        return selected
+
+    console.print()
+    console.print("[bold]Saved sessions[/bold]")
+    for index, entry in enumerate(entries, start=1):
+        console.print(format_session_log_entry(index, entry))
+    console.print()
+
+    if prompt_session is None:
+        console.print("[yellow]Cannot prompt for a selection here.[/yellow]")
+        return None
+
+    try:
+        choice = await prompt_session.prompt_async(
+            "Select session number (blank to cancel): "
+        )
+    except (EOFError, KeyboardInterrupt):
+        console.print("[dim]Resume cancelled.[/dim]")
+        return None
+    choice = choice.strip()
+    if not choice:
+        console.print("[dim]Resume cancelled.[/dim]")
+        return None
+    selected = resolve_session_log_arg(choice, entries, directory)
+    if selected is None:
+        console.print(f"[bold red]Invalid selection:[/bold red] {choice}")
+    return selected
+
+
 async def _handle_slash_command(
     cmd: str,
     config,
     session_holder: list,
     submission_queue: asyncio.Queue,
     submission_id: list[int],
+    prompt_session: PromptSession | None = None,
 ) -> Submission | None:
     """
     Handle a slash command. Returns a Submission to enqueue, or None if
@@ -775,6 +874,24 @@ async def _handle_slash_command(
             operation=Operation(op_type=OpType.COMPACT),
         )
 
+    if command == "/resume":
+        session = session_holder[0] if session_holder else None
+        if session is None:
+            get_console().print(
+                "[bold red]No active session to restore into.[/bold red]"
+            )
+            return None
+        selected_path = await _resume_picker(arg, prompt_session)
+        if selected_path is None:
+            return None
+        submission_id[0] += 1
+        return Submission(
+            id=f"sub_{submission_id[0]}",
+            operation=Operation(
+                op_type=OpType.RESUME, data={"path": str(selected_path)}
+            ),
+        )
+
     if command == "/model":
         console = get_console()
         if not arg:
@@ -1136,6 +1253,7 @@ async def main(model: str | None = None):
                     session_holder,
                     submission_queue,
                     submission_id,
+                    prompt_session,
                 )
                 if sub is None:
                     # Command handled locally, loop back for input

agent/utils/terminal_display.py

@@ -451,6 +451,7 @@ HELP_TEXT = f"""\
 {_I}  [cyan]/help[/cyan]            Show this help
 {_I}  [cyan]/undo[/cyan]            Undo last turn
 {_I}  [cyan]/compact[/cyan]         Compact context window
+{_I}  [cyan]/resume[/cyan] [index|id|path] Pick up from a log in ./session_logs
 {_I}  [cyan]/model[/cyan] [id]      Show available models or switch
 {_I}  [cyan]/effort[/cyan] [level]  Reasoning effort (minimal|low|medium|high|xhigh|max|off)
 {_I}  [cyan]/yolo[/cyan]            Toggle auto-approve mode

tests/unit/test_session_resume.py

@@ -0,0 +1,382 @@
+"""Tests for ``agent.core.session_resume``."""
+
+import json
+import os
+import time
+from pathlib import Path
+from types import SimpleNamespace
+
+from litellm import Message
+
+from agent.core import session_resume
+
+
+def _write_session_log(
+    directory: Path,
+    name: str,
+    *,
+    session_id: str,
+    content: str,
+    mtime: float,
+    user_id: str | None = "user-a",
+    extra_messages: list[dict] | None = None,
+    events: list[dict] | None = None,
+) -> Path:
+    directory.mkdir(exist_ok=True)
+    path = directory / name
+    payload = {
+        "session_id": session_id,
+        "user_id": user_id,
+        "session_start_time": "2026-01-01T00:00:00",
+        "session_end_time": "2026-01-01T00:05:00",
+        "model_name": "openai/gpt-5.5",
+        "messages": [
+            {"role": "system", "content": "old system"},
+            {"role": "user", "content": content},
+            *(extra_messages or []),
+        ],
+        "events": events
+        if events is not None
+        else [{"event_type": "turn_complete", "data": {}}],
+    }
+    path.write_text(json.dumps(payload))
+    os.utime(path, (mtime, mtime))
+    return path
+
+
+class _FakeContext:
+    def __init__(self) -> None:
+        self.items = [Message(role="system", content="current system")]
+        self.running_context_usage = 0
+        self.recompute_calls: list[str] = []
+
+    def _recompute_usage(self, model_name: str) -> None:
+        self.recompute_calls.append(model_name)
+        self.running_context_usage = 123
+
+
+class _FakeSession:
+    def __init__(self, *, user_id: str | None = "user-a") -> None:
+        self.context_manager = _FakeContext()
+        self.config = SimpleNamespace(model_name="moonshotai/Kimi-K2.6")
+        self.session_id = "current-session"
+        self.session_start_time = "2026-01-02T00:00:00"
+        self.user_id = user_id
+        self.logged_events: list[dict] = []
+        self._local_save_path: str | None = None
+        self.turn_count = 0
+        self.last_auto_save_turn = 0
+        self.pending_approval: dict | None = {"tool_calls": ["pending"]}
+
+    def update_model(self, model_name: str) -> None:
+        self.config.model_name = model_name
+
+
+def test_session_log_listing_newest_first(tmp_path):
+    log_dir = tmp_path / "session_logs"
+    older = _write_session_log(
+        log_dir,
+        "older.json",
+        session_id="older-session",
+        content="older prompt",
+        mtime=time.time() - 10,
+    )
+    newer = _write_session_log(
+        log_dir,
+        "newer.json",
+        session_id="newer-session",
+        content="newer prompt",
+        mtime=time.time(),
+    )
+
+    entries = session_resume.list_session_logs(log_dir)
+
+    assert [entry.path for entry in entries] == [newer, older]
+    assert entries[0].session_id == "newer-session"
+    assert entries[0].preview == "newer prompt"
+
+
+def test_restore_continues_when_user_id_matches(tmp_path):
+    log_dir = tmp_path / "session_logs"
+    path = _write_session_log(
+        log_dir,
+        "session.json",
+        session_id="saved-session",
+        content="continue this work",
+        mtime=time.time(),
+        user_id="user-a",
+    )
+
+    session = _FakeSession(user_id="user-a")
+
+    result = session_resume.restore_session_from_log(session, path)
+
+    assert result["restored_count"] == 1
+    assert result["dropped_count"] == 0
+    assert result["forked"] is False
+    assert result["model_name"] == "openai/gpt-5.5"
+    assert result["had_redacted_content"] is False
+    assert result["invalid_saved_model"] is None
+    assert session.config.model_name == "openai/gpt-5.5"
+    assert session.session_id == "saved-session"
+    # Source log path is never reused: future heartbeat saves write to a
+    # fresh file so the snapshot stays intact (regression: see source-log
+    # round-trip test below).
+    assert session._local_save_path is None
+    assert session.turn_count == 1
+    assert session.last_auto_save_turn == 1
+    assert session.pending_approval is None
+    assert [msg.role for msg in session.context_manager.items] == ["system", "user"]
+    assert session.context_manager.items[0].content == "current system"
+    assert session.context_manager.items[1].content == "continue this work"
+    assert session.context_manager.running_context_usage == 123
+    assert session.context_manager.recompute_calls == ["openai/gpt-5.5"]
+    assert len(session.logged_events) == 1
+    marker = session.logged_events[0]
+    assert marker["event_type"] == "resumed_from"
+    assert marker["data"]["forked"] is False
+    assert marker["data"]["original_session_id"] == "saved-session"
+    assert marker["data"]["original_event_count"] == 1
+
+
+def test_restore_forks_when_user_id_differs(tmp_path):
+    log_dir = tmp_path / "session_logs"
+    path = _write_session_log(
+        log_dir,
+        "session.json",
+        session_id="saved-session",
+        content="someone else's chat",
+        mtime=time.time(),
+        user_id="user-a",
+    )
+
+    session = _FakeSession(user_id="user-b")
+    original_session_id = session.session_id
+    original_start_time = session.session_start_time
+
+    result = session_resume.restore_session_from_log(session, path)
+
+    assert result["forked"] is True
+    assert session.session_id == original_session_id
+    assert session.session_start_time == original_start_time
+    assert session._local_save_path is None
+    marker = session.logged_events[0]
+    assert marker["event_type"] == "resumed_from"
+    assert marker["data"]["forked"] is True
+    assert marker["data"]["original_session_id"] == "saved-session"
+
+
+def test_restore_forks_when_one_side_is_anonymous(tmp_path):
+    log_dir = tmp_path / "session_logs"
+    path = _write_session_log(
+        log_dir,
+        "session.json",
+        session_id="saved-session",
+        content="anonymous save",
+        mtime=time.time(),
+        user_id=None,
+    )
+
+    session = _FakeSession(user_id="user-a")
+
+    result = session_resume.restore_session_from_log(session, path)
+
+    assert result["forked"] is True
+    assert session._local_save_path is None
+
+
+def test_restore_continues_when_both_sides_anonymous(tmp_path):
+    log_dir = tmp_path / "session_logs"
+    path = _write_session_log(
+        log_dir,
+        "session.json",
+        session_id="saved-session",
+        content="local-only chat",
+        mtime=time.time(),
+        user_id=None,
+    )
+
+    session = _FakeSession(user_id=None)
+
+    result = session_resume.restore_session_from_log(session, path)
+
+    assert result["forked"] is False
+    assert session.session_id == "saved-session"
+    assert session._local_save_path is None
+
+
+def test_restore_rejects_invalid_saved_model(tmp_path):
+    log_dir = tmp_path / "session_logs"
+    path = log_dir / "session.json"
+    log_dir.mkdir()
+    path.write_text(
+        json.dumps(
+            {
+                "session_id": "saved",
+                "user_id": "user-a",
+                "model_name": "not a real id with spaces",
+                "messages": [{"role": "user", "content": "hello"}],
+                "events": [],
+            }
+        )
+    )
+
+    session = _FakeSession(user_id="user-a")
+    original_model = session.config.model_name
+
+    result = session_resume.restore_session_from_log(session, path)
+
+    assert result["invalid_saved_model"] == "not a real id with spaces"
+    assert result["model_name"] == original_model
+    assert session.config.model_name == original_model
+
+
+def test_restore_counts_dropped_messages(tmp_path):
+    log_dir = tmp_path / "session_logs"
+    path = log_dir / "session.json"
+    log_dir.mkdir()
+    path.write_text(
+        json.dumps(
+            {
+                "session_id": "saved",
+                "user_id": "user-a",
+                "model_name": "openai/gpt-5.5",
+                "messages": [
+                    {"role": "user", "content": "hi"},
+                    {"role": "user", "content": 12345},  # invalid content type
+                ],
+                "events": [],
+            }
+        )
+    )
+
+    session = _FakeSession(user_id="user-a")
+
+    result = session_resume.restore_session_from_log(session, path)
+
+    assert result["restored_count"] == 1
+    assert result["dropped_count"] == 1
+
+
+def test_restore_does_not_overwrite_source_log_on_save(tmp_path, monkeypatch):
+    """Regression: resuming + saving must not destroy the source log on disk.
+
+    Without the always-fork ``_local_save_path`` reset, the next heartbeat
+    save would rewrite the source file with ``events=[resumed_from]`` and
+    ``total_cost_usd=0``, wiping the original audit trail. This builds a
+    real ``Session`` and exercises the round-trip.
+    """
+    monkeypatch.chdir(tmp_path)
+
+    from agent.context_manager.manager import ContextManager
+    from agent.core.session import Session
+
+    log_dir = tmp_path / "session_logs"
+    log_dir.mkdir()
+    src_path = log_dir / "src.json"
+    src_payload = {
+        "session_id": "saved-session",
+        "user_id": "user-a",
+        "session_start_time": "2026-01-01T00:00:00",
+        "session_end_time": "2026-01-01T00:05:00",
+        "model_name": "openai/gpt-5.5",
+        "messages": [
+            {"role": "system", "content": "old system"},
+            {"role": "user", "content": "earlier work"},
+        ],
+        "events": [
+            {"event_type": "llm_call", "data": {"cost_usd": 0.42}},
+            {"event_type": "turn_complete", "data": {}},
+        ],
+    }
+    src_path.write_text(json.dumps(src_payload, indent=2))
+    src_bytes_before = src_path.read_bytes()
+
+    class _Cfg:
+        model_name = "openai/gpt-5.5"
+        save_sessions = True
+        session_dataset_repo = None
+        auto_save_interval = 1
+        heartbeat_interval_s = 60
+        max_iterations = 10
+        yolo_mode = False
+        confirm_cpu_jobs = False
+        auto_file_upload = False
+        reasoning_effort = None
+        share_traces = False
+        personal_trace_repo_template = None
+        mcpServers: dict = {}
+
+    cm = ContextManager.__new__(ContextManager)
+    cm.items = [Message(role="system", content="current system")]
+    cm.tool_specs = []
+    cm.model_max_tokens = 200_000
+    cm.running_context_usage = 0
+    cm.compact_size = 0.1
+    cm.untouched_messages = 5
+    cm.hf_token = None
+    cm.local_mode = True
+    cm.system_prompt = "current system"
+    cm.on_message_added = None
+
+    import asyncio as _asyncio
+
+    session = Session(
+        event_queue=_asyncio.Queue(),
+        config=_Cfg(),
+        tool_router=None,
+        context_manager=cm,
+        hf_token=None,
+        user_id="user-a",
+        local_mode=True,
+    )
+
+    session_resume.restore_session_from_log(session, src_path)
+    assert session._local_save_path is None
+
+    saved_path = session.save_trajectory_local(directory=str(log_dir))
+
+    assert saved_path is not None
+    assert Path(saved_path) != src_path
+    assert src_path.read_bytes() == src_bytes_before
+
+
+def test_restore_flags_redacted_messages(tmp_path):
+    log_dir = tmp_path / "session_logs"
+    path = _write_session_log(
+        log_dir,
+        "session.json",
+        session_id="saved-session",
+        content="my token is [REDACTED_HF_TOKEN]",
+        mtime=time.time(),
+        user_id="user-a",
+    )
+
+    session = _FakeSession(user_id="user-a")
+
+    result = session_resume.restore_session_from_log(session, path)
+
+    assert result["had_redacted_content"] is True
+
+
+def test_resolve_session_log_arg_accepts_index_and_id_prefix(tmp_path):
+    log_dir = tmp_path / "session_logs"
+    older = _write_session_log(
+        log_dir,
+        "older.json",
+        session_id="abcdef-older",
+        content="x",
+        mtime=time.time() - 10,
+    )
+    newer = _write_session_log(
+        log_dir,
+        "newer.json",
+        session_id="123456-newer",
+        content="y",
+        mtime=time.time(),
+    )
+    entries = session_resume.list_session_logs(log_dir)
+
+    assert session_resume.resolve_session_log_arg("1", entries, log_dir) == newer
+    assert session_resume.resolve_session_log_arg("abc", entries, log_dir) == older
+    assert session_resume.resolve_session_log_arg("nope", entries, log_dir) is None