fix(cli): restore stored session runtime and reopen ended rows on oneshot resume

Review fixes (#105957):

- A resumed one-shot ignored the session's stored model/provider runtime:
  _resolve_model_and_provider()/resolve_runtime_provider() ran before
  _load_resume_target(), which only loaded the session id + transcript, so an
  ambient config (e.g. openrouter/ambient-model) served the resumed transcript
  instead of the stored route (custom:stored/stored-model). The stored runtime
  is now applied before runtime resolution, with the same contract as the
  interactive _restore_session_model(): stored model/provider/base_url/api_mode
  replace the ambient choice unless --model was passed explicitly, and a
  changed provider drops the ambient api_key so resolution re-fetches
  credentials for the restored endpoint.

- Passing the resumed id to AIAgent did not reopen the already-ended session
  row: end_session() only writes rows whose ended_at is null and the
  existing-row upsert never clears the end fields, so the resumed turn was
  recorded under a session that stayed closed and its new lifecycle boundary
  was lost. _load_resume_target() now reopens the row (best effort), same as
  the interactive resume does before continuing.

hermes_cli/oneshot.py

@@ -281,6 +281,7 @@ class _ModelChoice:
     provider: str | None
     base_url: str | None = None
     api_key: str | None = None
+    api_mode: str | None = None
 
 
 def _configured_model(model_cfg: object) -> str:
@@ -343,8 +344,8 @@ def _resolve_model_and_provider(cfg: dict, model: Optional[str], provider: Optio
     return choice
 
 
-def _load_resume_target(session_db, resume: Optional[str]) -> tuple[Optional[str], list]:
-    """Resolve ``resume`` to ``(session_id, conversation_history)`` for a oneshot turn.
+def _load_resume_target(session_db, resume: Optional[str]) -> tuple[Optional[str], list, Optional[dict]]:
+    """Resolve ``resume`` to ``(session_id, conversation_history, session_meta)`` for a oneshot turn.
 
     Follows the same contract as the interactive CLI resume: compression-chain redirect via
     ``resolve_resume_session_id``, safe-resume guard, model-projection history with
@@ -354,18 +355,65 @@ def _load_resume_target(session_db, resume: Optional[str]) -> tuple[Optional[str
     nothing but is recorded under the requested session — ``hermes -z "hello" -c <title>
     --create-if-missing`` must fill the titled session it created, not mint a fresh id
     (same contract as the interactive /resume of an empty session).
+
+    The resolved row is also reopened (best effort): the previous run stamped ``ended_at``,
+    the existing-row upsert never clears the end fields, and ``end_session()`` only writes
+    rows whose ``ended_at`` is null — so without this step the resumed turn would be recorded
+    under a session that stays closed and its new lifecycle boundary would be lost (same
+    reason the interactive resume calls ``reopen_session()`` before continuing).
     """
     if not resume:
-        return None, []
+        return None, [], None
     if session_db is None:
         raise RuntimeError(f"cannot resume session {resume}: session store unavailable")
     resolved = session_db.resolve_resume_session_id(resume) or resume
-    if not session_db.get_session(resolved):
+    session_meta = session_db.get_session(resolved)
+    if not session_meta:
         raise RuntimeError(f"session not found: {resume}")
     session_db.assert_resume_safe(resolved, tip_only=True)
     restored, _display = session_db.get_resume_conversations(resolved)
     history = [m for m in restored if m.get("role") != "session_meta"]
-    return resolved, history
+    try:
+        session_db.reopen_session(resolved)
+    except Exception:
+        logging.debug("reopen_session failed for resumed one-shot session %s", resolved, exc_info=True)
+    return resolved, history, session_meta
+
+
+def _apply_stored_session_runtime(
+    choice: _ModelChoice, session_meta: Optional[dict], *, explicit_model: bool,
+) -> _ModelChoice:
+    """Run a resumed one-shot on the session's stored runtime, not the ambient config.
+
+    Same contract as the interactive ``_restore_session_model`` (cli_model_switch_mixin): no
+    stored model, or an explicit ``--model``, keeps the resolved ambient choice; otherwise the
+    stored model/provider/base_url/api_mode replace it (a resumed transcript must not be sent
+    to whatever the ambient config happens to point at). A changed provider drops the resolved
+    ``api_key`` — it belongs to the ambient/alias endpoint, and api_key is never persisted to
+    the session DB, so runtime resolution re-fetches credentials for the restored provider.
+    No-op when the stored route already matches the resolved one.
+    """
+    stored_model = str((session_meta or {}).get("model") or "").strip()
+    if not stored_model or explicit_model:
+        return choice
+    from hermes_state import SessionDB as _SessionDB
+    from hermes_cli.cli_model_switch_mixin import _heal_bare_custom_provider
+
+    stored_runtime = _SessionDB.session_gateway_runtime(session_meta)
+    stored_base_url = stored_runtime.get("base_url") or None
+    # Stricter than the TUI gateway's recovery — the CLI's resolve path hard-fails on bare "custom".
+    stored_provider = _heal_bare_custom_provider(
+        stored_runtime.get("provider") or None, base_url=stored_base_url, model=stored_model)
+    if stored_model == choice.model and (not stored_provider or stored_provider == choice.provider):
+        return choice
+    choice.model = stored_model
+    if stored_provider and stored_provider != choice.provider:
+        choice.provider = stored_provider
+        choice.base_url = stored_base_url
+        choice.api_key = None
+    if stored_runtime.get("api_mode"):
+        choice.api_mode = str(stored_runtime["api_mode"])
+    return choice
 
 
 def _run_agent(
@@ -386,12 +434,20 @@ def _run_agent(
 
     cfg = load_config()
     choice = _resolve_model_and_provider(cfg, model, provider)
+    # Resume resolves BEFORE the runtime provider: the session's stored model/route must
+    # replace the ambient config (see _apply_stored_session_runtime) and the ended row must
+    # be reopened before the agent can stamp a new lifecycle boundary.
+    session_db = _create_session_db_for_oneshot()
+    resume_sid, conversation_history, resume_meta = _load_resume_target(session_db, resume)
+    choice = _apply_stored_session_runtime(choice, resume_meta, explicit_model=bool((model or "").strip()))
     runtime = resolve_runtime_provider(
         requested=choice.provider,
         target_model=choice.model or None,
         explicit_base_url=choice.base_url,
         explicit_api_key=choice.api_key,
     )
+    if choice.api_mode:
+        runtime["api_mode"] = choice.api_mode
 
     # sorted() gives stable ordering for config-derived sets; explicit values preserve user order.
     toolsets_list = _normalize_toolsets(toolsets)
@@ -410,8 +466,6 @@ def _run_agent(
 
     skills_prompt = _build_preloaded_skills_prompt(skills)
 
-    session_db = _create_session_db_for_oneshot()
-    resume_sid, conversation_history = _load_resume_target(session_db, resume)
     # The try spans agent construction (not just ``chat``) so the store is always closed, even when
     # ``AIAgent(...)`` raises — the one-shot exit path hard-exits via os._exit and skips finalizers.
     agent = None

tests/hermes_cli/test_oneshot_resume.py

@@ -5,7 +5,9 @@ them: ``_run_oneshot_from_args`` ran before any session-arg normalization and ne
 forwarded ``args.resume``, so every resumed one-shot turn started a FRESH session —
 the wire carried only ``[system, current user]`` and the model "forgot" everything.
 These tests pin the loader contract (chain redirect, unknown-session error,
-session_meta filtering, empty-session fresh start) and the resume kwarg wiring.
+session_meta filtering, empty-session fresh start, ended-row reopen) and the resume
+kwarg wiring, plus the stored-runtime restore (review on #105957): a resumed one-shot
+must run on the session's stored model/provider runtime and on a reopened session row.
 """
 
 from __future__ import annotations
@@ -13,7 +15,12 @@ from __future__ import annotations
 import pytest
 
 from hermes_state import SessionDB
-from hermes_cli.oneshot import _load_resume_target, run_oneshot
+from hermes_cli.oneshot import (
+    _apply_stored_session_runtime,
+    _load_resume_target,
+    _ModelChoice,
+    run_oneshot,
+)
 
 
 def _db_with_session(tmp_path, sid, *, messages=()):
@@ -28,8 +35,8 @@ class TestLoadResumeTarget:
     def test_no_resume_is_a_noop(self, tmp_path):
         db = _db_with_session(tmp_path, "s1")
         try:
-            assert _load_resume_target(db, None) == (None, [])
-            assert _load_resume_target(db, "") == (None, [])
+            assert _load_resume_target(db, None) == (None, [], None)
+            assert _load_resume_target(db, "") == (None, [], None)
         finally:
             db.close()
 
@@ -54,10 +61,11 @@ class TestLoadResumeTarget:
                       ("assistant", "I've noted the secret word: ZEBRA42.")],
         )
         try:
-            sid, history = _load_resume_target(db, "s1")
+            sid, history, meta = _load_resume_target(db, "s1")
             assert sid == "s1"
             assert [m["role"] for m in history] == ["user", "assistant"]
             assert history[0]["content"] == "Remember the secret word ZEBRA42"
+            assert meta["id"] == "s1"
         finally:
             db.close()
 
@@ -67,7 +75,7 @@ class TestLoadResumeTarget:
             messages=[("session_meta", "model switch"), ("user", "hello")],
         )
         try:
-            sid, history = _load_resume_target(db, "s1")
+            sid, history, _meta = _load_resume_target(db, "s1")
             assert sid == "s1"
             assert [m["role"] for m in history] == ["user"]
         finally:
@@ -80,7 +88,10 @@ class TestLoadResumeTarget:
         # leaving the freshly created titled session empty (review on #105957).
         db = _db_with_session(tmp_path, "s1")
         try:
-            assert _load_resume_target(db, "s1") == ("s1", [])
+            sid, history, meta = _load_resume_target(db, "s1")
+            assert sid == "s1"
+            assert history == []
+            assert meta["id"] == "s1"
         finally:
             db.close()
 
@@ -92,7 +103,9 @@ class TestLoadResumeTarget:
         db.create_session(session_id="head", source="cli")
         db.create_session(session_id="child", source="cli", parent_session_id="head")
         try:
-            assert _load_resume_target(db, "head") == ("head", [])
+            sid, history, _meta = _load_resume_target(db, "head")
+            assert sid == "head"
+            assert history == []
         finally:
             db.close()
 
@@ -104,12 +117,202 @@ class TestLoadResumeTarget:
         db.create_session(session_id="child", source="cli", parent_session_id="parent")
         db.append_message("child", "user", content="hi")
         try:
-            sid, history = _load_resume_target(db, "parent")
+            sid, history, _meta = _load_resume_target(db, "parent")
             assert sid == "child"
             assert [m["role"] for m in history] == ["user"]
         finally:
             db.close()
 
+    def test_resume_reopens_ended_session_row(self, tmp_path):
+        # The previous run stamped ended_at; without reopen_session() the resumed turn is
+        # recorded under a row that stays closed and end_session() cannot stamp the new
+        # boundary (it only writes rows whose ended_at is null) — review on #105957.
+        db = _db_with_session(tmp_path, "s1", messages=[("user", "hi")])
+        db.end_session("s1", "agent_close")
+        row = db.get_session("s1")
+        assert row["ended_at"] is not None and row["end_reason"] == "agent_close"
+        try:
+            sid, _history, _meta = _load_resume_target(db, "s1")
+            assert sid == "s1"
+            reopened = db.get_session("s1")
+            assert reopened["ended_at"] is None
+            assert reopened["end_reason"] is None
+        finally:
+            db.close()
+
+    def test_reopen_failure_is_best_effort(self, tmp_path, monkeypatch):
+        db = _db_with_session(tmp_path, "s1", messages=[("user", "hi")])
+        try:
+            def _boom(_sid):
+                raise RuntimeError("reopen unavailable")
+            monkeypatch.setattr(db, "reopen_session", _boom)
+            sid, history, _meta = _load_resume_target(db, "s1")
+            assert sid == "s1"
+            assert [m["role"] for m in history] == ["user"]
+        finally:
+            db.close()
+
+
+class TestApplyStoredSessionRuntime:
+    """A resumed one-shot must run on the session's stored runtime, not the ambient
+    config (review on #105957): with ambient ``openrouter/ambient-model`` and a stored
+    ``custom:stored/stored-model`` session, the agent previously received the ambient
+    model/provider on the wire."""
+
+    _AMBIENT_KEY = object()  # sentinel: any non-None token; resume must drop it when the provider changes
+
+    @staticmethod
+    def _stored_db(tmp_path, *, model="stored-model", route=None):
+        db = SessionDB(db_path=tmp_path / "state.db")
+        db.create_session(session_id="s1", source="cli")
+        if model:
+            db.update_session_model("s1", model)
+        if route is not None:
+            # Same two shapes _persist_model_switch_to_session writes (nested gateway_runtime
+            # for CLI resume, top-level keys for TUI).
+            db.patch_session_model_config("s1", {"gateway_runtime": route, **route})
+        meta = db.get_session("s1")
+        db.close()
+        return meta
+
+    def test_stored_runtime_replaces_ambient_choice(self, tmp_path):
+        meta = self._stored_db(tmp_path, route={"provider": "custom:stored", "base_url": "http://stored:9", "api_mode": "responses"})
+        choice = _apply_stored_session_runtime(
+            _ModelChoice("ambient-model", "openrouter", api_key=self._AMBIENT_KEY),
+            meta, explicit_model=False)
+        assert choice.model == "stored-model"
+        assert choice.provider == "custom:stored"
+        assert choice.base_url == "http://stored:9"
+        assert choice.api_mode == "responses"
+        # The ambient key belongs to the ambient endpoint and must not ride along.
+        assert choice.api_key is None
+
+    def test_explicit_model_flag_wins(self, tmp_path):
+        meta = self._stored_db(tmp_path, route={"provider": "custom:stored"})
+        choice = _apply_stored_session_runtime(
+            _ModelChoice("explicit-model", "openrouter", api_key=self._AMBIENT_KEY),
+            meta, explicit_model=True)
+        assert choice.model == "explicit-model"
+        assert choice.provider == "openrouter"
+        assert choice.api_key is self._AMBIENT_KEY
+
+    def test_no_stored_model_keeps_ambient_choice(self, tmp_path):
+        meta = self._stored_db(tmp_path, model=None)
+        choice = _apply_stored_session_runtime(
+            _ModelChoice("ambient-model", "openrouter", api_key=self._AMBIENT_KEY), meta, explicit_model=False)
+        assert choice.model == "ambient-model"
+        assert choice.provider == "openrouter"
+        assert choice.api_key is self._AMBIENT_KEY
+
+    def test_matching_route_is_noop_and_keeps_credentials(self, tmp_path):
+        meta = self._stored_db(tmp_path, route={"provider": "openrouter"})
+        choice = _apply_stored_session_runtime(
+            _ModelChoice("stored-model", "openrouter", base_url=None, api_key=self._AMBIENT_KEY),
+            meta, explicit_model=False)
+        assert choice.api_key is self._AMBIENT_KEY
+        assert choice.api_mode is None
+
+    def test_none_meta_is_a_noop(self):
+        choice = _ModelChoice("ambient-model", "openrouter")
+        assert _apply_stored_session_runtime(choice, None, explicit_model=False) == choice
+
+    def test_bare_custom_provider_is_healed_or_dropped(self, tmp_path):
+        # Bare "custom" persisted by older builds is not a routable identity; the CLI's
+        # resolve path would hard-fail on it, so it is healed via the base_url or dropped.
+        meta = self._stored_db(tmp_path, route={"provider": "custom"})
+        choice = _apply_stored_session_runtime(
+            _ModelChoice("ambient-model", "openrouter"), meta, explicit_model=False)
+        assert choice.model == "stored-model"
+        assert choice.provider in (None, "openrouter") or choice.provider.startswith("custom:")
+
+
+class TestRunAgentResumeRuntime:
+    """End-to-end wiring: ``_run_agent`` must hand AIAgent the session's stored runtime
+    and a reopened session row (both regressions from the review on #105957)."""
+
+    def test_run_agent_uses_stored_runtime_and_reopens_row(self, tmp_path, monkeypatch):
+        import hermes_cli.oneshot as oneshot_mod
+
+        db = SessionDB(db_path=tmp_path / "state.db")
+        db.create_session(session_id="s1", source="cli")
+        db.append_message("s1", "user", content="hi")
+        db.update_session_model("s1", "stored-model")
+        db.patch_session_model_config(
+            "s1", {"gateway_runtime": {"provider": "custom:stored", "base_url": "http://stored:9"},
+                   "provider": "custom:stored", "base_url": "http://stored:9"})
+        db.end_session("s1", "agent_close")
+
+        captured = {}
+
+        class _FakeAgent:
+            def __init__(self, **kwargs):
+                captured.update(kwargs)
+            def __setattr__(self, name, _value):
+                pass
+            def run_conversation(self, _prompt, conversation_history=None):
+                captured["history"] = conversation_history
+                return {"final_response": "ok", "session_id": "s1"}
+            def close(self):
+                pass
+
+        monkeypatch.setattr(oneshot_mod, "_create_session_db_for_oneshot", lambda: db)
+        monkeypatch.setattr("hermes_cli.config.load_config", lambda: {"model": {"default": "ambient-model", "provider": "openrouter"}})
+        monkeypatch.setattr(
+            "hermes_cli.runtime_provider.resolve_runtime_provider",
+            lambda **_kw: {"api_key": "resolved", "base_url": None, "provider": "custom:stored",
+                          "requested_provider": "custom:stored", "api_mode": "chat", "credential_pool": None})
+        monkeypatch.setattr("hermes_cli.tools_config._get_platform_tools", lambda _cfg, _p: [])
+        monkeypatch.setattr("hermes_cli.mcp_startup.ensure_mcp_discovery_before_agent_build", lambda **_kw: None)
+        monkeypatch.setattr("run_agent.AIAgent", _FakeAgent)
+
+        try:
+            text, result = oneshot_mod._run_agent("hello", resume="s1")
+            assert text == "ok" and result["final_response"] == "ok"
+            assert captured["model"] == "stored-model"
+            assert captured["provider"] == "custom:stored"
+            assert captured["session_id"] == "s1"
+            assert captured["history"][0]["role"] == "user"
+            row = db.get_session("s1")
+            assert row["ended_at"] is None and row["end_reason"] is None
+        finally:
+            db.close()
+
+    def test_run_agent_explicit_model_beats_stored_runtime(self, tmp_path, monkeypatch):
+        import hermes_cli.oneshot as oneshot_mod
+
+        db = SessionDB(db_path=tmp_path / "state.db")
+        db.create_session(session_id="s1", source="cli")
+        db.update_session_model("s1", "stored-model")
+
+        captured = {}
+
+        class _FakeAgent:
+            def __init__(self, **kwargs):
+                captured.update(kwargs)
+            def __setattr__(self, name, _value):
+                pass
+            def run_conversation(self, _prompt, conversation_history=None):
+                return {"final_response": "ok"}
+            def close(self):
+                pass
+
+        monkeypatch.setattr(oneshot_mod, "_create_session_db_for_oneshot", lambda: db)
+        monkeypatch.setattr("hermes_cli.config.load_config", lambda: {"model": {"default": "ambient-model", "provider": "openrouter"}})
+        monkeypatch.setattr(
+            "hermes_cli.runtime_provider.resolve_runtime_provider",
+            lambda **_kw: {"api_key": "resolved", "base_url": None, "provider": "openrouter",
+                          "requested_provider": "openrouter", "api_mode": "chat", "credential_pool": None})
+        monkeypatch.setattr("hermes_cli.tools_config._get_platform_tools", lambda _cfg, _p: [])
+        monkeypatch.setattr("hermes_cli.mcp_startup.ensure_mcp_discovery_before_agent_build", lambda **_kw: None)
+        monkeypatch.setattr("run_agent.AIAgent", _FakeAgent)
+
+        try:
+            oneshot_mod._run_agent("hello", model="explicit-model", provider="openrouter", resume="s1")
+            assert captured["model"] == "explicit-model"
+            assert captured["provider"] == "openrouter"
+        finally:
+            db.close()
+
 
 class TestRunOneshotForwardsResume:
     def test_resume_kwarg_reaches_run_agent(self, monkeypatch):