fix(agent): auto-start scenario chat, robust HITL resume, strict service auth

- frontend: fix auto-start on dashboards->/agent navigation (undefined params
  ReferenceError), route initial connect through ConnectionManager with auto-retry,
  reset runModel on objectId change and failed recovery
- agent: fix closure-over-loop-variable bug in _inject_env_id_into_tools (env now
  resolved from request-local ContextVar; idempotent wrapping), make
  execute_dashboard_result.result_key optional, resilient checkpoint resume with
  ToolMessage repair + direct-tool fallback, remove dead fast-path, consolidate
  tool_call parsing in _tool_resolver, context-safe ContextVar resets
- backend: llm-config gated by strict service-only auth (no user-JWT fallback),
  tighten idempotent run reuse (dashboard/env/intent match + 6h staleness),
  terminal event transitions run.status to COMPLETED/FAILED/CANCELLED,
  null-safe metric parsing in dashboard query model
- run.sh/docker-compose: require SERVICE_JWT (random per-run secret) instead of
  public default

agent/src/ss_tools/agent/_confirmation.py

@@ -12,6 +12,7 @@ from collections.abc import AsyncGenerator
 import json
 from typing import Any
 
+from langchain_core.messages import ToolMessage
 from langchain_openai import ChatOpenAI
 
 from ss_tools.agent._llm_params import chat_openai_kwargs
@@ -19,7 +20,7 @@ from ss_tools.shared._llm_http import get_shared_http_client
 from ss_tools.agent._tool_resolver import (
     extract_tool_call_from_state,
     find_tool,
-    normalize_tool_args,
+    pending_tool_calls_from_state,
 )
 from ss_tools.agent.langgraph_setup import create_agent
 from ss_tools.agent.tools import get_all_tools
@@ -27,6 +28,21 @@ from ss_tools.agent.tools import get_all_tools
 _pending_confirmations: dict[str, dict[str, Any]] = {}
 
 
+# #region AgentChat.Confirmation.PendingToolCalls [C:2] [TYPE Function] [SEMANTICS agent-chat,hitl,resume,repair]
+# @ingroup AgentChat
+# @BRIEF Extract tool calls from a graph state that are missing a ToolMessage.
+# @POST Returns list of (tool_name, tool_args, tool_call_id) — empty when the
+#       checkpoint history is consistent.
+# @RATIONALE Single source of truth lives in AgentChat.ToolResolver
+#   (pending_tool_calls_from_state); this alias keeps callers/tests stable and
+#   avoids a second divergent parser. Checkpoints taken with interrupt_before=
+#   ["tools"] (or left behind by a crashed tool run) contain AI messages whose
+#   tool calls have no ToolMessage — the resume fallback executes these directly
+#   so the user still receives a response instead of a dead stream.
+_pending_tool_calls = pending_tool_calls_from_state
+# #endregion AgentChat.Confirmation.PendingToolCalls
+
+
 # #region AgentChat.Confirmation.Contract [C:2] [TYPE Function] [SEMANTICS agent-chat,hitl,contract]
 # @ingroup AgentChat
 # @BRIEF Build confirmation contract dict — risk level, prompt, operation metadata.
