[issue-3200][slice-6/10] Session-resume substrate (#3186...

sandbox/llm/claude/runner.py

@@ -29,6 +29,7 @@ async def run_agent_async(
     cwd: Path | str | None = None,
     on_output: Callable[[str], None] | None = None,
     model: str | None = None,
+    resume: str | None = None,
 ) -> AgentResult:
     """Run agent via the Claude Agent SDK.
 
@@ -40,6 +41,11 @@ async def run_agent_async(
         on_output: Optional callback for streaming output line-by-line
         model: Model to use (default: opus[1m]). Can be a model alias ('opus', 'sonnet')
                or a full model identifier.
+        resume: Optional Claude ``session_id`` to re-enter (warm resume,
+            #3200 slice-6). Threaded straight through to the SDK client; it
+            is opt-in and default OFF there (a resume only happens when
+            ``EGG_SESSION_RESUME`` is enabled), and an absent/stale session
+            cold-starts from the protected root rather than erroring.
 
     Returns:
         AgentResult with response and status.
@@ -55,6 +61,7 @@ async def run_agent_async(
         cwd=cwd_path,
         timeout=effective_timeout,
         on_output=on_output,
+        resume=resume,
     )
 
     return sdk_result

shared/egg_agent/__main__.py

@@ -13,6 +13,7 @@
 import sys
 
 from egg_agent.client import run_agent
+from egg_agent.session import write_session_state
 
 
 def _stream_to_stdout(text: str) -> None:
@@ -42,6 +43,28 @@ def main() -> int:
         choices=["low", "medium", "high", "max"],
         help=("Reasoning effort for the session. Omit to inherit Claude Code's per-model default."),
     )
+    parser.add_argument(
+        "--resume",
+        default=None,
+        metavar="SESSION_ID",
+        help=(
+            "Claude session_id to re-enter (warm resume, #3200 slice-6). Opt-in "
+            "and default OFF: only resumes when EGG_SESSION_RESUME is enabled; an "
+            "absent/stale session cold-starts from the protected root. The "
+            "resume-vs-reseed decision lives in the slice-8 gate."
+        ),
+    )
+    parser.add_argument(
+        "--session-state-file",
+        default=None,
+        metavar="PATH",
+        help=(
+            "Where to persist this run's session_id + window occupancy so a later "
+            "event-pump invocation can resume it (defaults to $EGG_SESSION_STATE_FILE; "
+            "no-op when neither is set). The round-trip's read+decide side is the "
+            "slice-8 gate."
+        ),
+    )
 
     args = parser.parse_args()
 
@@ -63,6 +86,17 @@ def main() -> int:
         timeout=args.timeout,
         on_output=_stream_to_stdout,
         effort=args.effort,
+        resume=args.resume,
+    )
+
+    # Write side of the session-state round-trip (#3200 slice-6): persist this
+    # run's session_id + window occupancy so a later event-pump invocation can
+    # re-enter (or, per the slice-8 gate, reseed). Best-effort and inert when no
+    # path is configured; a persistence failure never changes the exit code.
+    write_session_state(
+        result.session_id,
+        result.window_occupancy,
+        path=args.session_state_file,
     )
 
     if result.stderr:

shared/egg_agent/_logging.py

@@ -0,0 +1,54 @@
+"""Shared logger resolution for ``egg_agent`` modules.
+
+Every module here logs through egg's structured logger (``event_type=``,
+``event_subtype=``, ``error=`` kwargs). Outside the sandbox ``egg_logging``
+may be absent, so the fallback must still accept those kwargs — a bare
+``logging.Logger`` raises ``TypeError`` on the first unknown keyword, which
+would defeat the "never raise" contract of modules like
+:mod:`egg_agent.session`. ``_StdlibLoggerAdapter`` drops the structured
+kwargs the stdlib doesn't understand; :func:`resolve_logger` wires it up.
+"""
+
+from __future__ import annotations
+
+import logging
+from typing import Any
+
+
+class _StdlibLoggerAdapter:
+    """Thin adapter so a stdlib logger ignores structured-log kwargs."""
+
+    def __init__(self, name: str) -> None:
+        self._logger = logging.getLogger(name)
+
+    def _log(self, level: int, msg: str, **kwargs: Any) -> None:
+        # Drop structured kwargs that stdlib doesn't understand
+        self._logger.log(level, msg)
+
+    def info(self, msg: str, **kwargs: Any) -> None:
+        self._log(logging.INFO, msg, **kwargs)
+
+    def debug(self, msg: str, **kwargs: Any) -> None:
+        self._log(logging.DEBUG, msg, **kwargs)
+
+    def warning(self, msg: str, **kwargs: Any) -> None:
+        self._log(logging.WARNING, msg, **kwargs)
+
+    def error(self, msg: str, **kwargs: Any) -> None:
+        self._log(logging.ERROR, msg, **kwargs)
+
+
+def resolve_logger(structured_name: str, fallback_name: str) -> Any:
+    """Return egg's structured logger, or a kwarg-dropping stdlib fallback.
+
+    ``structured_name`` is the logical name passed to ``egg_logging.get_logger``
+    inside the sandbox; ``fallback_name`` names the stdlib logger used when
+    ``egg_logging`` is unavailable. The fallback is wrapped in
+    :class:`_StdlibLoggerAdapter` so structured-log kwargs never raise.
+    """
+    try:
+        from egg_logging import get_logger
+
+        return get_logger(structured_name)
+    except ImportError:  # pragma: no cover - stdlib fallback outside the sandbox
+        return _StdlibLoggerAdapter(fallback_name)

shared/egg_agent/client.py

@@ -9,13 +9,14 @@
 
 import asyncio
 import json
-import logging
 import os
 from collections.abc import AsyncIterator, Callable
 from pathlib import Path
 from typing import TYPE_CHECKING, Any
 
+from egg_agent._logging import resolve_logger
 from egg_agent.result import AgentResult
+from egg_agent.session import session_resume_enabled
 
 if TYPE_CHECKING:
     # Annotation-only SDK types. With ``from __future__ import annotations``
@@ -88,35 +89,10 @@ def _compute_occupancy(usage: dict[str, Any] | None) -> int | None:
     return sum(components[key] for key in _OCCUPANCY_USAGE_KEYS)
 
 
-class _StdlibLoggerAdapter:
-    """Thin adapter so stdlib logger ignores structured-log kwargs."""
-
-    def __init__(self, name: str) -> None:
-        self._logger = logging.getLogger(name)
-
-    def _log(self, level: int, msg: str, **kwargs: Any) -> None:
-        # Drop structured kwargs that stdlib doesn't understand
-        self._logger.log(level, msg)
-
-    def info(self, msg: str, **kwargs: Any) -> None:
-        self._log(logging.INFO, msg, **kwargs)
-
-    def debug(self, msg: str, **kwargs: Any) -> None:
-        self._log(logging.DEBUG, msg, **kwargs)
-
-    def warning(self, msg: str, **kwargs: Any) -> None:
-        self._log(logging.WARNING, msg, **kwargs)
-
-    def error(self, msg: str, **kwargs: Any) -> None:
-        self._log(logging.ERROR, msg, **kwargs)
-
-
-try:
-    from egg_logging import get_logger
-
-    logger: Any = get_logger("egg-agent")
-except ImportError:
-    logger = _StdlibLoggerAdapter(__name__)
+# ``_StdlibLoggerAdapter`` and the egg_logging-vs-stdlib resolution now live in
+# ``egg_agent._logging`` so sibling modules (e.g. ``session.py``) can reuse the
+# kwarg-dropping fallback without importing this SDK-heavy module.
+logger: Any = resolve_logger("egg-agent", __name__)
 
 # Default model for sandbox agents
 DEFAULT_MODEL = "opus[1m]"
@@ -249,6 +225,7 @@ async def run_agent_async(
     env: dict[str, str] | None = None,
     intercept_tools: bool = True,
     effort: str | None = None,
+    resume: str | None = None,
 ) -> AgentResult:
     """Run a Claude agent using the Agent SDK.
 
