add checkpointer checker id

src/backend/base/langflow/services/checkpoint/store.py

@@ -55,6 +55,18 @@ class JobScopedCheckpointStore(CheckpointStore, Service):
     async def teardown(self) -> None:
         pass
 
+    async def save_blob(self, job_id: str, kind: str, blob: str) -> None:
+        """Durably persist the agent-thread blob (LE-1447) on the job-scoped row."""
+        parsed = _as_uuid(job_id)
+        if parsed is not None:
+            await self._jobs.save_checkpoint(parsed, kind, blob)
+
+    async def load_blob(self, job_id: str, kind: str) -> str | None:
+        parsed = _as_uuid(job_id)
+        if parsed is None:
+            return None
+        return await self._jobs.load_checkpoint(parsed, kind)
+
     async def save(self, checkpoint: GraphCheckpoint) -> None:
         if checkpoint.job_id is None:
             msg = "GraphCheckpoint.job_id is required to persist a durable checkpoint"

src/lfx/src/lfx/components/models_and_agents/agent.py

@@ -6,7 +6,11 @@ from datetime import datetime, timezone
 from typing import TYPE_CHECKING, Any, cast
 
 from langchain.agents import create_agent
-from langchain.agents.middleware import ModelCallLimitMiddleware, ToolRetryMiddleware
+from langchain.agents.middleware import (
+    HumanInTheLoopMiddleware,
+    ModelCallLimitMiddleware,
+    ToolRetryMiddleware,
+)
 
 from lfx.components.models_and_agents.agent_helpers.graph_event_adapter import (
     adapt_graph_events_to_executor_shape,
@@ -220,6 +224,16 @@ class AgentComponent(ToolCallingAgentComponent):
             advanced=True,
             range_spec=RangeSpec(min=1, max=128000, step=1, step_type="int"),
         ),
