fix(checkpoints): honor gateway config and task cwd

agent/tool_executor.py

@@ -53,6 +53,28 @@
 logger = logging.getLogger(__name__)
 
 
+def _ensure_file_checkpoint(
+    agent,
+    function_name: str,
+    function_args: dict,
+    effective_task_id: str,
+) -> None:
+    """Checkpoint the same workspace path that the file tool will mutate."""
+    file_path = function_args.get("path", "")
+    if not file_path:
+        return
+
+    # File tools resolve relative paths against the task's live/session cwd,
+    # which can differ from the Hermes process cwd (notably in Docker).  Resolve
+    # through that same path pipeline before asking the checkpoint manager to
+    # discover the project root.
+    from tools.file_tools import _resolve_path_for_task
+
+    resolved_path = _resolve_path_for_task(file_path, effective_task_id or "default")
+    work_dir = agent._checkpoint_mgr.get_working_dir_for_path(str(resolved_path))
+    agent._checkpoint_mgr.ensure_checkpoint(work_dir, f"before {function_name}")
+
+
 def _budget_for_agent(agent) -> BudgetConfig:
     """Resolve a tool-result BudgetConfig scaled to the agent's context window.
 
@@ -502,10 +524,12 @@ def execute_tool_calls_concurrent(agent, assistant_message, messages: list, effe
             # Checkpoint for file-mutating tools
             if function_name in {"write_file", "patch"} and agent._checkpoint_mgr.enabled:
                 try:
-                    file_path = function_args.get("path", "")
-                    if file_path:
-                        work_dir = agent._checkpoint_mgr.get_working_dir_for_path(file_path)
-                        agent._checkpoint_mgr.ensure_checkpoint(work_dir, f"before {function_name}")
+                    _ensure_file_checkpoint(
+                        agent,
+                        function_name,
+                        function_args,
+                        effective_task_id,
+                    )
                 except Exception:
                     pass
 
@@ -1188,12 +1212,12 @@ def execute_tool_calls_sequential(agent, assistant_message, messages: list, effe
         # Checkpoint: snapshot working dir before file-mutating tools
         if not _execution_blocked and function_name in {"write_file", "patch"} and agent._checkpoint_mgr.enabled:
             try:
-                file_path = function_args.get("path", "")
-                if file_path:
-                    work_dir = agent._checkpoint_mgr.get_working_dir_for_path(file_path)
-                    agent._checkpoint_mgr.ensure_checkpoint(
-                        work_dir, f"before {function_name}"
-                    )
+                _ensure_file_checkpoint(
+                    agent,
+                    function_name,
+                    function_args,
+                    effective_task_id,
+                )
             except Exception:
                 pass  # never block tool execution
 

gateway/platforms/api_server.py

@@ -1777,6 +1777,7 @@ def _create_agent(
         """
         from run_agent import AIAgent
         from gateway.run import (
+            _checkpoint_agent_kwargs,
             _current_max_iterations,
             _resolve_runtime_agent_kwargs,
             _resolve_gateway_model,
@@ -1858,6 +1859,7 @@ def _create_agent(
         agent = AIAgent(
             model=model,
             **runtime_kwargs,
+            **_checkpoint_agent_kwargs(user_config),
             max_iterations=max_iterations,
             quiet_mode=True,
             verbose_logging=False,

gateway/run.py

@@ -2592,6 +2592,35 @@ def _load_gateway_config() -> dict:
     return raw
 
 
+def _checkpoint_agent_kwargs(config: dict | None) -> dict:
+    """Translate gateway checkpoint config into ``AIAgent`` constructor args.
+
+    The gateway reads raw YAML instead of ``load_config()``, so checkpoint
+    defaults must be supplied here.  Keep legacy ``checkpoints: true`` configs
+    working while giving every gateway-created agent the same limits.
+    """
+    cp_cfg = config.get("checkpoints", {}) if isinstance(config, dict) else {}
+    if isinstance(cp_cfg, bool):
+        cp_cfg = {"enabled": cp_cfg}
+    elif not isinstance(cp_cfg, dict):
+        cp_cfg = {}
+
+    from hermes_cli.config import DEFAULT_CONFIG
+    defaults = DEFAULT_CONFIG["checkpoints"]
+    return {
+        "checkpoints_enabled": cp_cfg.get("enabled", defaults["enabled"]),
+        "checkpoint_max_snapshots": cp_cfg.get(
+            "max_snapshots", defaults["max_snapshots"],
+        ),
+        "checkpoint_max_total_size_mb": cp_cfg.get(
+            "max_total_size_mb", defaults["max_total_size_mb"],
+        ),
+        "checkpoint_max_file_size_mb": cp_cfg.get(
+            "max_file_size_mb", defaults["max_file_size_mb"],
+        ),
+    }
+
+
 def _load_gateway_runtime_config() -> dict:
     """Load gateway config for runtime reads, expanding supported ``${VAR}`` refs.
 
@@ -14777,6 +14806,7 @@ def run_sync():
                 agent = AIAgent(
                     model=turn_route["model"],
                     **turn_route["runtime"],
+                    **_checkpoint_agent_kwargs(user_config),
                     max_iterations=max_iterations,
                     quiet_mode=True,
                     verbose_logging=False,
@@ -17273,6 +17303,10 @@ async def _run_process_watcher(self, watcher: dict) -> None:
         ("compression", "protect_last_n"),
         ("agent", "disabled_toolsets"),
         ("memory", "provider"),
+        ("checkpoints", "enabled"),
+        ("checkpoints", "max_snapshots"),
+        ("checkpoints", "max_total_size_mb"),
+        ("checkpoints", "max_file_size_mb"),
     )
 
     _HONCHO_CACHE_BUSTING_KEYS = (
@@ -17336,7 +17370,11 @@ def _extract_cache_busting_config(cls, user_config: dict | None) -> dict:
         cfg = user_config if isinstance(user_config, dict) else {}
         for section, key in cls._CACHE_BUSTING_CONFIG_KEYS:
             section_val = cfg.get(section)
-            if isinstance(section_val, dict):
+            if section == "checkpoints" and isinstance(section_val, bool):
+                # Preserve legacy ``checkpoints: true`` behavior.  A live
+                # toggle must still rebuild the cached agent.
+                out[f"{section}.{key}"] = section_val if key == "enabled" else None
+            elif isinstance(section_val, dict):
                 out[f"{section}.{key}"] = section_val.get(key)
             else:
                 out[f"{section}.{key}"] = None
@@ -20303,6 +20341,7 @@ def _interim_assistant_cb(text: str, *, already_streamed: bool = False) -> None:
                 agent = AIAgent(
                     model=turn_route["model"],
                     **turn_route["runtime"],
+                    **_checkpoint_agent_kwargs(user_config),
                     max_iterations=max_iterations,
                     quiet_mode=True,
                     verbose_logging=False,

gateway/slash_commands.py

@@ -2668,31 +2668,19 @@ async def _handle_voice_command(self, event: MessageEvent) -> str:
 
     async def _handle_rollback_command(self, event: MessageEvent) -> str:
         """Handle /rollback command — list or restore filesystem checkpoints."""
-        from gateway.run import _hermes_home
+        from gateway.run import _checkpoint_agent_kwargs, _load_gateway_config
         from tools.checkpoint_manager import CheckpointManager, format_checkpoint_list
 
-        # Read checkpoint config from config.yaml
-        cp_cfg = {}
-        try:
-            import yaml as _y
-            _cfg_path = _hermes_home / "config.yaml"
-            if _cfg_path.exists():
-                with open(_cfg_path, encoding="utf-8") as _f:
-                    _data = _y.safe_load(_f) or {}
-                cp_cfg = _data.get("checkpoints", {})
-                if isinstance(cp_cfg, bool):
-                    cp_cfg = {"enabled": cp_cfg}
-        except Exception:
-            pass
+        cp_kwargs = _checkpoint_agent_kwargs(_load_gateway_config())
 
-        if not cp_cfg.get("enabled", False):
+        if not cp_kwargs["checkpoints_enabled"]:
             return t("gateway.rollback.not_enabled")
 
         mgr = CheckpointManager(
             enabled=True,
-            max_snapshots=cp_cfg.get("max_snapshots", 50),
-            max_total_size_mb=cp_cfg.get("max_total_size_mb", 500),
-            max_file_size_mb=cp_cfg.get("max_file_size_mb", 10),
+            max_snapshots=cp_kwargs["checkpoint_max_snapshots"],
+            max_total_size_mb=cp_kwargs["checkpoint_max_total_size_mb"],
+            max_file_size_mb=cp_kwargs["checkpoint_max_file_size_mb"],
         )
 
         cwd = os.getenv("TERMINAL_CWD", str(Path.home()))

tests/agent/test_tool_executor_checkpoint_paths.py

@@ -0,0 +1,40 @@
+"""Behavioral coverage for file-tool checkpoint path resolution."""
+
+from types import SimpleNamespace
+
+from agent.tool_executor import _ensure_file_checkpoint
+from tools.checkpoint_manager import CheckpointManager
+
+
+def test_relative_file_checkpoint_uses_task_workspace(tmp_path, monkeypatch):
+    """Checkpoint lookup must use the same cwd as a relative file mutation."""
+    process_cwd = tmp_path / "opt" / "hermes"
+    workspace_cwd = tmp_path / "opt" / "data" / "workspace"
+    process_cwd.mkdir(parents=True)
+    workspace_cwd.mkdir(parents=True)
+
+    # Both directories contain content so checkpointing the wrong one would
+    # still succeed and remain observable as the regression did in Docker.
+    (process_cwd / "pyproject.toml").write_text("[project]\nname = 'hermes'\n")
+    (workspace_cwd / "pyproject.toml").write_text("[project]\nname = 'workspace'\n")
+    (workspace_cwd / "existing.txt").write_text("before\n")
+
+    monkeypatch.chdir(process_cwd)
+    monkeypatch.setenv("TERMINAL_CWD", str(workspace_cwd))
+    monkeypatch.setattr(
+        "tools.checkpoint_manager.CHECKPOINT_BASE",
+        tmp_path / "checkpoints",
+    )
+
+    manager = CheckpointManager(enabled=True)
+    agent = SimpleNamespace(_checkpoint_mgr=manager)
+
+    _ensure_file_checkpoint(
+        agent,
+        "write_file",
+        {"path": "test_permissions2.txt"},
+        "gateway-session",
+    )
+
+    assert manager.list_checkpoints(str(workspace_cwd))
+    assert manager.list_checkpoints(str(process_cwd)) == []

tests/gateway/test_agent_cache.py

@@ -243,6 +243,32 @@ def test_reads_compression_subkeys(self):
         assert out["compression.protect_last_n"] == 25
         assert out["compression.codex_app_server_auto"] == "hermes"
 
+    def test_reads_checkpoint_subkeys(self):
+        from gateway.run import GatewayRunner
+
+        out = GatewayRunner._extract_cache_busting_config(
+            {
+                "checkpoints": {
+                    "enabled": True,
+                    "max_snapshots": 12,
+                    "max_total_size_mb": 333,
+                    "max_file_size_mb": 5,
+                }
+            }
+        )
+
+        assert out["checkpoints.enabled"] is True
+        assert out["checkpoints.max_snapshots"] == 12
+        assert out["checkpoints.max_total_size_mb"] == 333
+        assert out["checkpoints.max_file_size_mb"] == 5
+
+    def test_reads_legacy_checkpoint_boolean(self):
+        from gateway.run import GatewayRunner
+
+        out = GatewayRunner._extract_cache_busting_config({"checkpoints": True})
+
+        assert out["checkpoints.enabled"] is True
+
     def test_missing_keys_yield_none(self):
         """Absent config keys must produce None values (still contribute to signature)."""
         from gateway.run import GatewayRunner

tests/gateway/test_api_server.py

@@ -330,7 +330,7 @@ def test_invalid_port_from_env_falls_back_to_default(self, monkeypatch):
         adapter = APIServerAdapter(config)
         assert adapter._port == 8642
 
-    def test_create_agent_forwards_config_reasoning_effort(self, monkeypatch):
+    def test_create_agent_forwards_runtime_config(self, monkeypatch):
         captured = {}
 
         class FakeAgent:
@@ -349,7 +349,15 @@ def __init__(self, **kwargs):
         monkeypatch.setattr("gateway.run._resolve_gateway_model", lambda: "gpt-5.5")
         monkeypatch.setattr(
             "gateway.run._load_gateway_config",
-            lambda: {"agent": {"reasoning_effort": "xhigh"}},
+            lambda: {
+                "agent": {"reasoning_effort": "xhigh"},
+                "checkpoints": {
+                    "enabled": True,
+                    "max_snapshots": 7,
+                    "max_total_size_mb": 321,
+                    "max_file_size_mb": 4,
+                },
+            },
         )
         monkeypatch.setattr(
             "gateway.run.GatewayRunner._load_reasoning_config",
@@ -365,6 +373,10 @@ def __init__(self, **kwargs):
 
         assert isinstance(agent, FakeAgent)
         assert captured["reasoning_config"] == {"enabled": True, "effort": "xhigh"}
+        assert captured["checkpoints_enabled"] is True
+        assert captured["checkpoint_max_snapshots"] == 7
+        assert captured["checkpoint_max_total_size_mb"] == 321
+        assert captured["checkpoint_max_file_size_mb"] == 4
 
     def test_create_agent_refreshes_max_iterations_from_runtime_config(self, monkeypatch):
         captured = {}

tests/gateway/test_background_command.py

@@ -248,7 +248,16 @@ async def test_successful_task_sends_result(self):
 
         mock_result = {"final_response": "Hello from background!", "messages": []}
 
+        checkpoint_config = {
+            "checkpoints": {
+                "enabled": True,
+                "max_snapshots": 8,
+                "max_total_size_mb": 222,
+                "max_file_size_mb": 3,
+            }
+        }
         with patch("gateway.run._resolve_runtime_agent_kwargs", return_value={"api_key": "test-key"}), \
+             patch("gateway.run._load_gateway_config", return_value=checkpoint_config), \
              patch("run_agent.AIAgent") as MockAgent:
             mock_agent_instance = MagicMock()
             mock_agent_instance.shutdown_memory_provider = MagicMock()
@@ -264,6 +273,11 @@ async def test_successful_task_sends_result(self):
         content = call_args[1].get("content", call_args[0][1] if len(call_args[0]) > 1 else "")
         assert "Background task complete" in content
         assert "Hello from background!" in content
+        agent_kwargs = MockAgent.call_args.kwargs
+        assert agent_kwargs["checkpoints_enabled"] is True
+        assert agent_kwargs["checkpoint_max_snapshots"] == 8
+        assert agent_kwargs["checkpoint_max_total_size_mb"] == 222
+        assert agent_kwargs["checkpoint_max_file_size_mb"] == 3
         mock_agent_instance.shutdown_memory_provider.assert_called_once()
         mock_agent_instance.close.assert_called_once()
 

tests/gateway/test_checkpoint_config.py

@@ -0,0 +1,53 @@
+"""Runtime coverage for gateway filesystem-checkpoint configuration."""
+
+
+def test_gateway_checkpoint_config_reaches_real_agent(tmp_path, monkeypatch):
+    """Raw gateway YAML must configure the real agent checkpoint manager."""
+    from gateway import run as gateway_run
+    from run_agent import AIAgent
+
+    monkeypatch.setenv("HERMES_HOME", str(tmp_path))
+    monkeypatch.setattr(gateway_run, "_hermes_home", tmp_path)
+    (tmp_path / "config.yaml").write_text(
+        """checkpoints:
+  enabled: true
+  max_snapshots: 11
+  max_total_size_mb: 345
+  max_file_size_mb: 6
+""",
+        encoding="utf-8",
+    )
+
+    config = gateway_run._load_gateway_config()
+    agent = AIAgent(
+        model="anthropic/claude-sonnet-4",
+        api_key="test",
+        base_url="https://openrouter.ai/api/v1",
+        provider="openrouter",
+        quiet_mode=True,
+        skip_context_files=True,
+        skip_memory=True,
+        enabled_toolsets=[],
+        **gateway_run._checkpoint_agent_kwargs(config),
+    )
+    try:
+        manager = agent._checkpoint_mgr
+        assert manager.enabled is True
+        assert manager.max_snapshots == 11
+        assert manager.max_total_size_mb == 345
+        assert manager.max_file_size_mb == 6
+    finally:
+        agent.close()
+
+
+def test_checkpoint_agent_kwargs_supports_legacy_boolean_config():
+    from gateway.run import _checkpoint_agent_kwargs
+    from hermes_cli.config import DEFAULT_CONFIG
+
+    kwargs = _checkpoint_agent_kwargs({"checkpoints": True})
+    defaults = DEFAULT_CONFIG["checkpoints"]
+
+    assert kwargs["checkpoints_enabled"] is True
+    assert kwargs["checkpoint_max_snapshots"] == defaults["max_snapshots"]
+    assert kwargs["checkpoint_max_total_size_mb"] == defaults["max_total_size_mb"]
+    assert kwargs["checkpoint_max_file_size_mb"] == defaults["max_file_size_mb"]