@@ -269,6 +246,17 @@ async def run_agent_async(
             ``max``), passed to the CLI as ``--effort``. ``None``
             (default) omits the flag so the session inherits Claude
             Code's per-model default.
+        resume: Optional Claude ``session_id`` to re-enter (warm resume,
+            #3200 slice-6). The BRC event-pump runs one process per event,
+            so resuming lets a re-invocation continue the prior session
+            instead of re-seeding from scratch. Opt-in and default OFF: a
+            resume only happens when this is a non-empty session_id AND
+            ``EGG_SESSION_RESUME`` is enabled. When omitted, disabled, or
+            when the session is gone (expired / consensus reset / pod
+            death) the run cold-starts from the prompt's protected root —
+            a fresh session, never a hard failure. The resume-vs-reseed
+            *decision* (occupancy vs threshold) lives in the slice-8 gate;
+            this is only the substrate that makes resume possible.
 
     Returns:
         :class:`AgentResult` with response text and metadata.
@@ -398,6 +386,39 @@ async def _check_tool_permission(
         # The SDK passes this to the spawned CLI as ``--effort <level>``.
         options.effort = effort
 
+    # --- Warm-session resume substrate (#3200, slice-6) ---
+    # Opt-in, default OFF (staged rollout). Even when a caller passes a
+    # session_id to resume, we only re-enter that session when EGG_SESSION_RESUME
+    # is enabled — so the substrate can ship dark ahead of the slice-8 gate that
+    # drives it. ``options.resume`` is the SDK's "load conversation history from
+    # this session_id" hook; we deliberately leave ``fork_session`` unset so the
+    # resumed session CONTINUES (and keeps accumulating) rather than branching.
+    #
+    # Cold-start fallback: when resume is omitted, disabled, or the prior session
+    # is gone (first invocation / expired session / consensus reset / pod death),
+    # we simply do NOT set ``options.resume`` and the run starts a FRESH session
+    # seeded from the prompt (the protected root). This never raises — a missing
+    # warm session degrades to a fresh seed, not a hard failure. The
+    # resume-vs-reseed DECISION (occupancy vs threshold) is the slice-8 gate; this
+    # layer only makes resume possible.
+    resume_session_id = (resume or "").strip()
+    if resume_session_id and session_resume_enabled():
+        options.resume = resume_session_id
+        logger.info(
+            "Resuming prior agent session",
+            event_type="system",
+            event_subtype="session_resume",
+            session_id=resume_session_id,
+        )
+    elif resume_session_id:
+        logger.info(
+            "Session resume requested but disabled (EGG_SESSION_RESUME off); "
+            "cold-starting from the protected root",
+            event_type="system",
+            event_subtype="session_resume_skipped",
+            session_id=resume_session_id,
+        )
+
     # --- Register in-process SDK MCP servers with egg's agent tools ---
     # Default-on since issue #1942.  Set ``EGG_MCP_TOOLS=false`` (or
     # ``0`` / ``no`` / ``off``) on the pod env to opt out — the kill