+        StrInput(
+            name="tools_requiring_approval",
+            display_name="Tools Requiring Approval",
+            info=(
+                "Names of connected tools that pause for human approval before running "
+                "(human-in-the-loop). Leave empty to auto-approve every tool."
+            ),
+            is_list=True,
+            advanced=True,
+        ),
         MultilineInput(
             name="format_instructions",
             display_name="Output Format Instructions",
@@ -489,7 +503,7 @@ class AgentComponent(ToolCallingAgentComponent):
             prompt = prompt.replace(placeholder, value)
         return prompt
 
-    def create_agent_runnable(self):
+    def create_agent_runnable(self, *, allow_interrupts: bool = True):
         """Build the LangGraph `CompiledStateGraph` via `langchain.agents.create_agent`.
 
         Replaces the legacy `AgentExecutor` runnable inherited from
@@ -537,12 +551,14 @@ class AgentComponent(ToolCallingAgentComponent):
                 )
                 raise NotImplementedError(msg) from exc
 
-        middleware = self._build_middleware(llm)
+        middleware = self._build_middleware(llm, allow_interrupts=allow_interrupts)
+        checkpointer = self._build_agent_checkpointer() if allow_interrupts else None
         return create_agent(
             model=llm,
             tools=tools,
             system_prompt=self.system_prompt or "",
             middleware=middleware or None,
+            checkpointer=checkpointer,
         )
 
     def _compute_recursion_limit(self) -> int:
@@ -556,7 +572,7 @@ class AgentComponent(ToolCallingAgentComponent):
         run_limit = max(1, int(raw)) if raw is not None else 15
         return run_limit * 2 + 5
 
-    def _build_middleware(self, llm: Any) -> list:
+    def _build_middleware(self, llm: Any, *, allow_interrupts: bool = True) -> list:
         # `llm` is passed in (rather than re-fetched via `self._get_llm()`)
         # because some providers do credential resolution / client instantiation
         # lazily on each call. The caller — `create_agent_runnable` — already
@@ -589,8 +605,50 @@ class AgentComponent(ToolCallingAgentComponent):
         if is_watsonx_model(llm):
             middleware.append(SingleToolCallMiddleware())
             middleware.append(WatsonXPlaceholderMiddleware())
+        # Human-in-the-loop: attach only when a tool is gated AND interrupts are allowed
+        # (the structured-output path disables them), keeping ungated flows unchanged.
+        interrupt_on = self._gated_interrupt_on() if allow_interrupts else {}
+        if interrupt_on:
+            middleware.append(HumanInTheLoopMiddleware(interrupt_on=interrupt_on))
         return middleware
 
+    def _agent_thread_id(self) -> str | None:
+        """Per-run thread id for the agent HITL checkpoint (run_id, not session_id)."""
+        run_id = getattr(getattr(self, "graph", None), "run_id", None)
+        return str(run_id) if run_id else None
+
+    def _build_agent_checkpointer(self):
+        """Durable saver for a gated agent run, else None (no checkpointer overhead).
+
+        The blob store is the INJECTED checkpoint service (DB-backed in the Langflow
+        runtime, in-memory standalone) so lfx never imports langflow.
+        """
+        thread_id = self._agent_thread_id()
+        if not thread_id or not self._gated_interrupt_on():
+            return None
+        from lfx.components.models_and_agents.agent_helpers.job_checkpoint_saver import JobCheckpointSaver
+        from lfx.graph.checkpoint.store import default_checkpoint_store
+        from lfx.services.deps import get_checkpoint_service
+
+        try:
+            store = get_checkpoint_service()
+        except Exception:  # noqa: BLE001
+            store = None
+        store = store or default_checkpoint_store()
+        return JobCheckpointSaver(thread_id, store.save_blob, store.load_blob)
+
+    def _gated_interrupt_on(self) -> dict[str, bool]:
+        """Map each connected tool flagged for approval to a HITL interrupt entry.
+
+        Intersects the user-listed names with the real tool names so a stale name
+        never gates a tool that isn't wired.
+        """
+        requested = getattr(self, "tools_requiring_approval", None) or []
+        if not requested:
+            return {}
+        tool_names = {getattr(tool, "name", None) for tool in (self.tools or [])}
+        return {name: True for name in requested if name in tool_names}
+
     async def run_agent(self, agent) -> Message:
         """Run the LangGraph `CompiledStateGraph` and return the final agent Message.
 
@@ -623,19 +681,21 @@ class AgentComponent(ToolCallingAgentComponent):
         # for start/end overhead.
         recursion_limit = self._compute_recursion_limit()
 
+        agent_config: dict[str, Any] = {
+            "callbacks": [
+                AgentAsyncHandler(self.log),
+                token_usage_handler,
+                *self._get_shared_callbacks(),
+            ],
+            "recursion_limit": recursion_limit,
+        }
+        # The durable checkpointer keys on the per-run thread_id; it must be in the
+        # astream_events config (not just create_agent) for both initial run and resume.
+        thread_id = self._agent_thread_id()
+        if thread_id and self._gated_interrupt_on():
+            agent_config["configurable"] = {"thread_id": thread_id}
         stream = adapt_graph_events_to_executor_shape(
-            agent.astream_events(
-                input_dict,
-                config={
-                    "callbacks": [
-                        AgentAsyncHandler(self.log),
-                        token_usage_handler,
-                        *self._get_shared_callbacks(),
-                    ],
-                    "recursion_limit": recursion_limit,
-                },
-                version="v2",
-            )
+            agent.astream_events(input_dict, config=agent_config, version="v2")
         )
         try:
             result = await process_agent_events(

src/lfx/src/lfx/components/models_and_agents/agent_helpers/job_checkpoint_saver.py

@@ -0,0 +1,118 @@
+"""Durable LangGraph checkpoint saver for agent tool-approval HITL (LE-1447).
+
+Persists the paused agent thread (the interrupted checkpoint + its pending writes,
+including ``__interrupt__``) into the job-scoped durable store so an approval can be
+resumed after a restart. The langgraph checkpoint is NOT JSON-native (channel values
+hold raw messages; writes hold Interrupt tuples), so each value is encoded with the
+saver's own serde (``dumps_typed`` → ``('msgpack', bytes)``) and base64'd into the
+JSON blob — ``json.dumps``/``dumpd`` would lose the Interrupt.
+
+Only the latest checkpoint is kept (resume loads the latest), matching the observed
+contract: ``aput(ckpt)`` then ``aput_writes(__interrupt__)`` then ``aget_tuple`` →
+that checkpoint + the interrupt write. The store handle is INJECTED (two async
+callables) so lfx never imports langflow.
+"""
+
+from __future__ import annotations
+
+import asyncio
+import base64
+import json
+from collections.abc import Awaitable, Callable, Sequence
+from typing import Any
+
+from langgraph.checkpoint.base import BaseCheckpointSaver, CheckpointTuple
+
+_KIND = "agent"
+_ASYNC_ONLY = "JobCheckpointSaver is async-only; use the a* methods."
+
+SaveBlob = Callable[[str, str, str], Awaitable[None]]
+LoadBlob = Callable[[str, str], Awaitable[str | None]]
+
+
+class JobCheckpointSaver(BaseCheckpointSaver):
+    def __init__(self, job_id: str, save_blob: SaveBlob, load_blob: LoadBlob) -> None:
+        super().__init__()
+        self._job_id = job_id
+        self._save_blob = save_blob
+        self._load_blob = load_blob
+        self._lock = asyncio.Lock()
+
+    def _encode(self, value: Any) -> list[str]:
+        type_, data = self.serde.dumps_typed(value)
+        return [type_, base64.b64encode(data).decode("ascii")]
+
+    def _decode(self, pair: list[str]) -> Any:
+        type_, b64 = pair
+        return self.serde.loads_typed((type_, base64.b64decode(b64)))
+
+    async def _read(self) -> dict[str, Any]:
+        raw = await self._load_blob(self._job_id, _KIND)
+        return json.loads(raw) if raw else {}
+
+    @staticmethod
+    def _result_config(blob: dict[str, Any]) -> dict[str, Any]:
+        return {
+            "configurable": {
+                "thread_id": blob.get("thread_id"),
+                "checkpoint_ns": blob.get("checkpoint_ns", ""),
+                "checkpoint_id": blob.get("checkpoint_id"),
+            }
+        }
+
+    async def aput(self, config, checkpoint, metadata, new_versions) -> dict[str, Any]:  # noqa: ARG002
+        configurable = config.get("configurable", {})
+        async with self._lock:
+            # A new checkpoint supersedes the prior step; its writes start empty and
+            # accumulate via aput_writes (the interrupt write lands here next).
+            blob = {
+                "v": 1,
+                "thread_id": configurable.get("thread_id"),
+                "checkpoint_ns": configurable.get("checkpoint_ns", ""),
+                "checkpoint_id": checkpoint["id"],
+                "checkpoint": self._encode(checkpoint),
+                "metadata": self._encode(metadata),
+                "writes": [],
+            }
+            await self._save_blob(self._job_id, _KIND, json.dumps(blob))
+        return self._result_config(blob)
+
+    async def aput_writes(self, config, writes: Sequence[tuple[str, Any]], task_id: str, task_path: str = "") -> None:  # noqa: ARG002
+        async with self._lock:
+            blob = await self._read()
+            if not blob:
+                return
+            stored = blob.setdefault("writes", [])
+            for channel, value in writes:
+                stored.append([task_id, channel, *self._encode(value)])
+            await self._save_blob(self._job_id, _KIND, json.dumps(blob))
+
+    async def aget_tuple(self, config) -> CheckpointTuple | None:
+        blob = await self._read()
+        if "checkpoint" not in blob:
+            return None
+        requested = config.get("configurable", {}).get("checkpoint_id")
+        if requested is not None and requested != blob.get("checkpoint_id"):
+            return None
+        pending_writes = [(w[0], w[1], self._decode([w[2], w[3]])) for w in blob.get("writes", [])]
+        return CheckpointTuple(
+            config=self._result_config(blob),
+            checkpoint=self._decode(blob["checkpoint"]),
+            metadata=self._decode(blob["metadata"]),
+            parent_config=None,
+            pending_writes=pending_writes,
+        )
+
+    async def alist(self, config, *, filter=None, before=None, limit=None):  # noqa: A002, ARG002
+        tuple_ = await self.aget_tuple(config)
+        if tuple_ is not None:
+            yield tuple_
+
+    def put(self, *args, **kwargs):
+        raise NotImplementedError(_ASYNC_ONLY)
+
+    def put_writes(self, *args, **kwargs):
+        raise NotImplementedError(_ASYNC_ONLY)
+
+    def get_tuple(self, *args, **kwargs):
+        raise NotImplementedError(_ASYNC_ONLY)

src/lfx/src/lfx/graph/checkpoint/store.py

@@ -30,6 +30,21 @@ class CheckpointStore(ABC):
     @abstractmethod
     async def list_by_session(self, session_id: str) -> list[GraphCheckpoint]: ...
 
+    async def save_blob(self, job_id: str, kind: str, blob: str) -> None:
+        """Persist an opaque per-run blob keyed by ``(job_id, kind)``.
+
+        Used by the agent tool-approval saver (LE-1447, kind='agent'), separate from
+        the graph checkpoint. The default is in-memory (standalone/CLI); the durable
+        store overrides it to write the job-scoped table.
+        """
+        if not hasattr(self, "_blobs"):
+            self._blobs: dict[tuple[str, str], str] = {}
+        self._blobs[(job_id, kind)] = blob
+
+    async def load_blob(self, job_id: str, kind: str) -> str | None:
+        """Return the blob stored for ``(job_id, kind)``, or None."""
+        return getattr(self, "_blobs", {}).get((job_id, kind))
+
 
 def _expired(checkpoint: GraphCheckpoint) -> bool:
     return checkpoint.expires_at is not None and checkpoint.expires_at <= datetime.now(timezone.utc)

src/lfx/tests/unit/components/models_and_agents/test_agent_create_agent.py

@@ -1479,3 +1479,109 @@ async def test_should_pass_stream_true_to_get_llm_when_self_stream_toggle_is_tru
         component._get_llm()
 
     assert captured.get("stream") is True
+
+
+def test_should_attach_hitl_middleware_when_a_tool_is_gated() -> None:
+    """LE-1447 Slice 1: a tool listed for approval adds HumanInTheLoopMiddleware."""
+    from langchain.agents.middleware import HumanInTheLoopMiddleware
+
+    component = _build_component()
+    search = SimpleNamespace(name="search")
+    component.set_attributes({"tools": [search], "tools_requiring_approval": ["search"]})
+
+    middleware = component._build_middleware(MagicMock(name="fake_llm"))
+
+    hitl = [m for m in middleware if isinstance(m, HumanInTheLoopMiddleware)]
+    assert len(hitl) == 1
+    assert component._gated_interrupt_on() == {"search": True}
+
+
+def test_should_omit_hitl_middleware_when_no_tools_gated() -> None:
+    """No tool listed for approval keeps the existing graph shape (no HITL middleware)."""
+    from langchain.agents.middleware import HumanInTheLoopMiddleware
+
+    component = _build_component()
+    component.set_attributes(
+        {"tools": [SimpleNamespace(name="search")], "tools_requiring_approval": []}
+    )
+
+    middleware = component._build_middleware(MagicMock(name="fake_llm"))
+
+    assert not any(isinstance(m, HumanInTheLoopMiddleware) for m in middleware)
+
+
+def test_gated_interrupt_on_ignores_unknown_tool_names() -> None:
+    """A stale approval name never gates a tool that is not wired."""
+    component = _build_component()
+    component.set_attributes(
+        {"tools": [SimpleNamespace(name="search")], "tools_requiring_approval": ["nonexistent"]}
+    )
+
+    assert component._gated_interrupt_on() == {}
+
+
+def _capture_kwargs(captured: dict):
+    def _capture(**kwargs):
+        captured.update(kwargs)
+        return MagicMock(name="compiled_state_graph")
+
+    return _capture
+
+
+@pytest.mark.asyncio
+async def test_should_pass_durable_checkpointer_when_gated_with_run_context() -> None:
+    """LE-1447 Slice 3: a gated tool + a per-run id wires the durable saver + thread_id."""
+    from lfx.components.models_and_agents.agent_helpers.job_checkpoint_saver import JobCheckpointSaver
+
+    captured: dict = {}
+    component = _build_component()
+    component._run_id = "job-1"
+    component.set_attributes(
+        {"tools": [SimpleNamespace(name="transfer")], "tools_requiring_approval": ["transfer"]}
+    )
+    with (
+        patch.object(type(component), "_get_llm", return_value=MagicMock(name="fake_llm")),
+        patch("lfx.components.models_and_agents.agent.create_agent", side_effect=_capture_kwargs(captured)),
+    ):
+        component.create_agent_runnable()
+
+    assert isinstance(captured.get("checkpointer"), JobCheckpointSaver)
+    assert component._agent_thread_id() == "job-1"
+
+
+@pytest.mark.asyncio
+async def test_should_omit_checkpointer_when_no_tools_gated() -> None:
+    captured: dict = {}
+    component = _build_component()
+    component._run_id = "job-1"
+    component.set_attributes(
+        {"tools": [SimpleNamespace(name="transfer")], "tools_requiring_approval": []}
+    )
+    with (
+        patch.object(type(component), "_get_llm", return_value=MagicMock(name="fake_llm")),
+        patch("lfx.components.models_and_agents.agent.create_agent", side_effect=_capture_kwargs(captured)),
+    ):
+        component.create_agent_runnable()
+
+    assert captured.get("checkpointer") is None
+
+
+@pytest.mark.asyncio
+async def test_should_omit_checkpointer_and_hitl_when_allow_interrupts_false() -> None:
+    """The structured-output path disables interrupts: no checkpointer, no HITL middleware."""
+    from langchain.agents.middleware import HumanInTheLoopMiddleware
+
+    captured: dict = {}
+    component = _build_component()
+    component._run_id = "job-1"
+    component.set_attributes(
+        {"tools": [SimpleNamespace(name="transfer")], "tools_requiring_approval": ["transfer"]}
+    )
+    with (
+        patch.object(type(component), "_get_llm", return_value=MagicMock(name="fake_llm")),
+        patch("lfx.components.models_and_agents.agent.create_agent", side_effect=_capture_kwargs(captured)),
+    ):
+        component.create_agent_runnable(allow_interrupts=False)
+
+    assert captured.get("checkpointer") is None
+    assert not any(isinstance(m, HumanInTheLoopMiddleware) for m in (captured.get("middleware") or []))

src/lfx/tests/unit/components/models_and_agents/test_job_checkpoint_saver.py

@@ -0,0 +1,114 @@
+"""LE-1447 Slice 2: durable agent checkpoint saver round-trip.
+
+A scripted tool-calling model requests a gated tool (pause), the paused thread is
+serialized into a plain blob store via msgpack→base64→JSON, a FRESH saver is rebuilt
+from that blob alone, and `Command(resume=approve)` runs the gated tool to completion.
+No mocks of the serializer — the `__interrupt__` write is non-JSON and must survive.
+"""
+
+from __future__ import annotations
+
+import pytest
+from langchain.agents import create_agent
+from langchain.agents.middleware import HumanInTheLoopMiddleware
+from langchain_core.language_models import BaseChatModel
+from langchain_core.messages import AIMessage, ToolMessage
+from langchain_core.outputs import ChatGeneration, ChatResult
+from langchain_core.tools import tool
+from langgraph.types import Command
+from lfx.components.models_and_agents.agent_helpers.job_checkpoint_saver import (
+    JobCheckpointSaver,
+)
+
+
+@tool
+def transfer_money(amount: int) -> str:
+    """Transfer money (gated tool)."""
+    return f"transferred {amount}"
+
+
+class ScriptedModel(BaseChatModel):
+    """Request the gated tool until a tool result exists, then answer."""
+
+    @property
+    def _llm_type(self) -> str:
+        return "scripted"
+
+    def bind_tools(self, tools, **kwargs):  # noqa: ARG002
+        return self
+
+    def _generate(self, messages, stop=None, run_manager=None, **kwargs):  # noqa: ARG002
+        answered = any(isinstance(m, ToolMessage) for m in messages)
+        msg = (
+            AIMessage(content="Done — transferred.")
+            if answered
+            else AIMessage(
+                content="",
+                tool_calls=[{"name": "transfer_money", "args": {"amount": 200}, "id": "call_1"}],
+            )
+        )
+        return ChatResult(generations=[ChatGeneration(message=msg)])
+
+
+def _store():
+    blobs: dict[tuple[str, str], str] = {}
+
+    async def save_blob(job_id: str, kind: str, blob: str) -> None:
+        blobs[(job_id, kind)] = blob
+
+    async def load_blob(job_id: str, kind: str) -> str | None:
+        return blobs.get((job_id, kind))
+
+    return blobs, save_blob, load_blob
+
+
+def _agent(saver):
+    return create_agent(
+        model=ScriptedModel(),
+        tools=[transfer_money],
+        system_prompt="Use the tool.",
+        middleware=[HumanInTheLoopMiddleware(interrupt_on={"transfer_money": True})],
+        checkpointer=saver,
+    )
+
+
+@pytest.mark.asyncio
+async def test_durable_saver_round_trip_pauses_persists_and_resumes() -> None:
+    blobs, save_blob, load_blob = _store()
+    config = {"configurable": {"thread_id": "job-1"}}
+
+    # Run 1: pauses at the gated tool; the paused thread is persisted to the blob store.
+    saver = JobCheckpointSaver("job-1", save_blob, load_blob)
+    async for _ in _agent(saver).astream_events(
+        {"messages": [("user", "send 200")]}, version="v2", config=config
+    ):
+        pass
+
+    assert ("job-1", "agent") in blobs, "paused thread was not persisted"
+
+    # Rebuild a FRESH saver from the stored blob alone (simulates a restart).
+    saver2 = JobCheckpointSaver("job-1", save_blob, load_blob)
+    restored = await saver2.aget_tuple(config)
+    assert restored is not None
+    assert any(channel == "__interrupt__" for _task, channel, _value in restored.pending_writes)
+
+    # Resume approve on the rebuilt thread → the gated tool runs and the run completes.
+    agent2 = _agent(saver2)
+    async for _ in agent2.astream_events(
+        Command(resume={"decisions": [{"type": "approve"}]}), version="v2", config=config
+    ):
+        pass
+
+    final = await agent2.aget_state(config)
+    messages = final.values.get("messages", [])
+    assert any(isinstance(m, ToolMessage) and "transferred 200" in m.content for m in messages)
+    assert any(isinstance(m, AIMessage) and "Done" in m.content for m in messages)
+    assert not final.interrupts
+
+
+@pytest.mark.asyncio
+async def test_durable_saver_async_only() -> None:
+    _blobs, save_blob, load_blob = _store()
+    saver = JobCheckpointSaver("job-1", save_blob, load_blob)
+    with pytest.raises(NotImplementedError):
+        saver.get_tuple({"configurable": {"thread_id": "job-1"}})