@@ -244,6 +260,11 @@ def confirmation_metadata(
 # @ingroup AgentChat
 # @BRIEF Serialise confirmation into a JSON payload string for the Gradio event stream.
 # @POST Returns JSON string with content + metadata.
+# @SIDE_EFFECT Stores a title/args marker in _pending_confirmations so app.py can
+#   build a descriptive conversation title on resume ("✅ inspect_dashboard_query_model"
+#   instead of "HITL: confirm"). The marker is marked _fast_path=False so handle_resume
+#   still continues the LangGraph checkpoint (multi-step scenario continuation) rather
+#   than executing a single tool and stopping.
 def confirmation_payload(
     conv_id: str,
     state,
@@ -251,9 +272,17 @@ def confirmation_payload(
     user_role: str | None = None,
     target_env: str | None = None,
 ) -> str:
+    metadata = confirmation_metadata(conv_id, state, user_text, user_role, target_env)
+    tool_name = metadata.get("tool_name")
+    if tool_name and conv_id:
+        _pending_confirmations[conv_id] = {
+            "tool_name": tool_name,
+            "tool_args": metadata.get("tool_args", {}) or {},
+            "_fast_path": False,
+        }
     return json.dumps({
         "content": "⏸️ Требуется подтверждение",
-        "metadata": confirmation_metadata(conv_id, state, user_text, user_role, target_env),
+        "metadata": metadata,
     })
 # #endregion AgentChat.Confirmation.Payload
 
@@ -336,130 +365,162 @@ async def _format_tool_output_via_llm(
 
 # #region AgentChat.Confirmation.HandleResume [C:4] [TYPE Function] [SEMANTICS agent-chat,hitl,resume,streaming]
 # @ingroup AgentChat
-# @BRIEF Resume from HITL checkpoint — execute confirmed tool or abort on deny.
+# @BRIEF Resume from HITL checkpoint — continue the LangGraph run or abort on deny.
 # @PRE conversation_id is valid. action is "confirm" or "deny".
 # @POST Streams confirm_resolved, tool_start, tool_end/tool_error events via yield.
 # @SIDE_EFFECT Invokes LangChain tools; modifies _pending_confirmations dict.
 # @RELATION DEPENDS_ON -> [AgentChat.LangGraph.Setup]
 # @DATA_CONTRACT Input: (conv_id, action, user_jwt, env_id) -> Output: AsyncGenerator[str]
-# @RATIONALE Fast-path resume (direct tool execution via _pending_confirmations dict)
-#   chosen because the HITL confirmation payload already contains serialised tool
-#   name + args — re-entering LangGraph to invoke the same tool is redundant.
-#   Bypasses ~1-3s of LangGraph overhead (agent init, state reconstruction, tool
-#   re-selection) per resume. Falls back to full LangGraph checkpoint resume when
-#   _pending_confirmations is empty (e.g. after container restart).
-# @REJECTED ALWAYS checkpoint resume via create_agent(interrupt_before=[]) was
-#   rejected — adds 1-3s latency to every resume for no reliability gain when
-#   _pending_confirmations is populated. The full checkpoint path is preserved as
-#   the fallback, providing defense-in-depth for container restart scenarios.
+# @RATIONALE Resume ALWAYS continues the LangGraph checkpoint (create_agent +
+#   astream_events(None)) so multi-step scenario runs keep building after a
+#   confirmation. When the checkpoint resume itself fails (e.g. INVALID_CHAT_HISTORY
+#   from a stale run with unanswered tool calls), the still-pending tools are
+#   executed directly as a fallback and the checkpoint is repaired with synthetic
+#   ToolMessages so the thread stays usable. _pending_confirmations carries only a
+#   title/args marker for descriptive conversation titles.
+# @REJECTED Fast-path single-tool resume (direct tool execution + format, no graph
+#   continuation) — rejected because it stalls multi-step scenario flows after the
+#   first confirmation.
 # @REJECTED Pure streaming without checkpoint — would lose unconfirmed operations
 #   on crash with no rollback capability.
 async def handle_resume(  # noqa: C901
     conversation_id: str, action: str,
     user_jwt: str = "", env_id: str | None = None,
 ) -> AsyncGenerator[str]:
-    from ss_tools.agent.context import reset_user_jwt, set_user_jwt
+    from ss_tools.agent.context import reset_env_id, reset_user_jwt, set_env_id, set_user_jwt
     from ss_tools.shared.logger import logger
 
     user_jwt_token = set_user_jwt(user_jwt)
+    env_token = set_env_id(env_id or "")
     try:
-        pending = _pending_confirmations.pop(conversation_id, None)
-        if pending is not None:
-            if action == "deny":
-                yield json.dumps({
-                    "content": "⏹️ Операция отменена",
-                    "metadata": {"type": "confirm_resolved", "result": "denied"},
-                })
-                return
-            if action == "confirm":
-                logger.reason(
-                    "Fast-path confirmation resume",
-                    payload={"tool": pending.get("tool_name"), "conv_id": conversation_id},
-                    extra={"src": "AgentChat.Confirmation"},
-                )
-                tool_name = str(pending.get("tool_name") or "unknown_action")
-                tool_args = normalize_tool_args(pending.get("tool_args"))
-                yield json.dumps({
-                    "content": "▶️ Операция подтверждена",
-                    "metadata": {"type": "confirm_resolved", "result": "confirmed"},
-                })
-                yield json.dumps({
-                    "content": f"🛠️ {tool_name}",
-                    "metadata": {"type": "tool_start", "tool": tool_name, "input": tool_args},
-                })
-                tool_obj = find_tool(tool_name)
-                if tool_obj is None:
-                    error = f"Unknown tool: {tool_name}"
-                    logger.explore(
-                        "Unknown tool in resume",
-                        payload={"tool": tool_name}, error=error,
-                        extra={"src": "AgentChat.Confirmation"},
-                    )
-                    yield json.dumps({
-                        "content": f"❌ {tool_name} — {error}",
-                        "metadata": {"type": "tool_error", "tool": tool_name, "error": error},
-                    })
-                    return
-                try:
-                    output = await tool_obj.ainvoke(tool_args)
-                except Exception as exc:
-                    logger.explore(
-                        "Tool invocation failed in resume",
-                        payload={"tool": tool_name}, error=str(exc),
-                        extra={"src": "AgentChat.Confirmation"},
-                    )
-                    yield json.dumps({
-                        "content": f"❌ {tool_name} — {exc}",
-                        "metadata": {"type": "tool_error", "tool": tool_name, "error": str(exc)},
-                    })
-                    return
-                yield json.dumps({
-                    "content": f"✅ {tool_name}",
-                    "metadata": {"type": "tool_end", "tool": tool_name, "output": {"result": str(output)[:500]}},
-                })
-                # Format tool output via LLM for a human-readable response
-                async for chunk in _format_tool_output_via_llm(tool_name, str(output)):
-                    yield chunk
-                logger.reflect(
-                    "Fast-path confirmation completed",
-                    payload={"tool": tool_name},
-                    extra={"src": "AgentChat.Confirmation"},
-                )
-                return
+        # Consume the title/args marker stored by confirmation_payload (used by
+        # app.py to build a descriptive conversation title). Resume ALWAYS continues
+        # the LangGraph checkpoint so multi-step scenario runs keep building after a
+        # confirmation; executing a single tool and stopping (the former fast-path)
+        # would stall those flows, so that path has been removed.
+        _pending_confirmations.pop(conversation_id, None)
 
         logger.reason(
             "LangGraph checkpoint resume",
             payload={"conv_id": conversation_id, "action": action},
             extra={"src": "AgentChat.Confirmation"},
         )
-        agent = await create_agent(get_all_tools(), env_id, interrupt_before=[])
+        # Resume with the SAME env-injection parity as the original run so tool
+        # calls stored in the checkpoint (raw LLM args, without env_id) still get
+        # environment_id filled in by the runtime wrapper.
+        try:
+            from ss_tools.agent.app import _inject_env_id_into_tools
+            resume_tools = _inject_env_id_into_tools(get_all_tools(), env_id)
+        except Exception:
+            resume_tools = get_all_tools()
+        agent = await create_agent(resume_tools, env_id, interrupt_before=[])
         if action == "confirm":
             config = {"configurable": {"thread_id": conversation_id}}
             yield json.dumps({
                 "content": "▶️ Операция подтверждена",
                 "metadata": {"type": "confirm_resolved", "result": "confirmed"},
             })
-            async for event in agent.astream_events(None, config=config, version="v2"):
-                kind = event.get("event")
-                if kind == "on_chat_model_stream":
-                    chunk = event["data"]["chunk"]
-                    if hasattr(chunk, "content") and chunk.content:
+            try:
+                async for event in agent.astream_events(None, config=config, version="v2"):
+                    kind = event.get("event")
+                    if kind == "on_chat_model_stream":
+                        chunk = event["data"]["chunk"]
+                        if hasattr(chunk, "content") and chunk.content:
+                            yield json.dumps({
+                                "content": chunk.content,
+                                "metadata": {"type": "stream_token", "token": chunk.content},
+                            })
+                    elif kind == "on_tool_start":
+                        tool_name = event["name"]
                         yield json.dumps({
-                            "content": chunk.content,
-                            "metadata": {"type": "stream_token", "token": chunk.content},
+                            "content": f"🛠️ {tool_name}",
+                            "metadata": {"type": "tool_start", "tool": tool_name, "input": event["data"].get("input", {})},
                         })
-                elif kind == "on_tool_start":
-                    tool_name = event["name"]
+                    elif kind == "on_tool_end":
+                        tool_name = event["name"]
+                        output = event["data"].get("output", "")
+                        yield json.dumps({
+                            "content": f"✅ {tool_name}",
+                            "metadata": {"type": "tool_end", "tool": tool_name, "output": {"result": str(output)[:500]}},
+                        })
+            except Exception as exc:
+                # Checkpoint may contain an AI message whose tool call has no
+                # ToolMessage (interrupt_before=["tools"] or a crashed prior run).
+                # LangGraph then raises INVALID_CHAT_HISTORY in the model node and
+                # the whole resume would die. Fall back to executing the still-
+                # pending tool calls directly so the user still gets a response.
+                logger.explore(
+                    "Checkpoint resume failed — falling back to direct tool execution",
+                    payload={"conv_id": conversation_id},
+                    error=str(exc),
+                    extra={"src": "AgentChat.Confirmation"},
+                )
+                state = await agent.aget_state(config)
+                pending = _pending_tool_calls(state)
+                if pending:
+                    state_messages = list(state.values.get("messages", []))
+                    repaired_msgs: list[ToolMessage] = []
+                    for tool_name, tool_args, tcid in pending:
+                        yield json.dumps({
+                            "content": f"🛠️ {tool_name}",
+                            "metadata": {"type": "tool_start", "tool": tool_name, "input": tool_args},
+                        })
+                        tool_obj = find_tool(tool_name)
+                        if tool_obj is None:
+                            err = f"Unknown tool: {tool_name}"
+                            logger.explore("Unknown tool in resume fallback",
+                                           payload={"tool": tool_name}, error=err,
+                                           extra={"src": "AgentChat.Confirmation"})
+                            repaired_msgs.append(ToolMessage(
+                                content=f"Error: {err}", tool_call_id=tcid, name=tool_name))
+                            yield json.dumps({
+                                "content": f"❌ {tool_name} — {err}",
+                                "metadata": {"type": "tool_error", "tool": tool_name, "error": err},
+                            })
+                            continue
+                        try:
+                            output = await tool_obj.ainvoke(tool_args)
+                        except Exception as tool_exc:
+                            logger.explore("Tool invocation failed in resume fallback",
+                                           payload={"tool": tool_name}, error=str(tool_exc),
+                                           extra={"src": "AgentChat.Confirmation"})
+                            repaired_msgs.append(ToolMessage(
+                                content=f"Error: {tool_exc}", tool_call_id=tcid, name=tool_name))
+                            yield json.dumps({
+                                "content": f"❌ {tool_name} — {tool_exc}",
+                                "metadata": {"type": "tool_error", "tool": tool_name, "error": str(tool_exc)},
+                            })
+                            continue
+                        repaired_msgs.append(ToolMessage(
+                            content=str(output), tool_call_id=tcid, name=tool_name))
+                        yield json.dumps({
+                            "content": f"✅ {tool_name}",
+                            "metadata": {"type": "tool_end", "tool": tool_name, "output": {"result": str(output)[:500]}},
+                        })
+                        async for chunk in _format_tool_output_via_llm(tool_name, str(output)):
+                            yield chunk
+                    # Repair the checkpoint: answer the pending tool calls with
+                    # ToolMessages so the thread is consistent and the NEXT user
+                    # message does not fail LangGraph INVALID_CHAT_HISTORY, and a
+                    # repeated confirm cannot re-execute the same tool calls.
+                    if repaired_msgs:
+                        try:
+                            agent.update_state(config, {"messages": [*state_messages, *repaired_msgs]})
+                            logger.reason(
+                                "Checkpoint repaired after resume fallback",
+                                payload={"count": len(repaired_msgs), "conv_id": conversation_id},
+                                extra={"src": "AgentChat.Confirmation"},
+                            )
+                        except Exception as repair_exc:
+                            logger.explore(
+                                "Checkpoint repair failed after resume fallback",
+                                payload={"conv_id": conversation_id}, error=str(repair_exc),
+                                extra={"src": "AgentChat.Confirmation"},
+                            )
+                else:
                     yield json.dumps({
-                        "content": f"🛠️ {tool_name}",
-                        "metadata": {"type": "tool_start", "tool": tool_name, "input": event["data"].get("input", {})},
-                    })
-                elif kind == "on_tool_end":
-                    tool_name = event["name"]
-                    output = event["data"].get("output", "")
-                    yield json.dumps({
-                        "content": f"✅ {tool_name}",
-                        "metadata": {"type": "tool_end", "tool": tool_name, "output": {"result": str(output)[:500]}},
+                        "content": f"❌ Ошибка возобновления: {exc}",
+                        "metadata": {"type": "error", "code": "PROCESSING_ERROR", "detail": str(exc)},
                     })
         elif action == "deny":
             logger.reflect(
@@ -473,5 +534,6 @@ async def handle_resume(  # noqa: C901
             })
     finally:
         reset_user_jwt(user_jwt_token)
+        reset_env_id(env_token)
 # #endregion AgentChat.Confirmation.HandleResume
 # #endregion AgentChat.Confirmation

agent/src/ss_tools/agent/_tool_resolver.py

@@ -7,6 +7,8 @@
 
 from typing import Any
 
+from langchain_core.messages import AIMessage, ToolMessage
+
 _GRAPH_NODE_NAMES = {"agent", "tools", "__start__", "__end__"}
 
 
@@ -56,10 +58,47 @@ def coerce_tool_call(tool_call: Any) -> tuple[str | None, dict[str, Any]]:
 # #endregion AgentChat.ToolResolver.CoerceCall
 
 
+# #region AgentChat.ToolResolver.PendingCalls [C:2] [TYPE Function] [SEMANTICS agent-chat,tools,pending,state]
+# @ingroup AgentChat
+# @BRIEF Extract ALL tool calls from a LangGraph state that are missing a ToolMessage.
+# @POST Returns list of (tool_name, tool_args, tool_call_id) — empty when the
+#       checkpoint history is consistent.
+# @RATIONALE Single source of truth for "which tool calls are still pending" across
+#   the confirmation metadata (the gated tool is the first pending call) and the
+#   resume fallback (all pending calls are executed directly). Divergent parsers
+#   would let the confirmation title name a different tool than the fallback runs.
+def pending_tool_calls_from_state(state: Any) -> list[tuple[str, dict[str, Any], str]]:
+    messages = list(state.values.get("messages", [])) if hasattr(state, "values") else []
+    executed_ids = {
+        getattr(m, "tool_call_id", None)
+        for m in messages
+        if isinstance(m, ToolMessage)
+    }
+    pending: list[tuple[str, dict[str, Any], str]] = []
+    for m in messages:
+        if not isinstance(m, AIMessage):
+            continue
+        for tc in getattr(m, "tool_calls", None) or []:
+            tcid = tc.get("id") if isinstance(tc, dict) else getattr(tc, "id", None)
+            tname = tc.get("name") if isinstance(tc, dict) else getattr(tc, "name", None)
+            targs = tc.get("args") if isinstance(tc, dict) else getattr(tc, "args", None)
+            if tcid and tname and tcid not in executed_ids:
+                pending.append((str(tname), normalize_tool_args(targs), str(tcid)))
+    return pending
+# #endregion AgentChat.ToolResolver.PendingCalls
+
+
 # #region AgentChat.ToolResolver.ExtractCall [C:2] [TYPE Function] [SEMANTICS agent-chat,tools,extract,state]
 # @ingroup AgentChat
-# @BRIEF Extract pending tool call from LangGraph state messages.
+# @BRIEF Extract the pending tool call from LangGraph state messages.
 def extract_tool_call_from_state(state, user_text: str = "") -> tuple[str | None, dict[str, Any]]:
+    # Prefer the first unanswered tool call (the one the HITL gate is blocking).
+    # This keeps the confirmation marker/title in sync with what the resume will
+    # actually execute, unlike a raw "last AI message, first call" scan which can
+    # name an already-answered call or skip an older pending one.
+    pending = pending_tool_calls_from_state(state)
+    if pending:
+        return pending[0][0], pending[0][1]
     known_tools = known_agent_tool_names()
     try:
         messages = (state.values.get("messages") if hasattr(state, "values") else []) or []

agent/src/ss_tools/agent/app.py

@@ -59,7 +59,14 @@ from ss_tools.agent._persistence import (
     prefetch_databases,
     save_conversation,
 )
-from ss_tools.agent.context import reset_user_jwt, reset_user_role, set_user_jwt, set_user_role
+from ss_tools.agent.context import (
+    reset_env_id,
+    reset_user_jwt,
+    reset_user_role,
+    set_env_id,
+    set_user_jwt,
+    set_user_role,
+)
 from ss_tools.agent.document_parser import parse_upload
 from ss_tools.agent.langgraph_setup import create_agent, llm_diagnostics
 from ss_tools.agent.middleware import (
@@ -293,18 +300,33 @@ def _inject_uicontext(runtime_context: str, uicontext: dict) -> str:
 # @BRIEF Wrap tools to auto-inject environment_id from runtime context when LLM omits it.
 # @POST Tool functions injected with env_id; args_schema input intercepts before validation.
 # @RATIONALE LLM often omits environment_id even when system prompt instructs it. Auto-injection
-#   makes tools resilient — missing env_id gets filled from the runtime context.
-# @SIDE_EFFECT Mutates tool objects (_parse_input, coroutine).
+#   makes tools resilient — missing env_id gets filled from the request-local context.
+# @SIDE_EFFECT Mutates tool objects (_parse_input, coroutine) ONCE per tool.
+# @INVARIANT Idempotent: each tool is wrapped at most once; the wrappers resolve the
+#   environment from the request-local ContextVar (set_env_id/get_env_id) at call time,
+#   so repeated _inject_env_id_into_tools calls do not stack wrappers and concurrent
+#   conversations with different environments do not bleed env into each other.
 def _inject_env_id_into_tools(tools: list, env_id: str | None) -> list:
     """Wrap tools so environment_id is auto-injected from runtime context when LLM omits it.
 
     Works by intercepting _parse_input (before args_schema validation) and injecting
-    env_id into the tool input dict. Also wraps coroutine as a safety net.
+    env_id into the tool input dict. Also wraps coroutine as a safety net. The env is
+    read from the request-local ContextVar (get_env_id) at invocation time, and tools
+    are wrapped at most once (marked with _env_aware_injected), so repeated calls on
+    the shared tool singletons (send path + handle_resume) do not accumulate wrapper
+    layers or leak one conversation's environment into another.
     """
+    from ss_tools.agent.context import get_env_id
+
     if not env_id:
         return tools
 
     for tool in tools:
+        # Idempotent: skip tools already wrapped by a previous call (the wrapper
+        # resolves the env from request-local context, so re-wrapping is unnecessary
+        # and would only stack closures on the shared singletons).
+        if getattr(tool, "_env_aware_injected", False):
+            continue
         # Only wrap tools that accept 'environment_id' parameter
         sig = inspect.signature(tool.coroutine if tool.coroutine else (tool.func if tool.func else tool._run))
         if "environment_id" not in sig.parameters:
@@ -314,32 +336,42 @@ def _inject_env_id_into_tools(tools: list, env_id: str | None) -> list:
         # Use UNBOUND class method to avoid double-binding issues
         orig_unbound = type(tool)._parse_input
 
-        def _make_parse_wrapper(orig_fn, eid: str):
-            """Create a wrapper that injects env_id into tool_input dict."""
+        def _make_parse_wrapper(orig_fn):
+            """Create a wrapper that injects env_id from request-local context."""
 
             @functools.wraps(orig_fn)
             def wrapped(self, tool_input, tool_call_id=None):
-                if isinstance(tool_input, dict):
-                    if tool_input.get("environment_id") is None:
+                if isinstance(tool_input, dict) and tool_input.get("environment_id") is None:
+                    eid = get_env_id()
+                    if eid:
                         tool_input = {**tool_input, "environment_id": eid}
                 return orig_fn(self, tool_input, tool_call_id)
 
             return wrapped
 
-        tool._parse_input = _make_parse_wrapper(orig_unbound, env_id).__get__(tool, type(tool))
+        tool._parse_input = _make_parse_wrapper(orig_unbound).__get__(tool, type(tool))
 
         # Wrap coroutine as safety net
         original_coro = tool.coroutine
         if original_coro:
 
             @functools.wraps(original_coro)
-            async def env_aware_coro(*args, **kwargs):
-                if kwargs.get("environment_id") is None and env_id:
-                    kwargs["environment_id"] = env_id
-                return await original_coro(*args, **kwargs)
+            async def env_aware_coro(*args, _orig=original_coro, **kwargs):
+                # _orig binds original_coro AT DEFINITION TIME — the loop variable
+                # is rebound each iteration, and a closure over `original_coro`
+                # would otherwise make every wrapper call the LAST tool's coroutine
+                # (all env-injected tools silently became create_verification_run_tool,
+                # crashing tool execution with unexpected/missing keyword errors).
+                if kwargs.get("environment_id") is None:
+                    eid = get_env_id()
+                    if eid:
+                        kwargs["environment_id"] = eid
+                return await _orig(*args, **kwargs)
 
             tool.coroutine = env_aware_coro
 
+        tool._env_aware_injected = True
+
     return tools
 
 
@@ -372,6 +404,14 @@ async def agent_handler(  # noqa: C901 — intentionally complex C4 orchestratio
     """
     # ── Auth: user JWT passed from frontend via additional_input —─
     user_jwt_str = user_jwt_str_param or ""
+    if not user_jwt_str:
+        # Gradio may omit the hidden additional input on reconnects. Preserve
+        # the browser bearer token for tool calls and lifecycle audit writes,
+        # without logging or placing it in the event payload.
+        request_headers = getattr(request, "headers", {}) or {}
+        authorization = request_headers.get("authorization") or request_headers.get("Authorization")
+        if isinstance(authorization, str) and authorization.lower().startswith("bearer "):
+            user_jwt_str = authorization[7:].strip()
     token_payload: dict[str, Any] = {}
     if user_jwt_str:
         try:
@@ -382,12 +422,16 @@ async def agent_handler(  # noqa: C901 — intentionally complex C4 orchestratio
     user_jwt_token = set_user_jwt(user_jwt_str)
     user_role = token_payload.get("role") or token_payload.get("user_role") or "viewer"
     user_role_token = set_user_role(user_role)
+    # Request-local env for tool auto-injection (read by the env-aware wrappers at
+    # call time, so concurrent conversations with different envs stay isolated).
+    env_token = set_env_id(env_id or "")
 
     # ── Per-user lock ──
     user_id = user_id_str or (extract_user_id(user_jwt_str) if user_jwt_str else "admin")
     if _user_locks.get(user_id, False):
         reset_user_jwt(user_jwt_token)
         reset_user_role(user_role_token)
+        reset_env_id(env_token)
         yield json.dumps({"metadata": {"type": "error", "code": "CONCURRENT_SEND", "detail": "Другой запрос уже обрабатывается. Дождитесь завершения перед отправкой нового."}})
         return
     _user_locks[user_id] = True
@@ -936,6 +980,7 @@ async def agent_handler(  # noqa: C901 — intentionally complex C4 orchestratio
             del _conv_locks[conv_id]
         reset_user_jwt(user_jwt_token)
         reset_user_role(user_role_token)
+        reset_env_id(env_token)
 
 
 # #endregion AgentChat.GradioApp.Handler

agent/src/ss_tools/agent/context.py

@@ -13,6 +13,7 @@ from contextvars import ContextVar, Token
 _user_jwt: ContextVar[str] = ContextVar("agent_user_jwt", default="")
 _service_jwt: ContextVar[str] = ContextVar("agent_service_jwt", default="")
 _user_role: ContextVar[str] = ContextVar("agent_user_role", default="viewer")
+_env_id: ContextVar[str] = ContextVar("agent_env_id", default="")
 
 
 # #region AgentChat.Context.SetUserJwt [C:1] [TYPE Function] [SEMANTICS agent-chat,context,jwt,set]
@@ -30,6 +31,21 @@ def get_user_jwt() -> str:
 # #endregion AgentChat.Context.GetUserJwt
 
 
+# #region AgentChat.Context.SetEnvId [C:1] [TYPE Function] [SEMANTICS agent-chat,context,env,set]
+# @BRIEF Store request-local environment ID in a ContextVar for tool env auto-injection.
+# @POST Returns a reset token for restoring the previous request context.
+def set_env_id(env_id: str) -> Token[str]:
+    return _env_id.set(env_id or "")
+# #endregion AgentChat.Context.SetEnvId
+
+
+# #region AgentChat.Context.GetEnvId [C:1] [TYPE Function] [SEMANTICS agent-chat,context,env,get]
+# @BRIEF Retrieve request-local environment ID for tool env auto-injection.
+def get_env_id() -> str:
+    return _env_id.get()
+# #endregion AgentChat.Context.GetEnvId
+
+
 # #region AgentChat.Context.SetUserRole [C:1] [TYPE Function] [SEMANTICS agent-chat,context,role,set]
 # @BRIEF Store request-local user role for RBAC enforcement in tool pipeline.
 # @POST Returns a reset token for restoring the previous request context.
@@ -64,15 +80,37 @@ def get_service_jwt() -> str:
 # @BRIEF Restore request-local JWT and role values after a request completes.
 # @PRE Tokens were returned by the corresponding set_* functions in the same context.
 # @POST Previous ContextVar values are restored; concurrent request contexts remain isolated.
+# @RATIONALE ContextVar.reset(token) raises ValueError when the async generator is
+#   closed (GeneratorExit) from a different asyncio context than the one where the
+#   token was created — e.g. gradio closing a SSE stream mid-yield. The reset is
+#   best-effort: the offending context is being torn down, so there is nothing to
+#   restore, and swallowing the error prevents the cleanup crash from masking the
+#   actual stream result.
 def reset_user_jwt(token: Token[str]) -> None:
-    _user_jwt.reset(token)
+    try:
+        _user_jwt.reset(token)
+    except ValueError:
+        pass
 
 
 def reset_user_role(token: Token[str]) -> None:
-    _user_role.reset(token)
+    try:
+        _user_role.reset(token)
+    except ValueError:
+        pass
 
 
 def reset_service_jwt(token: Token[str]) -> None:
-    _service_jwt.reset(token)
+    try:
+        _service_jwt.reset(token)
+    except ValueError:
+        pass
+
+
+def reset_env_id(token: Token[str]) -> None:
+    try:
+        _env_id.reset(token)
+    except ValueError:
+        pass
 # #endregion AgentChat.Context.Reset
 # #endregion AgentChat.Context

agent/src/ss_tools/agent/langgraph_setup.py

@@ -21,7 +21,7 @@ from psycopg.rows import dict_row
 import pydantic as _pydantic
 import pydantic_core as _pydantic_core
 
-from ss_tools.agent._config import AGENT_CONFIRM_TOOLS, AGENT_INTERRUPT_BEFORE as _INTERRUPT_BEFORE, FASTAPI_URL
+from ss_tools.agent._config import AGENT_CONFIRM_TOOLS, AGENT_INTERRUPT_BEFORE as _INTERRUPT_BEFORE, FASTAPI_URL, SERVICE_JWT
 from ss_tools.agent._llm_params import chat_openai_kwargs
 from ss_tools.shared._llm_http import get_shared_http_client
 from ss_tools.shared.logger import logger
@@ -189,8 +189,12 @@ async def _fetch_llm_config() -> dict | None:
     )
     try:
         fastapi_url = FASTAPI_URL
+        headers = {}
+        service_token = (SERVICE_JWT or "").strip()
+        if service_token:
+            headers["Authorization"] = f"Bearer {service_token}"
         client = get_shared_http_client(timeout=10)
-        resp = await client.get(f"{fastapi_url}/api/agent/llm-config")
+        resp = await client.get(f"{fastapi_url}/api/agent/llm-config", headers=headers)
         if resp.status_code == 200:
             config = resp.json()
             if config.get("configured"):

agent/src/ss_tools/agent/tools.py

@@ -1299,7 +1299,7 @@ class ExecuteDashboardResultInput(BaseModel):
     dashboard_id: int = Field(..., description="Superset dashboard ID")
     chart_id: int | None = Field(None, description="Chart ID to execute (required for chart queries)")
     dataset_id: int | None = Field(None, description="Dataset ID (alternative to chart_id)")
-    result_key: str = Field(..., description="Metric key to extract from result")
+    result_key: str | None = Field(None, description="Metric key to extract from result (optional — pass the metric name from the dashboard query model; omit if unknown)")
     normalized_filters_json: str = Field(default="", description="JSON string of normalized filter context (from normalize_filters)")
 # #endregion AgentChat.Tools.ExecuteDashboardResultInput
 
@@ -1309,11 +1309,15 @@ class ExecuteDashboardResultInput(BaseModel):
 # @BRIEF Execute a Superset-native chart/dataset query and return normalized result.
 # @PRE User authenticated, dashboard/chart accessible.
 # @POST Returns normalized metric value — no SQL injection possible.
+# @RATIONALE result_key is optional because the LLM cannot always know the metric
+#   keys ahead of time; a missing key falls back to the raw record instead of
+#   crashing the tool (previously a hard TypeError/ValidationError aborted the
+#   whole HITL resume stream for dashboard-testing scenarios).
 @tool(args_schema=ExecuteDashboardResultInput)
 async def execute_dashboard_result(
     environment_id: str,
     dashboard_id: int,
-    result_key: str,
+    result_key: str | None = None,
     chart_id: int | None = None,
     dataset_id: int | None = None,
     normalized_filters_json: str = "",
@@ -1321,7 +1325,7 @@ async def execute_dashboard_result(
     """Execute a dashboard chart query through Superset-native APIs — no SQL."""
     logger.reason("Execute dashboard result",
                   payload={"environment_id": environment_id, "dashboard_id": dashboard_id,
-                           "chart_id": chart_id, "result_key": result_key},
+                           "chart_id": chart_id, "result_key": result_key or None},
                   extra={"src": "AgentChat.Tools.ExecuteDashboardResult"})
 
     # Parse normalized filters if provided
@@ -1336,7 +1340,7 @@ async def execute_dashboard_result(
     body: dict[str, Any] = {
         "environment_id": environment_id,
         "dashboard_id": dashboard_id,
-        "result_key": result_key,
+        "result_key": result_key or "",
         "normalized_filters": normalized_filters or {"schema_version": 1, "filters": [], "filters_hash": "sha256:empty"},
     }
     if chart_id:

agent/tests/agent/test_confirmation.py

@@ -239,80 +239,58 @@ class TestFormatToolOutput:
 # ═══════════════════════════════════════════════════════════════════
 
 # #region Test.AgentChat.TestHandleResumeIntegration [C:3] [TYPE Class]
-# @BRIEF Integration tests for handle_resume — fast-path confirm/deny, error paths,
-#         LLM formatting integration, and title race-condition coverage.
+# @BRIEF Integration tests for handle_resume — checkpoint resume, error paths,
+#         fallback direct tool execution, and title race-condition coverage.
 class TestHandleResumeIntegration:
     @pytest.mark.asyncio
     async def test_deny_yields_cancelled(self):
+        """Deny always goes through the LangGraph checkpoint path and yields the
+        cancel event; the pending title marker is consumed."""
         from ss_tools.agent._confirmation import handle_resume, _pending_confirmations
         _pending_confirmations["conv-deny"] = {
             "tool_name": "list_environments",
             "tool_args": {},
+            "_fast_path": False,
         }
-        chunks = await _collect(handle_resume("conv-deny", "deny"))
+
+        mock_agent = MagicMock()
+        with patch("ss_tools.agent._confirmation.create_agent", return_value=mock_agent), \
+             patch("ss_tools.agent._confirmation.get_all_tools", return_value=[]):
+            chunks = await _collect(handle_resume("conv-deny", "deny"))
         data = _json_chunks(chunks)
         assert len(data) == 1
         assert data[0]["metadata"]["type"] == "confirm_resolved"
         assert data[0]["metadata"]["result"] == "denied"
         assert "отменена" in data[0]["content"].lower()
-        # Pending is popped
+        # Pending marker is popped
         assert "conv-deny" not in _pending_confirmations
 
     @pytest.mark.asyncio
-    async def test_confirm_executes_tool_and_formats_via_llm(self):
+    async def test_fallback_unknown_tool_yields_error_and_repairs_checkpoint(self):
+        """When the checkpoint resume fails and the pending tool is unknown, the
+        fallback yields tool_error and still appends a ToolMessage so the thread
+        stays consistent."""
+        from langchain_core.messages import AIMessage, HumanMessage
         from ss_tools.agent._confirmation import handle_resume, _pending_confirmations
 
-        tool = MagicMock()
-        tool.ainvoke = AsyncMock(return_value='[{"id":"ss-dev","name":"Dev"}]')
-        _pending_confirmations["conv-confirm"] = {
-            "tool_name": "list_environments",
-            "tool_args": {},
-        }
-
-        async def collect():
-            with patch("ss_tools.agent._confirmation.find_tool", return_value=tool), \
-                 patch("ss_tools.agent._confirmation._format_tool_output_via_llm") as mock_fmt:
-                async def _fake_fmt(tool_name, output):
-                    yield json.dumps({
-                        "content": f"SUMMARY: {tool_name} -> {output[:20]}",
-                        "metadata": {"type": "stream_token", "token": "X"},
-                    })
-                mock_fmt.side_effect = _fake_fmt
-                chunks = [c async for c in handle_resume("conv-confirm", "confirm")]
-            return chunks
-
-        chunks = await collect()
-        data = _json_chunks(chunks)
-
-        types = [d["metadata"]["type"] for d in data]
-        assert "confirm_resolved" in types
-        assert "tool_start" in types
-        assert "tool_end" in types
-        assert "stream_token" in types
-
-        # Check tool_end has the output summary
-        tool_end = [d for d in data if d["metadata"]["type"] == "tool_end"]
-        assert len(tool_end) == 1
-        assert tool_end[0]["metadata"]["output"]["result"].startswith('[{"id":"ss-dev"')
-
-        # Check LLM formatting chunk
-        stream_tokens = [d for d in data if d["metadata"]["type"] == "stream_token"]
-        assert len(stream_tokens) >= 1
-        assert "SUMMARY" in stream_tokens[0]["content"]
-
-        # Pending is popped
-        assert "conv-confirm" not in _pending_confirmations
-
-    @pytest.mark.asyncio
-    async def test_confirm_unknown_tool_yields_error(self):
-        from ss_tools.agent._confirmation import handle_resume, _pending_confirmations
+        pending_state = MagicMock()
+        pending_state.values.get.return_value = [
+            HumanMessage(content="start"),
+            AIMessage(content="", tool_calls=[{
+                "name": "nonexistent_tool_xyz", "args": {}, "id": "call_xyz", "type": "tool_call",
+            }]),
+        ]
+        mock_agent = MagicMock()
+        mock_agent.astream_events = MagicMock(side_effect=ValueError("INVALID_CHAT_HISTORY"))
+        mock_agent.aget_state = AsyncMock(return_value=pending_state)
+        mock_agent.update_state = MagicMock(return_value=None)
 
         _pending_confirmations["conv-unknown"] = {
-            "tool_name": "nonexistent_tool_xyz",
-            "tool_args": {},
+            "tool_name": "nonexistent_tool_xyz", "tool_args": {}, "_fast_path": False,
         }
-
-        with patch("ss_tools.agent._confirmation.find_tool", return_value=None):
+        with patch("ss_tools.agent._confirmation.create_agent", return_value=mock_agent), \
+             patch("ss_tools.agent._confirmation.get_all_tools", return_value=[]), \
+             patch("ss_tools.agent._confirmation.find_tool", return_value=None):
             chunks = await _collect(handle_resume("conv-unknown", "confirm"))
         data = _json_chunks(chunks)
 
@@ -321,23 +299,39 @@ class TestHandleResumeIntegration:
         assert "tool_start" in types
         assert "tool_error" in types
         assert "stream_token" not in types, "No LLM output on tool error"
-
         error_chunk = [d for d in data if d["metadata"]["type"] == "tool_error"]
         assert len(error_chunk) == 1
         assert "Unknown tool" in error_chunk[0]["metadata"]["error"]
+        # Checkpoint repair ran with a ToolMessage for the pending call.
+        assert mock_agent.update_state.called
+        repaired = mock_agent.update_state.call_args.args[1]["messages"]
+        assert repaired[-1].tool_call_id == "call_xyz"
 
     @pytest.mark.asyncio
-    async def test_confirm_tool_invocation_failure_yields_error(self):
-        from ss_tools.agent._confirmation import handle_resume, _pending_confirmations
+    async def test_fallback_tool_invocation_failure_yields_error_and_repairs(self):
+        """When the fallback tool invocation raises, tool_error is yielded and a
+        ToolMessage with the error content repairs the checkpoint."""
+        from langchain_core.messages import AIMessage, HumanMessage
+        from ss_tools.agent._confirmation import handle_resume
+
+        pending_state = MagicMock()
+        pending_state.values.get.return_value = [
+            HumanMessage(content="start"),
+            AIMessage(content="", tool_calls=[{
+                "name": "list_environments", "args": {}, "id": "call_env", "type": "tool_call",
+            }]),
+        ]
+        mock_agent = MagicMock()
+        mock_agent.astream_events = MagicMock(side_effect=ValueError("INVALID_CHAT_HISTORY"))
+        mock_agent.aget_state = AsyncMock(return_value=pending_state)
+        mock_agent.update_state = MagicMock(return_value=None)
 
         tool = MagicMock()
         tool.ainvoke = AsyncMock(side_effect=RuntimeError("API timeout"))
-        _pending_confirmations["conv-fail"] = {
-            "tool_name": "list_environments",
-            "tool_args": {},
-        }
 
-        with patch("ss_tools.agent._confirmation.find_tool", return_value=tool):
+        with patch("ss_tools.agent._confirmation.create_agent", return_value=mock_agent), \
+             patch("ss_tools.agent._confirmation.get_all_tools", return_value=[]), \
+             patch("ss_tools.agent._confirmation.find_tool", return_value=tool):
             chunks = await _collect(handle_resume("conv-fail", "confirm"))
         data = _json_chunks(chunks)
 
@@ -345,10 +339,14 @@ class TestHandleResumeIntegration:
         assert "tool_error" in types
         assert "tool_end" not in types, "No tool_end on failure"
         assert "stream_token" not in types, "No LLM output on failure"
-
         error_chunk = [d for d in data if d["metadata"]["type"] == "tool_error"]
         assert len(error_chunk) == 1
         assert "API timeout" in error_chunk[0]["metadata"]["error"]
+        # Checkpoint repaired with the error ToolMessage.
+        assert mock_agent.update_state.called
+        repaired = mock_agent.update_state.call_args.args[1]["messages"]
+        assert repaired[-1].tool_call_id == "call_env"
+        assert "Error: API timeout" in repaired[-1].content
 
     @pytest.mark.asyncio
     async def test_confirm_with_no_pending_falls_to_langgraph(self):
@@ -380,6 +378,153 @@ class TestHandleResumeIntegration:
         assert data[0]["metadata"]["type"] == "confirm_resolved"
         assert data[0]["metadata"]["result"] == "denied"
 
+    @pytest.mark.asyncio
+    async def test_checkpoint_resume_failure_falls_back_to_direct_tool_execution(self):
+        """When the LangGraph checkpoint resume raises (e.g. INVALID_CHAT_HISTORY
+        from a checkpoint whose tool call has no ToolMessage), handle_resume falls
+        back to executing the still-pending tool directly and streaming its result."""
+        from langchain_core.messages import AIMessage, HumanMessage
+        from ss_tools.agent._confirmation import handle_resume
+
+        pending_state = MagicMock()
+        pending_state.values.get.return_value = [
+            HumanMessage(content="start"),
+            AIMessage(content="", tool_calls=[{
+                "name": "list_environments",
+                "args": {},
+                "id": "call_env",
+                "type": "tool_call",
+            }]),
+        ]
+
+        mock_agent = MagicMock()
+        mock_agent.astream_events = MagicMock(side_effect=ValueError(
+            "Found AIMessages with tool_calls that do not have a corresponding ToolMessage"
+        ))
+        mock_agent.aget_state = AsyncMock(return_value=pending_state)
+
+        tool = MagicMock()
+        tool.ainvoke = AsyncMock(return_value='[{"id":"ss-dev"}]')
+
+        async def _fake_fmt(tool_name, output):
+            yield json.dumps({
+                "content": f"SUMMARY: {tool_name}",
+                "metadata": {"type": "stream_token", "token": "S"},
+            })
+
+        with patch("ss_tools.agent._confirmation.create_agent", return_value=mock_agent), \
+             patch("ss_tools.agent._confirmation.get_all_tools", return_value=[]), \
+             patch("ss_tools.agent._confirmation.find_tool", return_value=tool), \
+             patch("ss_tools.agent._confirmation._format_tool_output_via_llm") as mock_fmt:
+            mock_fmt.side_effect = _fake_fmt
+            chunks = await _collect(handle_resume("conv-fallback", "confirm"))
+
+        data = _json_chunks(chunks)
+        types = [d["metadata"]["type"] for d in data]
+        assert "confirm_resolved" in types
+        assert "tool_start" in types
+        assert "tool_end" in types
+        assert "stream_token" in types
+        tool_end = [d for d in data if d["metadata"]["type"] == "tool_end"]
+        assert tool_end[0]["metadata"]["output"]["result"].startswith('[{"id":"ss-dev"')
+        # No error event — the fallback produced a usable response
+        assert "error" not in types
+
+    @pytest.mark.asyncio
+    async def test_checkpoint_resume_failure_with_no_pending_yields_error(self):
+        """When the checkpoint resume fails and there are no pending tool calls,
+        a clear error event is yielded instead of a silent dead stream."""
+        from ss_tools.agent._confirmation import handle_resume
+
+        pending_state = MagicMock()
+        pending_state.values.get.return_value = []  # no messages
+
+        mock_agent = MagicMock()
+        mock_agent.astream_events = MagicMock(side_effect=ValueError("boom"))
+        mock_agent.aget_state = AsyncMock(return_value=pending_state)
+
+        with patch("ss_tools.agent._confirmation.create_agent", return_value=mock_agent), \
+             patch("ss_tools.agent._confirmation.get_all_tools", return_value=[]):
+            chunks = await _collect(handle_resume("conv-no-pending2", "confirm"))
+        data = _json_chunks(chunks)
+        types = [d["metadata"]["type"] for d in data]
+        assert "error" in types
+        assert data[-1]["metadata"]["code"] == "PROCESSING_ERROR"
+
+    @pytest.mark.asyncio
+    async def test_pending_tool_calls_skips_answered_calls(self):
+        """_pending_tool_calls only returns tool calls that lack a ToolMessage."""
+        from langchain_core.messages import AIMessage, HumanMessage, ToolMessage
+        from ss_tools.agent._confirmation import _pending_tool_calls
+
+        state = MagicMock()
+        state.values.get.return_value = [
+            HumanMessage(content="start"),
+            AIMessage(content="", tool_calls=[{
+                "name": "list_environments", "args": {}, "id": "call_env", "type": "tool_call",
+            }]),
+            ToolMessage(content="[ok]", tool_call_id="call_env", name="list_environments"),
+            AIMessage(content="", tool_calls=[{
+                "name": "get_health_summary", "args": {"x": 1}, "id": "call_health", "type": "tool_call",
+            }]),
+        ]
+        pending = _pending_tool_calls(state)
+        assert len(pending) == 1
+        assert pending[0][0] == "get_health_summary"
+        assert pending[0][1] == {"x": 1}
+        assert pending[0][2] == "call_health"
+
+    def test_confirmation_payload_stores_graph_resume_marker(self):
+        """confirmation_payload stores a title marker with _fast_path=False so
+        the multi-step scenario continues via the LangGraph checkpoint, while
+        app.py can still build a descriptive conversation title."""
+        from langchain_core.messages import AIMessage
+        from ss_tools.agent._confirmation import confirmation_payload, _pending_confirmations
+
+        state = MagicMock()
+        state.values.get.return_value = [
+            AIMessage(content="", tool_calls=[{
+                "name": "inspect_dashboard_query_model",
+                "args": {"dashboard_id": 3},
+                "id": "call_inspect", "type": "tool_call",
+            }]),
+        ]
+        out = confirmation_payload("conv-marker", state, "start")
+        data = json.loads(out)
+        assert data["metadata"]["type"] == "confirm_required"
+        marker = _pending_confirmations.pop("conv-marker", None)
+        assert marker is not None
+        assert marker["tool_name"] == "inspect_dashboard_query_model"
+        assert marker["tool_args"] == {"dashboard_id": 3}
+        assert marker["_fast_path"] is False
+
+    @pytest.mark.asyncio
+    async def test_resume_always_uses_langgraph_checkpoint(self):
+        """Resume always continues the LangGraph checkpoint (multi-step scenario
+        continuation); the pending marker is consumed for the title only."""
+        from ss_tools.agent._confirmation import handle_resume, _pending_confirmations
+
+        _pending_confirmations["conv-graph"] = {
+            "tool_name": "list_environments",
+            "tool_args": {},
+            "_fast_path": False,
+        }
+        mock_agent = MagicMock()
+        mock_agent.astream_events = MagicMock(return_value=_make_async_iter([]))
+        with patch("ss_tools.agent._confirmation.create_agent", return_value=mock_agent), \
+             patch("ss_tools.agent._confirmation.get_all_tools", return_value=[]):
+            chunks = await _collect(handle_resume("conv-graph", "confirm"))
+        data = _json_chunks(chunks)
+        types = [d["metadata"]["type"] for d in data]
+        # Confirm resolved then the graph resume path runs (create_agent called).
+        assert "confirm_resolved" in types
+        # The marker was consumed.
+        assert "conv-graph" not in _pending_confirmations
+        assert "tool_start" not in types, "Fast-path must be bypassed for graph markers"
+        assert "tool_error" not in types
+
+
+
     @pytest.mark.asyncio
     async def test_confirm_streams_langgraph_events_when_no_pending(self):
         """When no pending and LangGraph agent streams events, they are forwarded."""

agent/tests/test_agent/test_app.py

@@ -182,6 +182,8 @@ class TestConfirmationMetadata:
         assert meta["risk_level"] == "safe"
 
     def test_fast_resume_deny_closes_without_langgraph(self):
+        """Deny closes via the LangGraph checkpoint path and yields the cancel
+        event; the pending title marker is consumed."""
         from ss_tools.agent._confirmation import _pending_confirmations, handle_resume
 
         _pending_confirmations["conv-fast-deny"] = {
@@ -190,35 +192,45 @@ class TestConfirmationMetadata:
         }
 
         async def collect():
-            return [chunk async for chunk in handle_resume("conv-fast-deny", "deny")]
+            with patch("ss_tools.agent._confirmation.create_agent", return_value=MagicMock()), \
+                 patch("ss_tools.agent._confirmation.get_all_tools", return_value=[]):
+                return [chunk async for chunk in handle_resume("conv-fast-deny", "deny")]
 
         chunks = asyncio.run(collect())
         data = json.loads(chunks[0])
         assert data["metadata"] == {"type": "confirm_resolved", "result": "denied"}
+        assert "conv-fast-deny" not in _pending_confirmations
 
-    def test_fast_resume_confirm_executes_tool_directly(self):
+    def test_resume_confirm_continues_langgraph_checkpoint(self):
+        """Confirm always continues the LangGraph checkpoint (multi-step scenario
+        continuation) and yields the confirm_resolved event."""
         from ss_tools.agent._confirmation import _pending_confirmations, handle_resume
 
-        tool = MagicMock()
-        tool.ainvoke = AsyncMock(return_value='[{"id":"ss-dev"}]')
+        mock_chunk = MagicMock()
+        mock_chunk.content = "resumed output"
+        mock_agent = MagicMock()
+        mock_agent.astream_events = MagicMock(return_value=_make_async_iter([
+            {"event": "on_chat_model_stream", "data": {"chunk": mock_chunk}},
+            {"event": "on_tool_start", "name": "list_environments", "data": {"input": {}}},
+            {"event": "on_tool_end", "name": "list_environments", "data": {"output": "ok"}},
+        ]))
         _pending_confirmations["conv-fast-confirm"] = {
             "tool_name": "list_environments",
             "tool_args": {},
         }
 
         async def collect():
-            with patch("ss_tools.agent._confirmation.find_tool", return_value=tool), patch("ss_tools.agent._confirmation._format_tool_output_via_llm") as mock_format:
-                # Make the mock format helper yield nothing — don't need LLM in unit test
-                async def _empty_format(*_args, **_kwargs):
-                    return
-                    yield  # pragma: no cover — async generator requires at least one yield
-
-                mock_format.side_effect = _empty_format
+            with patch("ss_tools.agent._confirmation.create_agent", return_value=mock_agent), \
+                 patch("ss_tools.agent._confirmation.get_all_tools", return_value=[]):
                 return [chunk async for chunk in handle_resume("conv-fast-confirm", "confirm")]
 
         chunks = asyncio.run(collect())
         metadata_types = [json.loads(chunk)["metadata"]["type"] for chunk in chunks]
-        assert metadata_types[:3] == ["confirm_resolved", "tool_start", "tool_end"]
+        assert metadata_types[0] == "confirm_resolved"
+        assert "stream_token" in metadata_types
+        assert "tool_start" in metadata_types
+        assert "tool_end" in metadata_types
+        assert "conv-fast-confirm" not in _pending_confirmations
 
 
 # #endregion Test.AgentChat.TestConfirmationMetadata