human-in-the-loop pause/resume for v2 background runs

src/backend/base/langflow/api/build.py

@@ -350,6 +350,43 @@ async def create_flow_response(
     )
 
 
+async def _persist_human_input_card(data: dict, flow_id: uuid.UUID, session_id: str, job_id) -> None:
+    """Persist the pause as a chat message so the interactive card survives reload.
+
+    The card carries request_id + job_id, so a reloaded session can resume the run.
+    """
+    from lfx.memory import astore_message
+    from lfx.schema.content_block import ContentBlock
+    from lfx.schema.content_types import HumanInputContent
+    from lfx.schema.message import Message
+
+    content = HumanInputContent(
+        request_id=data.get("request_id", ""),
+        job_id=str(job_id) if job_id else None,
+        kind=data.get("kind", "node_input"),
+        prompt=data.get("prompt"),
+        options=data.get("options") or [],
+        fields=data.get("schema") or [],
+        allowed_decisions=data.get("allowed_decisions") or [],
+    )
+    block = ContentBlock(title="Human input required", contents=[content])
+    message = Message(
+        text="",
+        sender="Machine",
+        sender_name="AI",
+        session_id=session_id,
+        flow_id=flow_id,
+        content_blocks=[block],
+    )
+    try:
+        stored = await astore_message(message, flow_id=flow_id, run_id=str(job_id) if job_id else None)
+        # Record the card's message id so resume can mark it answered (see the resume route).
+        if stored and job_id is not None:
+            await get_job_service().update_job_metadata(uuid.UUID(str(job_id)), {"card_message_id": str(stored[0].id)})
+    except Exception:  # noqa: BLE001
+        await logger.awarning("Failed to persist human-input card for flow %s", flow_id, exc_info=True)
+
+
 async def generate_flow_events(
     *,
     flow_id: uuid.UUID,
@@ -420,6 +457,12 @@ async def generate_flow_events(
                 graph = await create_graph(fresh_session, flow_id_str, flow_name)
 
             graph.set_run_id(run_id)
+            if job_id is not None:
+                from lfx.services.deps import get_checkpoint_service
+
+                graph.job_id = str(job_id)  # the checkpoint is keyed by job_id
+                graph.checkpointing_enabled = True  # arm the pause seam for producers
+                graph.checkpoint_store = get_checkpoint_service()
             first_layer = sort_vertices(graph)
 
             for vertex_id in first_layer:
@@ -718,12 +761,14 @@ async def generate_flow_events(
         _build_run_id = uuid.UUID(graph.run_id) if graph.run_id else None
         if _build_run_id is not None:
             _build_job_svc = get_job_service()
-            await _build_job_svc.create_job(
-                job_id=_build_run_id,
-                flow_id=flow_id,
-                user_id=current_user.id,
-                job_type=JobType.WORKFLOW,
-            )
+            # Background path already created the job; re-creating it = UNIQUE violation.
+            if await _build_job_svc.get_job_by_job_id(_build_run_id) is None:
+                await _build_job_svc.create_job(
+                    job_id=_build_run_id,
+                    flow_id=flow_id,
+                    user_id=current_user.id,
+                    job_type=JobType.WORKFLOW,
+                )
     except Exception:  # noqa: BLE001
         await logger.awarning(
             "Failed to create workflow job for /build — memory base tracking disabled for flow %s",
@@ -774,10 +819,18 @@ async def generate_flow_events(
             event_manager.on_error(data=error_message.data)
             raise
 
-    if _build_job_svc and _build_run_id:
-        await _build_job_svc.execute_with_status(_build_run_id, _run_vertex_build)
-    else:
-        await _run_vertex_build()
+    try:
+        runner_owns_status = job_id is not None  # background path: JobRunner already wraps execute_with_status
+        if _build_job_svc and _build_run_id and not runner_owns_status:
+            await _build_job_svc.execute_with_status(_build_run_id, _run_vertex_build)
+        else:
+            await _run_vertex_build()
+    except GraphPausedException as exc:
+        # Non-terminal: persist the card to history, emit the pause event, end without on_end.
+        await _persist_human_input_card(exc.data or {}, flow_id, graph.session_id or str(flow_id), job_id)
+        event_manager.send_event(event_type="human_input_required", data=exc.data or {})
+        await event_manager.queue.put((None, None, time.time()))
+        return
 
     build_duration = sum(vertex_timedeltas)
     event_manager.on_end(data={"build_duration": build_duration})

src/backend/base/langflow/api/v2/adapters/langflow.py

@@ -34,6 +34,7 @@ _LANGFLOW_DURABLE_EVENTS: frozenset[str] = frozenset(
         "remove_message",
         "error",
         "end",
+        "human_input_required",
     }
 )
 

src/backend/base/langflow/api/v2/agui_translator.py

@@ -124,6 +124,10 @@ class AGUITranslator:
                 del self._buffered_messages[removed_id]
                 self._emitted_text_message_ids.add(removed_id)
             return [CustomEvent(name="langflow.message.removed", value={"message_id": removed_id})]
+        if event_type == "human_input_required":
+            # Non-terminal: the run suspends for human input. Must precede the end/error
+            # branches so it never closes the open message or emits RUN_FINISHED/RUN_ERROR.
+            return [CustomEvent(name="langflow.human_input_required", value=data)]
 
         # Only terminal events close an open text message. Non-terminal events
         # (build_start, end_vertex, log, ...) interleave with tokens of the same

src/backend/base/langflow/api/v2/hitl.py

@@ -0,0 +1,57 @@
+"""Human-in-the-loop persistence helpers for the v2 workflows API.
+
+The pause is stored as a chat message so the interactive card survives reload;
+on resume the same message is updated with the chosen action so a reloaded
+session renders it as resolved instead of re-offering the decision.
+"""
+
+from __future__ import annotations
+
+from uuid import UUID
+
+from sqlalchemy.orm.attributes import flag_modified
+
+from langflow.services.deps import get_job_service
+from lfx.log import logger
+
+
+def _set_submitted_action(content_blocks: list, action_id: str | None) -> bool:
+    """Stamp ``submitted_action`` on the card's content in place, preserving the rest."""
+    changed = False
+    for block in content_blocks or []:
+        contents = block.get("contents") if isinstance(block, dict) else getattr(block, "contents", None)
+        for content in contents or []:
+            ctype = content.get("type") if isinstance(content, dict) else getattr(content, "type", None)
+            if ctype != "human_input":
+                continue
+            if isinstance(content, dict):
+                content["submitted_action"] = action_id
+            else:
+                content.submitted_action = action_id
+            changed = True
+    return changed
+
+
+async def mark_card_answered(job_id: UUID, request_id: str, decision: dict) -> None:  # noqa: ARG001
+    """Record the chosen action on the persisted card message for ``job_id``.
+
+    Patches the existing card in place so the prompt/options it was stored with are
+    preserved — rebuilding from the suspend event would drop them.
+    """
+    from langflow.services.database.models.message.model import MessageTable
+    from lfx.services.deps import session_scope
+
+    job = await get_job_service().get_job_by_job_id(job_id)
+    card_message_id = (job.job_metadata or {}).get("card_message_id") if job else None
+    if not card_message_id:
+        return
+    try:
+        async with session_scope() as session:
+            message = await session.get(MessageTable, UUID(str(card_message_id)))
+            if message is None:
+                return
+            if _set_submitted_action(message.content_blocks, decision.get("action_id")):
+                flag_modified(message, "content_blocks")
+                session.add(message)
+    except Exception as _e:  # noqa: BLE001
+        await logger.awarning("Failed to mark human-input card answered for job %s: %r", job_id, _e)

src/backend/base/langflow/api/v2/workflow.py

@@ -44,6 +44,8 @@ from lfx.schema.workflow import (
     JobStatus,
     WorkflowExecutionResponse,
     WorkflowJobResponse,
+    WorkflowResumeRequest,
+    WorkflowResumeResponse,
     WorkflowRunRequest,
     WorkflowStopRequest,
     WorkflowStopResponse,
@@ -688,7 +690,11 @@ async def _stream_event_frames(
                 )
                 seq += 1
             for event in adapter.translate(event_type, event_data):
-                yield _frame(event, seq)
+                frame_bytes, frame_type = _frame(event, seq)
+                # Runner detects a pause by the langflow-side type; agui maps it to CUSTOM.
+                if event_type == "human_input_required":
+                    frame_type = "human_input_required"
+                yield (frame_bytes, frame_type)
                 seq += 1
         for event in adapter.final_events():
             yield _frame(event, seq)
@@ -1152,6 +1158,64 @@ async def stop_workflow(
         ) from exc
 
 
+@router.post(
+    "/{job_id}/resume",
+    summary="Resume Workflow",
+    description="Resume a suspended (human-in-the-loop) workflow with a decision.",
+)
+async def resume_workflow(
+    job_id: str,
+    request: WorkflowResumeRequest,
+    current_user: Annotated[UserRead, Depends(get_current_user_for_workflow)],
+) -> WorkflowResumeResponse:
+    """Resume a SUSPENDED workflow run with a human decision.
+
+    Owner-or-superuser; a non-owner non-superuser (or unknown/non-workflow job)
+    maps to 404 to avoid leaking other users' runs. A stale/duplicate request_id
+    or a non-suspended job maps to 409 (single-use enforced behind ``resume_job``).
+    """
+
+    def _not_found() -> HTTPException:
+        return HTTPException(
+            status_code=status.HTTP_404_NOT_FOUND,
+            detail={"error": "Workflow job not found", "code": "JOB_NOT_FOUND", "job_id": job_id},
+        )
+
+    try:
+        parsed_job_id = UUID(job_id)
+    except ValueError as exc:
+        raise _not_found() from exc
+
+    job = await get_job_service().get_job_by_job_id(parsed_job_id)
+    is_owner = job is not None and (job.user_id is None or job.user_id == current_user.id)
+    if job is None or job.type != JobType.WORKFLOW or not (is_owner or current_user.is_superuser):
+        raise _not_found()
+
+    service = get_background_execution_service()
+    if service._frame_source_factory is None:  # noqa: SLF001
+        service._frame_source_factory = _default_frame_source_factory  # noqa: SLF001
+    accepted = await service.resume_job(
+        parsed_job_id,
+        current_user,
+        request_id=request.request_id,
+        decision=request.decision or {},
+    )
+    if not accepted:
+        raise HTTPException(
+            status_code=status.HTTP_409_CONFLICT,
+            detail={
+                "error": "Job is not resumable",
+                "code": "NOT_RESUMABLE",
+                "message": "Job is not suspended, already resumed, or the request_id is stale.",
+                "job_id": job_id,
+            },
+        )
+    from langflow.api.v2.hitl import mark_card_answered
+
+    await mark_card_answered(parsed_job_id, request.request_id, request.decision or {})
+    return WorkflowResumeResponse(job_id=job_id, status="resuming", message="Resume accepted")
+
+
 @router.get(
     "/{job_id}/events",
     response_model=None,

src/backend/base/langflow/schema/content_block.py

@@ -3,7 +3,15 @@ from typing import Annotated
 from pydantic import BaseModel, Discriminator, Field, Tag, field_serializer, field_validator
 from typing_extensions import TypedDict
 
-from .content_types import CodeContent, ErrorContent, JSONContent, MediaContent, TextContent, ToolContent
+from .content_types import (
+    CodeContent,
+    ErrorContent,
+    HumanInputContent,
+    JSONContent,
+    MediaContent,
+    TextContent,
+    ToolContent,
+)
 
 
 def _get_type(d: dict | BaseModel) -> str | None:
@@ -19,7 +27,8 @@ ContentType = Annotated[
     | Annotated[TextContent, Tag("text")]
     | Annotated[MediaContent, Tag("media")]
     | Annotated[CodeContent, Tag("code")]
-    | Annotated[JSONContent, Tag("json")],
+    | Annotated[JSONContent, Tag("json")]
+    | Annotated[HumanInputContent, Tag("human_input")],
     Discriminator(_get_type),
 ]
 

src/backend/base/langflow/schema/content_types.py

@@ -78,6 +78,24 @@ class CodeContent(BaseContent):
     title: str | None = None
 
 
+class HumanInputContent(BaseContent):
+    """Content type for a human-in-the-loop pause persisted in the chat history.
+
+    Carries the pending decision so the interactive card survives reload: the
+    request_id and job_id let the card resume the suspended run after an F5.
+    """
+
+    type: Literal["human_input"] = Field(default="human_input")
+    request_id: str
+    job_id: str | None = None
+    kind: str = "node_input"
+    prompt: str | None = None
+    options: list[dict[str, Any]] = Field(default_factory=list)
+    fields: list[dict[str, Any]] = Field(default_factory=list)
+    allowed_decisions: list[str] = Field(default_factory=list)
+    submitted_action: str | None = None
+
+
 class ToolContent(BaseContent):
     """Content type for tool start content."""
 

src/backend/base/langflow/services/background_execution/service.py

@@ -225,8 +225,12 @@ class BackgroundExecutionService(Service):
             return
 
         async def _is_terminal() -> bool:
+            # SUSPENDED ends the tail too: a run that connected while IN_PROGRESS and
+            # then suspended has no live tail to wait on (the pause isn't published live).
             current = await job_service.get_job_by_job_id(job_id)
-            return current is not None and current.status in _TERMINAL_STATUSES
+            return current is not None and (
+                current.status in _TERMINAL_STATUSES or current.status == JobStatus.SUSPENDED
+            )
 
         async for frame in self._bus.reattach(
             str(job_id), last_seq=last_seq, read_durable=read_durable, is_done=_is_terminal
@@ -266,20 +270,21 @@ class BackgroundExecutionService(Service):
         await job_service.write_signal(job_id, SignalType.STOP)
         await self._executor.cancel(str(job_id))
 
-    async def resume_job(self, job_id: UUID, user: UserRead, *, request_id: str, decision: Any) -> bool:
+    async def resume_job(self, job_id: UUID, user: UserRead, *, request_id: str, decision: Any) -> bool:  # noqa: ARG002
         """Carry a human decision back into a SUSPENDED run and re-enqueue it.
 
         Returns True when the run was accepted for resume, False on a conflict
         (not suspended, stale request_id, or lost the single-flight flip) — the
-        route maps False to 409. Ownership is enforced by ``_validate``.
+        route maps False to 409. Ownership (owner-or-superuser) is enforced at the
+        HTTP route, so this fetches by id alone and trusts the already-validated user.
         """
-        job = await self._validate(job_id, user)
-        if job.status != JobStatus.SUSPENDED:
+        job_service = get_job_service()
+        job = await job_service.get_job_by_job_id(job_id)
+        if job is None or job.type != JobType.WORKFLOW or job.status != JobStatus.SUSPENDED:
             return False
         pending = (job.job_metadata or {}).get("pending_request_id")
         if pending is not None and request_id != pending:
             return False
-        job_service = get_job_service()
         # Win the single-flight flip BEFORE writing the RESUME signal, so exactly one
         # RESUME row exists per suspend and a loser never strands a stray decision.
         if not await job_service.claim_suspended_for_resume(job_id, owner=self._owner):

src/backend/base/langflow/services/jobs/service.py

@@ -10,6 +10,7 @@ if TYPE_CHECKING:
 from datetime import datetime, timezone
 from uuid import UUID
 
+from lfx.graph.exceptions import GraphPausedException
 from sqlalchemy.exc import IntegrityError, OperationalError
 from sqlmodel import col, func, select
 
@@ -751,7 +752,7 @@ class JobService(Service):
             await logger.ainfo(f"Executing job function for job_id={job_id}")
             result = await run_coro_func(*args, **kwargs)
 
-        except PauseRequested:
+        except (PauseRequested, GraphPausedException):
             # A producer paused the run for human input. The runner suspends it
             # (SUSPENDED); do NOT write a terminal status here.
             await logger.adebug(f"Job {job_id} paused for human input")

src/backend/tests/unit/api/v2/adapters/test_human_input_event.py

@@ -0,0 +1,66 @@
+"""HITL human_input_required translation on both stream protocols (LE-1450).
+
+The pause is a non-terminal durable milestone: langflow passes it through and
+marks it durable; agui rides the existing CUSTOM mechanism as a single
+``langflow.human_input_required`` event that never closes the message or ends the run.
+"""
+
+from __future__ import annotations
+
+import json
+
+from langflow.api.v2.adapters import StreamAdapterContext, get_stream_adapter
+
+_PAYLOAD = {
+    "request_id": "node:job-1",
+    "kind": "node_input",
+    "prompt": "Approve refund?",
+    "options": [{"action_id": "approve", "label": "Approve"}, {"action_id": "reject", "label": "Reject"}],
+    "schema": [],
+    "allowed_decisions": ["approve", "reject"],
+}
+
+
+def _ctx() -> StreamAdapterContext:
+    return StreamAdapterContext(run_id="run-1", thread_id="thread-1")
+
+
+class TestLangflowHumanInputEvent:
+    def test_translate_passes_through_one_frame(self):
+        adapter = get_stream_adapter("langflow", _ctx())
+        events = list(adapter.translate("human_input_required", _PAYLOAD))
+        assert len(events) == 1
+        payload = json.loads(events[0].data_json)
+        assert payload == {"event": "human_input_required", "data": _PAYLOAD}
+
+    def test_is_durable_true(self):
+        adapter = get_stream_adapter("langflow", _ctx())
+        assert adapter.is_durable("human_input_required") is True
+
+    def test_not_the_terminal_error_type(self):
+        adapter = get_stream_adapter("langflow", _ctx())
+        assert adapter.terminal_error_type != "human_input_required"
+
+
+class TestAguiHumanInputEvent:
+    def test_translate_emits_single_custom_event(self):
+        adapter = get_stream_adapter("agui", _ctx())
+        events = list(adapter.translate("human_input_required", _PAYLOAD))
+        assert len(events) == 1
+        assert events[0].type == "CUSTOM"
+        payload = json.loads(events[0].data_json)
+        assert payload["name"] == "langflow.human_input_required"
+        assert payload["value"] == _PAYLOAD
+
+    def test_is_durable_via_custom(self):
+        adapter = get_stream_adapter("agui", _ctx())
+        assert adapter.is_durable("CUSTOM") is True
+
+    def test_pause_does_not_end_the_run_or_close_the_message(self):
+        adapter = get_stream_adapter("agui", _ctx())
+        list(adapter.translate("token", {"id": "m1", "chunk": "Working"}))  # open a text message
+        events = list(adapter.translate("human_input_required", _PAYLOAD))
+        types = [e.type for e in events]
+        assert "RUN_FINISHED" not in types
+        assert "RUN_ERROR" not in types
+        assert "TEXT_MESSAGE_END" not in types  # the open message stays open across the pause

src/backend/tests/unit/api/v2/test_resume_route.py

@@ -0,0 +1,95 @@
+"""HTTP resume route (LE-1450): POST /api/v2/workflows/{job_id}/resume.
+
+Route-level tests against a hand-written SUSPENDED job row (the full HTTP
+round-trip gates on a real pausing flow): owner-or-superuser auth with
+deny-to-404, single-use 409, and accepted-resume 200.
+"""
+
+from __future__ import annotations
+
+from uuid import uuid4
+
+import pytest
+from langflow.services.database.models.flow.model import Flow
+from langflow.services.database.models.jobs.model import Job, JobStatus, JobType
+from lfx.services.deps import session_scope
+
+pytestmark = pytest.mark.usefixtures("client")
+
+
+def _headers(api_key) -> dict:
+    return {"x-api-key": api_key.api_key}
+
+
+@pytest.fixture
+async def suspended_job(created_api_key):
+    """A hand-written SUSPENDED workflow job owned by the api-key user, with a real flow."""
+    user_id = created_api_key.user_id
+    flow_id, job_id = uuid4(), uuid4()
+    request = {"flow_id": str(flow_id), "mode": "background", "stream_protocol": "langflow", "input_value": "hi"}
+    async with session_scope() as session:
+        session.add(Flow(id=flow_id, name=f"f-{flow_id}", data={"nodes": [], "edges": []}, user_id=user_id))
+        session.add(
+            Job(
+                job_id=job_id,
+                flow_id=flow_id,
+                user_id=user_id,
+                type=JobType.WORKFLOW,
+                status=JobStatus.SUSPENDED,
+                job_metadata={"pending_request_id": "req-1", "request": request},
+            )
+        )
+        await session.flush()
+    yield job_id
+    async with session_scope() as session:
+        job = await session.get(Job, job_id)
+        if job:
+            await session.delete(job)
+        flow = await session.get(Flow, flow_id)
+        if flow:
+            await session.delete(flow)
+
+
+async def test_resume_owner_accepts_200(client, created_api_key, suspended_job):
+    body = {"request_id": "req-1", "decision": {"action_id": "approve"}}
+    resp = await client.post(f"api/v2/workflows/{suspended_job}/resume", json=body, headers=_headers(created_api_key))
+    assert resp.status_code == 200
+    assert resp.json()["status"] == "resuming"
+
+
+async def test_resume_stale_request_id_409(client, created_api_key, suspended_job):
+    body = {"request_id": "WRONG", "decision": {"action_id": "approve"}}
+    resp = await client.post(f"api/v2/workflows/{suspended_job}/resume", json=body, headers=_headers(created_api_key))
+    assert resp.status_code == 409
+
+
+async def test_resume_unknown_job_404(client, created_api_key):
+    body = {"request_id": "req-1", "decision": {}}
+    resp = await client.post(f"api/v2/workflows/{uuid4()}/resume", json=body, headers=_headers(created_api_key))
+    assert resp.status_code == 404
+
+
+async def test_resume_non_suspended_job_409(client, created_api_key):
+    """A QUEUED (not suspended) job is not resumable → 409."""
+    user_id = created_api_key.user_id
+    flow_id, job_id = uuid4(), uuid4()
+    async with session_scope() as session:
+        session.add(Flow(id=flow_id, name=f"f-{flow_id}", data={"nodes": [], "edges": []}, user_id=user_id))
+        session.add(Job(job_id=job_id, flow_id=flow_id, user_id=user_id, type=JobType.WORKFLOW, status=JobStatus.QUEUED))
+        await session.flush()
+    try:
+        body = {"request_id": "req-1", "decision": {}}
+        resp = await client.post(f"api/v2/workflows/{job_id}/resume", json=body, headers=_headers(created_api_key))
+        assert resp.status_code == 409
+    finally:
+        async with session_scope() as session:
+            for model, key in ((Job, job_id), (Flow, flow_id)):
+                row = await session.get(model, key)
+                if row:
+                    await session.delete(row)
+
+
+async def test_resume_invalid_job_id_404(client, created_api_key):
+    body = {"request_id": "req-1", "decision": {}}
+    resp = await client.post("api/v2/workflows/not-a-uuid/resume", json=body, headers=_headers(created_api_key))
+    assert resp.status_code == 404

src/frontend/src/components/core/chatComponents/ContentDisplay.tsx

@@ -2,20 +2,28 @@ import type { ReactNode } from "react";
 import Markdown from "react-markdown";
 import rehypeMathjax from "rehype-mathjax/browser";
 import remarkGfm from "remark-gfm";
-import type { ContentType, JSONValue } from "@/types/chat";
+import type { ContentType, InteractiveContent, JSONValue } from "@/types/chat";
 import { extractLanguage, isCodeBlock } from "@/utils/codeBlockUtils";
 import ForwardedIconComponent from "../../common/genericIconComponent";
 import SimplifiedCodeTabComponent from "../codeTabsComponent";
 import DurationDisplay from "./DurationDisplay";
+import HumanInputCard, { type HumanInputDecision } from "./HumanInputCard";
 
 export default function ContentDisplay({
   content,
   chatId,
   playgroundPage,
+  humanInputSubmitted,
+  onHumanInputSubmit,
 }: {
   content: ContentType;
   chatId: string;
   playgroundPage?: boolean;
+  humanInputSubmitted?: boolean;
+  onHumanInputSubmit?: (
+    content: InteractiveContent,
+    decision: HumanInputDecision,
+  ) => void;
 }) {
   const renderDuration = content.duration !== undefined && !playgroundPage && (
     <div className="absolute right-2 top-4">
@@ -242,6 +250,19 @@ export default function ContentDisplay({
         </div>
       );
       break;
+    case "human_input":
+      contentData = (
+        <HumanInputCard
+          content={content}
+          submitted={humanInputSubmitted}
+          onSubmit={
+            onHumanInputSubmit
+              ? (decision) => onHumanInputSubmit(content, decision)
+              : undefined
+          }
+        />
+      );
+      break;
   }
 
   return (

src/frontend/src/components/core/chatComponents/HumanInputCard.tsx

@@ -0,0 +1,161 @@
+import { useState } from "react";
+import Markdown from "react-markdown";
+import remarkGfm from "remark-gfm";
+import { queryClient } from "@/contexts";
+import {
+  consumeBackgroundEvents,
+  getResumeContext,
+  markHumanInputSubmitted,
+} from "@/controllers/API/agui/run-flow-bridge";
+import { useResumeWorkflow } from "@/controllers/API/queries/workflows/use-resume-workflow";
+import { Button } from "@/components/ui/button";
+import { Input } from "@/components/ui/input";
+import useAlertStore from "@/stores/alertStore";
+import useFlowStore from "@/stores/flowStore";
+import type { InteractiveContent } from "@/types/chat";
+import ForwardedIconComponent from "../../common/genericIconComponent";
+
+export interface HumanInputDecision {
+  action_id: string;
+  values: Record<string, string>;
+}
+
+/**
+ * Interactive human-in-the-loop card: a prompt, optional editable form fields, and
+ * one button per decision. Mirrors the convergent HITL pattern (a paused turn the
+ * user resolves inline) using Langflow's own components. Submit is single-use in the
+ * UI (best-effort; the server is authoritative and returns 409 on a stale resume).
+ */
+export default function HumanInputCard({
+  content,
+  onSubmit,
+  submitted = false,
+}: {
+  content: InteractiveContent;
+  onSubmit?: (decision: HumanInputDecision) => void;
+  submitted?: boolean;
+}) {
+  const [values, setValues] = useState<Record<string, string>>({});
+  const [localChosen, setLocalChosen] = useState<string | null>(null);
+  const fields = content.schema ?? content.fields ?? [];
+  const { mutate: resume, isPending } = useResumeWorkflow();
+  const setErrorData = useAlertStore((s) => s.setErrorData);
+  // Derive from the persisted/cached content first so the choice survives re-renders
+  // (the resume reattach replays the stream and re-renders the message list).
+  const chosen = content.submitted_action ?? localChosen;
+  const isSubmitted = submitted || chosen !== null;
+
+  // Resume the suspended run and reattach to its durable event stream so the
+  // continued run renders into the same chat session.
+  const resumeRun = (decision: HumanInputDecision) => {
+    const jobId = content.job_id;
+    if (!jobId) {
+      setErrorData({ title: "Cannot resume: missing job id" });
+      setLocalChosen(null);
+      return;
+    }
+    resume(
+      { jobId, requestId: content.request_id, decision },
+      {
+        onSuccess: () => {
+          const flowStore = useFlowStore.getState();
+          flowStore.setAwaitingInput(false);
+          const reattach = getResumeContext(content.request_id);
+          if (reattach) {
+            flowStore.setIsBuilding(true);
+            void consumeBackgroundEvents(reattach.jobId, reattach.opts);
+          } else {
+            // After a reload there is no live stream to reattach to; refetch the
+            // history so the resumed run's persisted output shows up.
+            void queryClient.invalidateQueries({
+              queryKey: ["useGetMessagesQuery"],
+            });
+          }
+        },
+        onError: (err: Error) => {
+          // 409 means the run was already resumed (single-use) — keep the card
+          // locked on the choice; only a genuine failure re-opens the buttons.
+          const status = (err as { response?: { status?: number } })?.response
+            ?.status;
+          if (status === 409) return;
+          setLocalChosen(null);
+          setErrorData({ title: "Resume failed", list: [err.message] });
+        },
+      },
+    );
+  };
+
+  const handleDecision = (actionId: string) => {
+    if (isSubmitted || isPending) return;
+    setLocalChosen(actionId);
+    markHumanInputSubmitted(content.request_id, actionId);
+    const decision = { action_id: actionId, values };
+    if (onSubmit) onSubmit(decision);
+    else resumeRun(decision);
+  };
+
+  return (
+    <div
+      data-testid="human-input-card"
+      className="flex flex-col gap-3 rounded-md border border-border bg-muted/40 p-4"
+    >
+      <div className="flex items-center gap-2 text-sm font-medium text-foreground">
+        <ForwardedIconComponent name="CircleHelp" className="h-4 w-4" />
+        <span>Human input required</span>
+      </div>
+
+      {content.prompt && (
+        <div className="prose prose-sm dark:prose-invert max-w-none text-sm">
+          <Markdown remarkPlugins={[remarkGfm]}>{content.prompt}</Markdown>
+        </div>
+      )}
+
+      {fields.length > 0 && (
+        <div className="flex flex-col gap-2">
+          {fields.map((field) => (
+            <label key={field.name} className="flex flex-col gap-1 text-xs">
+              <span className="text-muted-foreground">
+                {field.name}
+                {field.required ? " *" : ""}
+              </span>
+              <Input
+                data-testid={`human-input-field-${field.name}`}
+                disabled={isSubmitted}
+                value={values[field.name] ?? ""}
+                onChange={(e) =>
+                  setValues((prev) => ({
+                    ...prev,
+                    [field.name]: e.target.value,
+                  }))
+                }
+              />
+            </label>
+          ))}
+        </div>
+      )}
+
+      <div className="flex flex-wrap gap-2">
+        {content.options
+          .filter((option) => chosen === null || option.action_id === chosen)
+          .map((option) => (
+            <Button
+              key={option.action_id}
+              data-testid={`human-input-decision-${option.action_id}`}
+              size="sm"
+              variant={chosen === option.action_id ? "default" : "outline"}
+              disabled={isSubmitted || isPending}
+              onClick={() => handleDecision(option.action_id)}
+            >
+              {chosen === option.action_id && (
+                <ForwardedIconComponent
+                  name="Check"
+                  className="mr-1 h-3.5 w-3.5"
+                />
+              )}
+              {option.label || option.action_id}
+            </Button>
+          ))}
+      </div>
+    </div>
+  );
+}

src/frontend/src/components/core/chatComponents/__tests__/HumanInputCard.test.tsx

@@ -0,0 +1,155 @@
+import { fireEvent, render, screen } from "@testing-library/react";
+import type { InteractiveContent } from "@/types/chat";
+import HumanInputCard from "../HumanInputCard";
+
+const mockResume = jest.fn();
+const mockConsume = jest.fn();
+const mockSetAwaitingInput = jest.fn();
+const mockSetIsBuilding = jest.fn();
+const mockSetErrorData = jest.fn();
+
+jest.mock("@/contexts", () => ({
+  queryClient: { invalidateQueries: jest.fn() },
+}));
+jest.mock("@/controllers/API/queries/workflows/use-resume-workflow", () => ({
+  useResumeWorkflow: () => ({ mutate: mockResume, isPending: false }),
+}));
+jest.mock("@/controllers/API/agui/run-flow-bridge", () => ({
+  consumeBackgroundEvents: (...args: unknown[]) => mockConsume(...args),
+  getResumeContext: () => ({ jobId: "job-1", opts: { flowId: "f1" } }),
+  markHumanInputSubmitted: jest.fn(),
+}));
+jest.mock("@/stores/flowStore", () => ({
+  __esModule: true,
+  default: {
+    getState: () => ({
+      setAwaitingInput: mockSetAwaitingInput,
+      setIsBuilding: mockSetIsBuilding,
+    }),
+  },
+}));
+jest.mock("@/stores/alertStore", () => ({
+  __esModule: true,
+  default: (selector: (s: { setErrorData: unknown }) => unknown) =>
+    selector({ setErrorData: mockSetErrorData }),
+}));
+
+jest.mock("../../../common/genericIconComponent", () => ({
+  __esModule: true,
+  default: ({ name }: { name: string }) => <span data-testid={`icon-${name}`} />,
+}));
+
+jest.mock("react-markdown", () => ({
+  __esModule: true,
+  default: ({ children }: { children: string }) => <span>{children}</span>,
+}));
+jest.mock("remark-gfm", () => ({ __esModule: true, default: () => {} }));
+
+jest.mock("@/components/ui/button", () => ({
+  Button: ({ children, ...props }: any) => <button {...props}>{children}</button>,
+}));
+jest.mock("@/components/ui/input", () => ({
+  Input: (props: any) => <input {...props} />,
+}));
+
+const _approval: InteractiveContent = {
+  type: "human_input",
+  kind: "tool_approval",
+  request_id: "node:job-1",
+  prompt: "Approve refund?",
+  options: [
+    { action_id: "approve", label: "Approve" },
+    { action_id: "reject", label: "Reject" },
+  ],
+  allowed_decisions: ["approve", "reject"],
+};
+
+describe("HumanInputCard", () => {
+  beforeEach(() => jest.clearAllMocks());
+
+  it("renders the prompt and one button per option", () => {
+    render(<HumanInputCard content={_approval} onSubmit={jest.fn()} />);
+    expect(screen.getByText("Approve refund?")).toBeInTheDocument();
+    expect(screen.getByTestId("human-input-decision-approve")).toBeInTheDocument();
+    expect(screen.getByTestId("human-input-decision-reject")).toBeInTheDocument();
+  });
+
+  it("submits the chosen action with empty values when there are no form fields", () => {
+    const onSubmit = jest.fn();
+    render(<HumanInputCard content={_approval} onSubmit={onSubmit} />);
+    fireEvent.click(screen.getByTestId("human-input-decision-approve"));
+    expect(onSubmit).toHaveBeenCalledWith({ action_id: "approve", values: {} });
+  });
+
+  it("disables the controls once submitted", () => {
+    const onSubmit = jest.fn();
+    render(<HumanInputCard content={_approval} onSubmit={onSubmit} submitted />);
+    fireEvent.click(screen.getByTestId("human-input-decision-approve"));
+    expect(onSubmit).not.toHaveBeenCalled();
+    expect(screen.getByTestId("human-input-decision-approve")).toBeDisabled();
+  });
+
+  it("collects form field values into the decision (node_input)", () => {
+    const onSubmit = jest.fn();
+    const nodeInput: InteractiveContent = {
+      type: "human_input",
+      kind: "node_input",
+      request_id: "node:job-2",
+      prompt: "Pick one",
+      options: [
+        { action_id: "a", label: "A" },
+        { action_id: "b", label: "B" },
+        { action_id: "c", label: "C" },
+      ],
+      schema: [{ name: "reason", type: "str", required: true }],
+      allowed_decisions: ["a", "b", "c"],
+    };
+    render(<HumanInputCard content={nodeInput} onSubmit={onSubmit} />);
+    expect(screen.getAllByRole("button")).toHaveLength(3);
+    fireEvent.change(screen.getByTestId("human-input-field-reason"), {
+      target: { value: "fraud" },
+    });
+    fireEvent.click(screen.getByTestId("human-input-decision-b"));
+    expect(onSubmit).toHaveBeenCalledWith({
+      action_id: "b",
+      values: { reason: "fraud" },
+    });
+  });
+
+  it("resumes the run itself when no onSubmit is provided", () => {
+    const content: InteractiveContent = { ..._approval, job_id: "job-1" };
+    render(<HumanInputCard content={content} />);
+    fireEvent.click(screen.getByTestId("human-input-decision-approve"));
+    expect(mockResume).toHaveBeenCalledWith(
+      {
+        jobId: "job-1",
+        requestId: "node:job-1",
+        decision: { action_id: "approve", values: {} },
+      },
+      expect.objectContaining({ onSuccess: expect.any(Function) }),
+    );
+    const { onSuccess } = mockResume.mock.calls[0][1];
+    onSuccess();
+    expect(mockSetAwaitingInput).toHaveBeenCalledWith(false);
+    expect(mockSetIsBuilding).toHaveBeenCalledWith(true);
+    expect(mockConsume).toHaveBeenCalledWith("job-1", { flowId: "f1" });
+  });
+
+  it("disables controls after a self-resume click (single-use)", () => {
+    const content: InteractiveContent = { ..._approval, job_id: "job-1" };
+    render(<HumanInputCard content={content} />);
+    fireEvent.click(screen.getByTestId("human-input-decision-approve"));
+    expect(screen.getByTestId("human-input-decision-approve")).toBeDisabled();
+  });
+
+  it("keeps only the chosen option and removes the others after selecting", () => {
+    const content: InteractiveContent = { ..._approval, job_id: "job-1" };
+    render(<HumanInputCard content={content} />);
+    expect(screen.getByTestId("human-input-decision-reject")).toBeInTheDocument();
+    fireEvent.click(screen.getByTestId("human-input-decision-approve"));
+    expect(screen.getByTestId("human-input-decision-approve")).toBeInTheDocument();
+    expect(
+      screen.queryByTestId("human-input-decision-reject"),
+    ).not.toBeInTheDocument();
+  });
+});

src/frontend/src/components/core/playgroundComponent/chat-view/chat-messages/components/bot-message.tsx

@@ -6,7 +6,9 @@ import IconComponent, {
 } from "@/components/common/genericIconComponent";
 import MessageMetadata from "@/components/common/messageMetadataComponent";
 import { ContentBlockDisplay } from "@/components/core/chatComponents/ContentBlockDisplay";
+import HumanInputCard from "@/components/core/chatComponents/HumanInputCard";
 import { useUpdateMessage } from "@/controllers/API/queries/messages";
+import type { InteractiveContent } from "@/types/chat";
 import { CustomMarkdownField } from "@/customization/components/custom-markdown-field";
 import useAlertStore from "@/stores/alertStore";
 import useFlowStore from "@/stores/flowStore";
@@ -44,6 +46,12 @@ export const BotMessage = memo(
 
     const isEmpty = decodedMessage?.trim() === "";
     const chatMessage = chat.message ? chat.message.toString() : "";
+    // ContentBlockDisplay renders only tool_use content, so the HITL card is rendered directly below.
+    const humanInputContent = chat.content_blocks
+      ?.flatMap((block) => block.contents ?? [])
+      .find((content) => content?.type === "human_input") as
+      | InteractiveContent
+      | undefined;
     const { mutate: updateMessageMutation } = useUpdateMessage();
 
     const handleEditMessage = (message: string) => {
@@ -138,7 +146,10 @@ export const BotMessage = memo(
             )}
           >
             <div className="flex w-full items-start gap-3">
-              {(thinkingActive || displayTime > 0 || chatMessage !== "") && (
+              {(thinkingActive ||
+                displayTime > 0 ||
+                chatMessage !== "" ||
+                humanInputContent) && (
                 <div
                   className="relative hidden h-6 w-6 mt-[-1px] flex-shrink-0 items-center justify-center overflow-hidden rounded bg-white text-2xl @[45rem]/chat-panel:!flex border-0"
                   style={
@@ -184,6 +195,12 @@ export const BotMessage = memo(
                   </span>
                 </div>
 
+                {humanInputContent && (
+                  <div className="mt-2">
+                    <HumanInputCard content={humanInputContent} />
+                  </div>
+                )}
+
                 {((chat.content_blocks && chat.content_blocks.length > 0) ||
                   (isBuilding && lastMessage)) && (
                   <ContentBlockDisplay
@@ -211,9 +228,10 @@ export const BotMessage = memo(
                           data-testid={`chat-message-${chat.sender_name}-${chatMessage}`}
                           className="flex w-full flex-col"
                         >
-                          {(chatMessage === "" || (isEmpty && !isStreaming)) &&
-                          isBuilding &&
-                          lastMessage ? (
+                          {humanInputContent ? null : (chatMessage === "" ||
+                              (isEmpty && !isStreaming)) &&
+                              isBuilding &&
+                              lastMessage ? (
                             <IconComponent
                               name="MoreHorizontal"
                               className="h-8 w-8 animate-pulse"

src/frontend/src/controllers/API/agui/__tests__/consume-background-events.test.ts

@@ -0,0 +1,169 @@
+/**
+ * Drives the fetch-based durable-event consumer end to end with a mocked SSE
+ * stream. The pause case is the regression that mattered: a stream that closes
+ * after ``langflow.human_input_required`` (no terminal RUN_FINISHED) must still
+ * inject the card and clear the building spinner.
+ */
+
+import { ReadableStream as NodeReadableStream } from "stream/web";
+import { TextDecoder as NodeTextDecoder, TextEncoder as NodeTextEncoder } from "util";
+
+Object.assign(globalThis, {
+  TextEncoder: globalThis.TextEncoder ?? NodeTextEncoder,
+  TextDecoder: globalThis.TextDecoder ?? NodeTextDecoder,
+  ReadableStream: globalThis.ReadableStream ?? NodeReadableStream,
+});
+
+const addMessage = jest.fn();
+const setIsBuilding = jest.fn();
+const setAwaitingInput = jest.fn();
+const setBuildInfo = jest.fn();
+const updateEdgesRunningByNodes = jest.fn();
+const revertBuiltStatusFromBuilding = jest.fn();
+const setErrorData = jest.fn();
+let storeMessages: { id: string }[] = [];
+
+jest.mock("@/stores/flowStore", () => ({
+  __esModule: true,
+  default: {
+    getState: () => ({
+      setIsBuilding,
+      setAwaitingInput,
+      setBuildInfo,
+      updateEdgesRunningByNodes,
+      revertBuiltStatusFromBuilding,
+    }),
+  },
+}));
+jest.mock("@/stores/messagesStore", () => ({
+  useMessagesStore: {
+    getState: () => ({
+      addMessage,
+      messages: storeMessages,
+    }),
+  },
+}));
+jest.mock("@/stores/alertStore", () => ({
+  __esModule: true,
+  default: { getState: () => ({ setErrorData }) },
+}));
+jest.mock(
+  "@/components/core/playgroundComponent/chat-view/utils/message-event-handler",
+  () => ({ handleMessageEvent: jest.fn() }),
+);
+const updateMessageMock = jest.fn();
+jest.mock(
+  "@/components/core/playgroundComponent/chat-view/utils/message-utils",
+  () => ({ updateMessage: (...args: unknown[]) => updateMessageMock(...args) }),
+);
+
+import { consumeBackgroundEvents } from "@/controllers/API/agui/run-flow-bridge";
+
+function sseStream(frames: string[]): ReadableStream<Uint8Array> {
+  const encoder = new TextEncoder();
+  return new ReadableStream({
+    start(controller) {
+      for (const frame of frames) {
+        controller.enqueue(encoder.encode(`data: ${frame}\n\n`));
+      }
+      controller.close();
+    },
+  });
+}
+
+/** Emits real `data:\nid:` frames and closes WITHOUT a trailing blank line. */
+function sseStreamNoTrailingBlank(frames: string[]): ReadableStream<Uint8Array> {
+  const encoder = new TextEncoder();
+  const body =
+    frames.map((f, i) => `data: ${f}\nid: ${i + 1}`).join("\n\n") + "\n";
+  return new ReadableStream({
+    start(controller) {
+      controller.enqueue(encoder.encode(body));
+      controller.close();
+    },
+  });
+}
+
+describe("consumeBackgroundEvents", () => {
+  beforeEach(() => {
+    jest.clearAllMocks();
+    storeMessages = [];
+  });
+
+  it("injects the card and parks awaiting-input when the run suspends", async () => {
+    const humanInput = JSON.stringify({
+      type: "CUSTOM",
+      name: "langflow.human_input_required",
+      value: {
+        request_id: "HI:job-1",
+        kind: "node_input",
+        prompt: "Approve?",
+        options: [{ action_id: "approve", label: "Approve" }],
+        schema: [],
+        allowed_decisions: ["approve"],
+      },
+    });
+    global.fetch = jest.fn().mockResolvedValue({
+      ok: true,
+      body: sseStream([
+        JSON.stringify({ type: "RUN_STARTED", runId: "job-1" }),
+        humanInput,
+      ]),
+    }) as unknown as typeof fetch;
+
+    await consumeBackgroundEvents("job-1", { flowId: "f1", threadId: "s1" });
+
+    expect(addMessage).toHaveBeenCalledTimes(1);
+    expect(addMessage.mock.calls[0][0].id).toBe("human-input-HI:job-1");
+    // The new playground reads the react-query cache, not useMessagesStore.
+    expect(updateMessageMock).toHaveBeenCalledTimes(1);
+    expect(updateMessageMock.mock.calls[0][0].id).toBe("human-input-HI:job-1");
+    expect(setIsBuilding).toHaveBeenLastCalledWith(false);
+    expect(setAwaitingInput).toHaveBeenLastCalledWith(true);
+  });
+
+  it("processes the final pause frame even when the stream closes without a trailing blank line", async () => {
+    const humanInput = JSON.stringify({
+      type: "CUSTOM",
+      name: "langflow.human_input_required",
+      value: {
+        request_id: "HI:job-2",
+        kind: "node_input",
+        prompt: "Approve?",
+        options: [{ action_id: "approve", label: "Approve" }],
+        schema: [],
+        allowed_decisions: ["approve"],
+      },
+    });
+    global.fetch = jest.fn().mockResolvedValue({
+      ok: true,
+      body: sseStreamNoTrailingBlank([
+        JSON.stringify({ type: "RUN_STARTED", runId: "job-2" }),
+        JSON.stringify({ type: "STATE_DELTA", delta: [] }),
+        humanInput,
+      ]),
+    }) as unknown as typeof fetch;
+
+    await consumeBackgroundEvents("job-2", { flowId: "f1", threadId: "s1" });
+
+    expect(addMessage).toHaveBeenCalledTimes(1);
+    expect(addMessage.mock.calls[0][0].id).toBe("human-input-HI:job-2");
+    expect(setAwaitingInput).toHaveBeenLastCalledWith(true);
+  });
+
+  it("clears awaiting-input when a resumed run reaches RUN_FINISHED", async () => {
+    global.fetch = jest.fn().mockResolvedValue({
+      ok: true,
+      body: sseStream([
+        JSON.stringify({ type: "RUN_STARTED", runId: "job-1" }),
+        JSON.stringify({ type: "RUN_FINISHED" }),
+      ]),
+    }) as unknown as typeof fetch;
+
+    await consumeBackgroundEvents("job-1", { flowId: "f1", threadId: "s1" });
+
+    expect(setIsBuilding).toHaveBeenLastCalledWith(false);
+    expect(setAwaitingInput).toHaveBeenLastCalledWith(false);
+    expect(setBuildInfo).toHaveBeenCalledWith({ success: true });
+  });
+});

src/frontend/src/controllers/API/agui/__tests__/run-flow-bridge.test.ts

@@ -20,6 +20,10 @@ function makeRecordingContext() {
     setRunId: (runId) => calls.push(`setRunId:${runId}`),
     applyDelta: (ops) => calls.push(`applyDelta:${ops.length}`),
     handleCustomEvent: (eventType) => calls.push(`custom:${eventType}`),
+    onHumanInput: (payload) =>
+      calls.push(
+        `humanInput:${(payload as { request_id?: string })?.request_id ?? ""}`,
+      ),
     onFinished: () => calls.push("finished"),
     onError: (message) => calls.push(`error:${message}`),
   };
@@ -139,3 +143,21 @@ describe("handleAGUIEvent non-terminal contract", () => {
     expect(calls).toEqual([]);
   });
 });
+
+describe("handleAGUIEvent human-input contract", () => {
+  it("surfaces a human_input_required CUSTOM event non-terminally", () => {
+    const { ctx, calls } = makeRecordingContext();
+
+    const terminal = handleAGUIEvent(
+      {
+        type: EventType.CUSTOM,
+        name: "langflow.human_input_required",
+        value: { request_id: "node:job-1", kind: "node_input" },
+      } as unknown as BaseEvent,
+      ctx,
+    );
+
+    expect(terminal).toBe(false); // pause is NOT terminal
+    expect(calls).toEqual(["humanInput:node:job-1"]);
+  });
+});

src/frontend/src/controllers/API/agui/run-agent.ts

@@ -189,3 +189,36 @@ export function createWorkflowAgent(
   };
   return agent;
 }
+
+/**
+ * Agent that reattaches to a background run's durable event stream via
+ * `GET /api/v2/workflows/{job_id}/events` (LE-1442/1450). HITL runs are submitted
+ * in background mode (a separate JSON POST) and consumed here; `lastEventId` is the
+ * SSE `Last-Event-ID` so a resume continues gap-free past the suspend boundary.
+ */
+export function createEventsAgent(opts: {
+  jobId: string;
+  lastEventId?: string;
+  headers?: Record<string, string>;
+}): HttpAgent {
+  const config: HttpAgentConfig = {
+    url: `${WORKFLOWS_ENDPOINT}/${encodeURIComponent(opts.jobId)}/events`,
+    headers: opts.headers,
+  };
+  const agent = new HttpAgent(config);
+  (
+    agent as unknown as { requestInit: (input: RunAgentInput) => RequestInit }
+  ).requestInit = function requestInit(_input: RunAgentInput): RequestInit {
+    const headers: Record<string, string> = {
+      ...agent.headers,
+      Accept: "text/event-stream",
+    };
+    if (opts.lastEventId) headers["Last-Event-ID"] = opts.lastEventId;
+    return {
+      method: "GET",
+      headers,
+      signal: agent.abortController.signal,
+    };
+  };
+  return agent;
+}

src/frontend/src/controllers/API/agui/run-flow-bridge.ts

@@ -10,18 +10,25 @@
 
 import { type BaseEvent, EventType } from "@ag-ui/client";
 import { handleMessageEvent } from "@/components/core/playgroundComponent/chat-view/utils/message-event-handler";
+import { updateMessage } from "@/components/core/playgroundComponent/chat-view/utils/message-utils";
+import { queryClient } from "@/contexts";
 import { BuildStatus } from "@/constants/enums";
 import useAlertStore from "@/stores/alertStore";
 import useFlowStore from "@/stores/flowStore";
+import { useMessagesStore } from "@/stores/messagesStore";
 import type {
   ChatInputType,
   ChatOutputType,
   VertexBuildTypeAPI,
   VertexDataTypeAPI,
 } from "@/types/api";
+import type { ContentBlock, InteractiveContent } from "@/types/chat";
+import type { Message } from "@/types/messages";
+import { api } from "../api";
 import {
   buildWorkflowRunRequest,
   createWorkflowAgent,
+  WORKFLOWS_ENDPOINT,
   WORKFLOWS_PUBLIC_ENDPOINT,
   type WorkflowRunOptions,
 } from "./run-agent";
@@ -53,6 +60,7 @@ export interface BridgeContext {
   setRunId: (runId: string) => void;
   applyDelta: (ops: JsonPatchOp[]) => void;
   handleCustomEvent: (eventType: string, data: unknown) => void;
+  onHumanInput: (payload: unknown) => void;
   onFinished: () => void;
   onError: (message: string) => void;
 }
@@ -83,6 +91,9 @@ export function handleAGUIEvent(event: BaseEvent, ctx: BridgeContext): boolean {
     };
     if (custom.name === "langflow.event" && custom.value?.event_type) {
       ctx.handleCustomEvent(custom.value.event_type, custom.value.data);
+    } else if (custom.name === "langflow.human_input_required") {
+      // Non-terminal: surface the card; the SSE ends at suspend and reattaches on resume.
+      ctx.onHumanInput(custom.value);
     }
     return false;
   }
@@ -239,6 +250,11 @@ export async function runFlowAGUI(
     },
     applyDelta: (ops) => applyStateDelta(ops, runId, touchedNodeIds),
     handleCustomEvent: (eventType, data) => handleMessageEvent(eventType, data),
+    onHumanInput: () => {
+      // Stream mode can't durably suspend; HITL runs go through the background
+      // path (runFlowHITL). Mark awaiting-input so the UI reflects the pause.
+      flowStore.setAwaitingInput(true);
+    },
     onFinished: () => {
       terminalEventSeen = true;
       flowStore.setBuildInfo({ success: true });
@@ -321,3 +337,289 @@ export async function runFlowAGUI(
     });
   });
 }
+
+/** Background-run body (mode="background"); HITL needs the durable substrate. */
+function buildBackgroundRunRequest(opts: WorkflowRunOptions) {
+  const body: Record<string, unknown> = {
+    flow_id: opts.flowId,
+    input_value: opts.message ?? "",
+    mode: "background",
+    stream_protocol: "agui",
+  };
+  if (opts.threadId) body.session_id = opts.threadId;
+  if (opts.tweaks) body.tweaks = opts.tweaks;
+  if (opts.startComponentId) body.start_component_id = opts.startComponentId;
+  if (opts.stopComponentId) body.stop_component_id = opts.stopComponentId;
+  if (opts.flowData) body.data = opts.flowData;
+  if (opts.files && opts.files.length > 0) body.files = opts.files;
+  return body;
+}
+
+function toInteractiveContent(
+  payload: Record<string, unknown>,
+  jobId: string,
+): InteractiveContent {
+  const allowed = (payload.allowed_decisions as string[]) ?? [];
+  return {
+    type: "human_input",
+    kind: (payload.kind as InteractiveContent["kind"]) ?? "node_input",
+    request_id: String(payload.request_id ?? ""),
+    prompt: payload.prompt as string | undefined,
+    options: (payload.options as InteractiveContent["options"]) ?? [],
+    schema: payload.schema as InteractiveContent["schema"],
+    allowed_decisions: allowed,
+    job_id: jobId,
+  };
+}
+
+/**
+ * Reattach context per pause, keyed by request id. The interactive card resumes the
+ * run itself (no prop-drilling through the chat render chain), so it needs the exact
+ * run opts to stream the continued run back into the right session.
+ */
+const resumeRegistry = new Map<
+  string,
+  { jobId: string; opts: WorkflowRunOptions }
+>();
+
+export function getResumeContext(
+  requestId: string,
+): { jobId: string; opts: WorkflowRunOptions } | undefined {
+  return resumeRegistry.get(requestId);
+}
+
+function _cardSubmittedAction(messageId: string): string | undefined {
+  for (const query of queryClient.getQueryCache().getAll()) {
+    const key = query.queryKey;
+    if (!Array.isArray(key) || key[0] !== "useGetMessagesQuery") continue;
+    const messages = query.state.data as Message[] | undefined;
+    const msg = Array.isArray(messages)
+      ? messages.find((m) => m.id === messageId)
+      : undefined;
+    const content = msg?.content_blocks?.[0]?.contents?.find(
+      (c) => c?.type === "human_input",
+    ) as { submitted_action?: string } | undefined;
+    if (content?.submitted_action) return content.submitted_action;
+  }
+  return undefined;
+}
+
+function cardAlreadyAnswered(messageId: string): boolean {
+  return _cardSubmittedAction(messageId) !== undefined;
+}
+
+/**
+ * Stamp the chosen action onto the card message in the react-query cache so the
+ * selection is derived from a stable source — local React state is lost when the
+ * resume reattach replays the stream and re-renders the message list.
+ */
+export function markHumanInputSubmitted(
+  requestId: string,
+  actionId: string,
+): void {
+  const messageId = `human-input-${requestId}`;
+  for (const query of queryClient.getQueryCache().getAll()) {
+    const key = query.queryKey;
+    if (!Array.isArray(key) || key[0] !== "useGetMessagesQuery") continue;
+    const messages = query.state.data as Message[] | undefined;
+    if (!Array.isArray(messages) || !messages.some((m) => m.id === messageId)) {
+      continue;
+    }
+    queryClient.setQueryData(key, (old: Message[] = []) =>
+      old.map((m) =>
+        m.id === messageId
+          ? {
+              ...m,
+              content_blocks: (m.content_blocks ?? []).map((block) => ({
+                ...block,
+                contents: (block.contents ?? []).map((c) =>
+                  c?.type === "human_input"
+                    ? { ...c, submitted_action: actionId }
+                    : c,
+                ),
+              })),
+            }
+          : m,
+      ),
+    );
+  }
+}
+
+/** Render the pause as an interactive card in the chat and flag awaiting-input. */
+function injectHumanInputCard(
+  payload: Record<string, unknown>,
+  jobId: string,
+  opts: WorkflowRunOptions,
+): void {
+  const content = toInteractiveContent(payload, jobId);
+  resumeRegistry.set(content.request_id, { jobId, opts });
+  const messageId = `human-input-${content.request_id}`;
+  // A resume reattach replays the pause event; if the user already answered (the
+  // card carries submitted_action), re-injecting would clobber that choice. Skip.
+  if (cardAlreadyAnswered(messageId)) {
+    useFlowStore.getState().setAwaitingInput(true);
+    return;
+  }
+  const block: ContentBlock = {
+    title: "Human input required",
+    contents: [content],
+    allow_markdown: true,
+    component: "HumanInput",
+  };
+  const message: Message = {
+    flow_id: opts.flowId,
+    text: "",
+    sender: "Machine",
+    sender_name: "AI",
+    session_id: opts.threadId ?? opts.flowId,
+    timestamp: new Date().toISOString(),
+    files: [],
+    id: messageId,
+    edit: false,
+    background_color: "",
+    text_color: "",
+    content_blocks: [block],
+  };
+  // The new playground reads the react-query messages cache; the legacy IOModal
+  // playground reads useMessagesStore. Write both so the card renders either way.
+  updateMessage(message);
+  useMessagesStore.getState().addMessage(message);
+  useFlowStore.getState().setAwaitingInput(true);
+}
+
+/**
+ * Consume a background run's durable event stream (`GET /{job_id}/events`) and fold
+ * events into the flow store. Resolves when the run ends, suspends for human input,
+ * or the request errors. Reused for the initial run and for resume reattach.
+ *
+ * Uses a direct fetch + SSE parser rather than the AG-UI ``HttpAgent``: a paused run
+ * closes the stream WITHOUT a terminal ``RUN_FINISHED``, which the agent's protocol
+ * verifier treats as a violation and can swallow the pause event we depend on.
+ */
+export async function consumeBackgroundEvents(
+  jobId: string,
+  opts: WorkflowRunOptions & { signal?: AbortSignal },
+  lastEventId?: string,
+): Promise<void> {
+  const flowStore = useFlowStore.getState();
+  const setErrorData = useAlertStore.getState().setErrorData;
+  const touchedNodeIds = new Set<string>();
+  let runId = jobId;
+  let suspended = false;
+
+  const ctx: BridgeContext = {
+    setRunId: (r) => {
+      runId = r;
+    },
+    applyDelta: (ops) => applyStateDelta(ops, runId, touchedNodeIds),
+    handleCustomEvent: (eventType, data) => handleMessageEvent(eventType, data),
+    onHumanInput: (payload) => {
+      suspended = true;
+      injectHumanInputCard(payload as Record<string, unknown>, jobId, opts);
+    },
+    onFinished: () => flowStore.setBuildInfo({ success: true }),
+    onError: (message) => {
+      flowStore.setBuildInfo({ error: [message], success: false });
+      setErrorData({ title: "Workflow run failed", list: [message] });
+    },
+  };
+
+  const finish = () => {
+    flowStore.updateEdgesRunningByNodes([...touchedNodeIds], false);
+    // A suspended run stops building and parks awaiting input; the spinner reads
+    // isBuilding, so it must clear on suspend too — only awaitingInput differs.
+    flowStore.setIsBuilding(false);
+    flowStore.setAwaitingInput(suspended);
+    flowStore.revertBuiltStatusFromBuilding();
+  };
+
+  const headers: Record<string, string> = { Accept: "text/event-stream" };
+  if (lastEventId) headers["Last-Event-ID"] = lastEventId;
+  const url = `${WORKFLOWS_ENDPOINT}/${encodeURIComponent(jobId)}/events`;
+
+  try {
+    const response = await fetch(url, {
+      method: "GET",
+      headers,
+      credentials: "same-origin",
+      signal: opts.signal,
+    });
+    if (!response.ok || !response.body) {
+      ctx.onError(`Events stream failed with status ${response.status}`);
+      finish();
+      return;
+    }
+    const reader = response.body.getReader();
+    const decoder = new TextDecoder();
+    let buffer = "";
+    let terminal = false;
+
+    const dispatch = (frame: string): boolean => {
+      const dataLine = frame
+        .split("\n")
+        .find((line) => line.startsWith("data:"));
+      if (!dataLine) return false;
+      let event: BaseEvent;
+      try {
+        event = JSON.parse(dataLine.slice(5).trim()) as BaseEvent;
+      } catch {
+        return false;
+      }
+      if (handleAGUIEvent(event, ctx)) {
+        // A terminal event supersedes a replayed pause (resume reattach re-emits it).
+        suspended = false;
+        return true;
+      }
+      return false;
+    };
+
+    while (!terminal) {
+      const { value, done } = await reader.read();
+      if (done) {
+        // The stream can close right after the pause event without a trailing
+        // blank line; flush whatever is buffered so that last frame isn't lost.
+        if (buffer.trim()) dispatch(buffer);
+        break;
+      }
+      buffer += decoder.decode(value, { stream: true });
+      let boundary = buffer.indexOf("\n\n");
+      while (boundary !== -1) {
+        const frame = buffer.slice(0, boundary);
+        buffer = buffer.slice(boundary + 2);
+        boundary = buffer.indexOf("\n\n");
+        if (dispatch(frame)) {
+          terminal = true;
+          break;
+        }
+      }
+    }
+    finish();
+  } catch (err) {
+    if ((err as Error)?.name !== "AbortError") {
+      ctx.onError((err as Error)?.message ?? "Events stream error");
+    }
+    finish();
+  }
+}
+
+/**
+ * Run a flow that contains a Human Input node through the durable background path:
+ * submit in background mode, then consume the run's event stream. The pause renders
+ * an interactive card; resume reattaches via {@link consumeBackgroundEvents}.
+ */
+export async function runFlowHITL(
+  opts: WorkflowRunOptions & { signal?: AbortSignal },
+): Promise<void> {
+  const body = buildBackgroundRunRequest(opts);
+  const { data } = await api.post(WORKFLOWS_ENDPOINT, body);
+  const jobId: string | undefined = data?.job_id;
+  if (!jobId) {
+    useFlowStore.getState().setBuildInfo({
+      error: ["Background run did not return a job id"],
+      success: false,
+    });
+    useFlowStore.getState().setIsBuilding(false);
+    return;
+  }
+  await consumeBackgroundEvents(jobId, opts);
+}

src/frontend/src/controllers/API/queries/workflows/use-resume-workflow.ts

@@ -0,0 +1,43 @@
+import type { useMutationFunctionType } from "@/types/api";
+import { WORKFLOWS_ENDPOINT } from "../../agui/run-agent";
+import { api } from "../../api";
+import { UseRequestProcessor } from "../../services/request-processor";
+
+export interface ResumeWorkflowPayload {
+  jobId: string;
+  requestId: string;
+  decision: { action_id: string; values: Record<string, string> };
+}
+
+export interface ResumeWorkflowResponse {
+  job_id: string;
+  status: string;
+  message?: string;
+}
+
+/**
+ * Resume a SUSPENDED human-in-the-loop run with the human's decision.
+ *
+ * POSTs the decision to `/api/v2/workflows/{job_id}/resume` (LE-1450). The caller
+ * re-attaches a fresh `GET /{job_id}/events` from the last seen event id so the
+ * continued run streams gap-free.
+ */
+export const useResumeWorkflow: useMutationFunctionType<
+  undefined,
+  ResumeWorkflowPayload,
+  ResumeWorkflowResponse
+> = (options?) => {
+  const { mutate } = UseRequestProcessor();
+
+  const resumeFn = async (
+    payload: ResumeWorkflowPayload,
+  ): Promise<ResumeWorkflowResponse> => {
+    const res = await api.post(
+      `${WORKFLOWS_ENDPOINT}/${encodeURIComponent(payload.jobId)}/resume`,
+      { request_id: payload.requestId, decision: payload.decision },
+    );
+    return res.data;
+  };
+
+  return mutate(["useResumeWorkflow"], resumeFn, options);
+};

src/frontend/src/stores/flowStore.ts

@@ -10,7 +10,10 @@ import { cloneDeep } from "lodash";
 import { create } from "zustand";
 import { checkCodeValidity } from "@/CustomNodes/helpers/check-code-validity";
 import { queryClient } from "@/contexts";
-import { runFlowAGUI } from "@/controllers/API/agui/run-flow-bridge";
+import {
+  runFlowAGUI,
+  runFlowHITL,
+} from "@/controllers/API/agui/run-flow-bridge";
 import { ENABLE_INSPECTION_PANEL } from "@/customization/feature-flags";
 import { track, trackFlowBuild } from "@/customization/utils/analytics";
 import { brokenEdgeMessage } from "@/utils/utils";
@@ -168,6 +171,7 @@ const useFlowStore = create<FlowStoreType>((set, get) => ({
   nodes: [],
   edges: [],
   isBuilding: false,
+  awaitingInput: false,
   buildStartTime: null,
   buildDuration: null,
   buildingFlowId: null,
@@ -368,6 +372,9 @@ const useFlowStore = create<FlowStoreType>((set, get) => ({
       buildingSessionId: !isBuilding ? null : current.buildingSessionId,
     });
   },
+  setAwaitingInput: (awaitingInput) => {
+    set({ awaitingInput });
+  },
   setBuildStartTime: (time) => {
     set({ buildStartTime: time });
   },
@@ -926,10 +933,8 @@ const useFlowStore = create<FlowStoreType>((set, get) => ({
     const buildController = new AbortController();
     get().setBuildController(buildController);
 
-    // Always run through the v2 workflows endpoint. Current frontend nodes
-    // + edges are sent so unsaved tweaks (dropdowns, text inputs) run as
-    // the user sees them.
-    await runFlowAGUI({
+    // A Human Input node can suspend mid-run — only the durable background path supports it.
+    const runArgs = {
       flowId: currentFlow!.id,
       message: input_value,
       threadId: session,
@@ -938,7 +943,11 @@ const useFlowStore = create<FlowStoreType>((set, get) => ({
       flowData: { nodes: get().nodes, edges: get().edges },
       files,
       signal: buildController.signal,
-    });
+    };
+    const hasHumanInput = get().nodes.some(
+      (node) => node.data?.type === "HumanInput",
+    );
+    await (hasHumanInput ? runFlowHITL(runArgs) : runFlowAGUI(runArgs));
 
     // Invalidate KB-related caches so any KnowledgeIngestion node that ran
     // inside this build surfaces its updated stats / runs the next time the

src/frontend/src/types/chat/index.ts

@@ -133,6 +133,32 @@ export interface ToolContent extends BaseContent {
   error?: JSONValue | string;
 }
 
+// A human-in-the-loop pause: a decision the user must make before the run resumes.
+export interface HumanInputOption {
+  action_id: string;
+  label?: string;
+}
+
+export interface HumanInputFormField {
+  name: string;
+  type?: string;
+  required?: boolean;
+}
+
+export interface InteractiveContent extends BaseContent {
+  type: "human_input";
+  kind: "tool_approval" | "node_input";
+  request_id: string;
+  prompt?: string;
+  options: HumanInputOption[];
+  schema?: HumanInputFormField[];
+  fields?: HumanInputFormField[];
+  allowed_decisions: string[];
+  // Correlation for the resume round-trip; stamped from the run that paused.
+  job_id?: string;
+  submitted_action?: string;
+}
+
 // Union type for all content types
 export type ContentType =
   | ErrorContent
@@ -140,7 +166,8 @@ export type ContentType =
   | MediaContent
   | JSONContent
   | CodeContent
-  | ToolContent;
+  | ToolContent
+  | InteractiveContent;
 
 // Updated ContentBlock interface
 export interface ContentBlock {

src/frontend/src/types/zustand/flow/index.ts

@@ -118,7 +118,9 @@ export type FlowStoreType = {
   buildingFlowId: string | null;
   buildingSessionId: string | null;
   isPending: boolean;
+  awaitingInput: boolean;
   setIsBuilding: (isBuilding: boolean) => void;
+  setAwaitingInput: (awaitingInput: boolean) => void;
   setBuildStartTime: (time: number) => void;
   setBuildDuration: (duration: number) => void;
   setBuildingSession: (flowId: string | null, sessionId: string | null) => void;

src/lfx/src/lfx/graph/checkpoint/schema.py

@@ -59,11 +59,16 @@ def serialize_value(value: Any) -> dict[str, Any] | None:
     if value is None or isinstance(value, (str, int, float, bool)):
         return {_WIRE_KIND: "raw", "value": value}
     if isinstance(value, BaseModel):
+        try:
+            dumped = value.model_dump(mode="json")
+        except Exception:  # noqa: BLE001
+            # A model with an opaque field (LLM client, model class) can't round-trip; degrade to None so the checkpoint stays writable.
+            return None
         return {
             _WIRE_KIND: "model",
             "module": type(value).__module__,
             "name": type(value).__qualname__,
-            "value": value.model_dump(mode="json"),
+            "value": dumped,
         }
     if isinstance(value, dict):
         if not all(isinstance(k, str) for k in value):

src/lfx/src/lfx/schema/content_block.py

@@ -3,7 +3,15 @@ from typing import Annotated
 from pydantic import BaseModel, Discriminator, Field, Tag, field_serializer, field_validator
 from typing_extensions import TypedDict
 
-from .content_types import CodeContent, ErrorContent, JSONContent, MediaContent, TextContent, ToolContent
+from .content_types import (
+    CodeContent,
+    ErrorContent,
+    HumanInputContent,
+    JSONContent,
+    MediaContent,
+    TextContent,
+    ToolContent,
+)
 
 
 def _get_type(d: dict | BaseModel) -> str | None:
@@ -19,7 +27,8 @@ ContentType = Annotated[
     | Annotated[TextContent, Tag("text")]
     | Annotated[MediaContent, Tag("media")]
     | Annotated[CodeContent, Tag("code")]
-    | Annotated[JSONContent, Tag("json")],
+    | Annotated[JSONContent, Tag("json")]
+    | Annotated[HumanInputContent, Tag("human_input")],
     Discriminator(_get_type),
 ]
 

src/lfx/src/lfx/schema/content_types.py

@@ -78,6 +78,24 @@ class CodeContent(BaseContent):
     title: str | None = None
 
 
+class HumanInputContent(BaseContent):
+    """Content type for a human-in-the-loop pause persisted in the chat history.
+
+    Carries the pending decision so the interactive card survives reload: the
+    request_id and job_id let the card resume the suspended run after an F5.
+    """
+
+    type: Literal["human_input"] = Field(default="human_input")
+    request_id: str
+    job_id: str | None = None
+    kind: str = "node_input"
+    prompt: str | None = None
+    options: list[dict[str, Any]] = Field(default_factory=list)
+    fields: list[dict[str, Any]] = Field(default_factory=list)
+    allowed_decisions: list[str] = Field(default_factory=list)
+    submitted_action: str | None = None
+
+
 class ToolContent(BaseContent):
     """Content type for tool start content."""
 

src/lfx/src/lfx/schema/workflow.py

@@ -457,6 +457,21 @@ class WorkflowStopResponse(BaseModel):
     message: str | None = None
 
 
+class WorkflowResumeRequest(BaseModel):
+    """Request schema for resuming a suspended (human-in-the-loop) workflow."""
+
+    request_id: str
+    decision: dict | None = None
+
+
+class WorkflowResumeResponse(BaseModel):
+    """Response schema for resuming a workflow."""
+
+    job_id: JobId
+    status: str
+    message: str | None = None
+
+
 # OpenAPI response definitions
 WORKFLOW_EXECUTION_RESPONSES = {
     200: {

src/lfx/tests/unit/graph/checkpoint/_static_pauser.py

@@ -0,0 +1,42 @@
+"""A real, code-wireable component that pauses then resumes on an injected decision.
+
+HumanInput's branch outputs are dynamic and can't be wired via ``.set()`` in a
+unit test, so this helper mirrors HumanInput's pause/resume CONTRACT with a single
+static output: first run requests a pause; on resume it reads the decision the
+build path injects into ``graph.human_input_decisions`` and carries it downstream.
+This exercises the real resume_from_checkpoint + inject + un-build mechanics that
+``build.py``'s resume branch relies on, with a graph that wires in code.
+"""
+
+from __future__ import annotations
+
+from lfx.custom import Component
+from lfx.io import MessageTextInput, Output
+from lfx.schema.message import Message
+
+HUMAN_INPUT_REQUIRED = "human_input_required"
+
+
+class StaticPauser(Component):
+    display_name = "Static Pauser"
+    name = "StaticPauser"
+
+    inputs = [MessageTextInput(name="input_value", display_name="In")]
+    outputs = [Output(display_name="Out", name="out", method="run_it")]
+
+    def _request_id(self) -> str:
+        return f"{self._id}:{self.graph.run_id}"
+
+    def _decision(self) -> dict | None:
+        decisions = getattr(self.graph, "human_input_decisions", None)
+        return decisions.get(self._request_id()) if isinstance(decisions, dict) else None
+
+    def run_it(self) -> Message:
+        decision = self._decision()
+        if decision is None:
+            self.graph.request_pause(
+                reason=HUMAN_INPUT_REQUIRED,
+                data={"request_id": self._request_id(), "kind": "node_input"},
+            )
+            return Message(text="")
+        return Message(text=str(decision.get("action_id", "")))

src/lfx/tests/unit/graph/checkpoint/test_resume_round_trip.py

@@ -0,0 +1,89 @@
+"""End-to-end graph resume round-trip (LE-1449 / LE-1446 mechanics).
+
+A real ChatInput -> StaticPauser -> ChatOutput graph: first run suspends with a
+node_input request and a durable checkpoint; resume hydrates the checkpoint,
+injects the decision, un-builds the paused node, and runs to completion without
+re-executing already-built vertices. This proves the exact resume-from-checkpoint
++ inject + un-build sequence that build.py's resume branch performs.
+"""
+
+from __future__ import annotations
+
+import sys
+from pathlib import Path
+
+import pytest
+from lfx.components.input_output import ChatInput, ChatOutput
+from lfx.graph import Graph
+from lfx.graph.checkpoint.store import InMemoryCheckpointStore
+from lfx.graph.exceptions import GraphPausedException
+
+sys.path.insert(0, str(Path(__file__).parent))
+from _static_pauser import StaticPauser
+
+
+def _graph(store, *, job_id="job-1"):
+    chat_input = ChatInput(_id="chat_input", input_value="hello")
+    chat_input.set(should_store_message=False)
+    pauser = StaticPauser(_id="pauser")
+    pauser.set(input_value=chat_input.message_response)
+    chat_output = ChatOutput(_id="chat_output")
+    chat_output.set(input_value=pauser.run_it, should_store_message=False)
+    graph = Graph(chat_input, chat_output)
+    graph.session_id = "sess-1"
+    graph.set_run_id(job_id)
+    graph.job_id = job_id
+    graph.checkpointing_enabled = True
+    graph.checkpoint_store = store
+    return graph
+
+
+async def test_first_run_suspends_with_durable_checkpoint():
+    store = InMemoryCheckpointStore()
+    graph = _graph(store)
+
+    with pytest.raises(GraphPausedException) as excinfo:
+        await graph.process(fallback_to_env_vars=False)
+
+    assert excinfo.value.reason == "human_input_required"
+    assert excinfo.value.data["request_id"] == "pauser:job-1"
+    assert graph.get_vertex("chat_input").built is True
+    assert graph.get_vertex("chat_output").built is False
+    assert await store.load_by_run_id("job-1") is not None
+
+
+async def test_resume_injects_decision_runs_to_completion_without_reexec():
+    store = InMemoryCheckpointStore()
+    first = _graph(store)
+    with pytest.raises(GraphPausedException):
+        await first.process(fallback_to_env_vars=False)
+
+    # The build.py resume branch performs exactly these four steps:
+    checkpoint = await store.load_by_run_id("job-1")
+    resumed = Graph.resume_from_checkpoint(checkpoint, checkpoint_store=store)
+    resumed.human_input_decisions = {"pauser:job-1": {"action_id": "approve", "values": {}}}
+    resumed.get_vertex("pauser").built = False
+
+    assert resumed.resume_first_layer() == ["pauser"]
+    assert resumed.get_vertex("chat_input").built is True  # not re-run
+
+    await resumed.process(fallback_to_env_vars=False)
+
+    assert resumed.get_vertex("pauser").built is True
+    assert resumed.get_vertex("chat_output").built is True
+    pauser_result = resumed.get_vertex("pauser").results
+    assert pauser_result["out"].text == "approve"  # the injected decision flowed downstream
+
+
+async def test_resume_without_decision_suspends_again():
+    store = InMemoryCheckpointStore()
+    first = _graph(store)
+    with pytest.raises(GraphPausedException):
+        await first.process(fallback_to_env_vars=False)
+
+    checkpoint = await store.load_by_run_id("job-1")
+    resumed = Graph.resume_from_checkpoint(checkpoint, checkpoint_store=store)
+    resumed.get_vertex("pauser").built = False  # re-run, but no decision injected
+
+    with pytest.raises(GraphPausedException):
+        await resumed.process(fallback_to_env_vars=False)

src/lfx/tests/unit/graph/checkpoint/test_schema.py

@@ -109,3 +109,18 @@ def test_serialize_value_returns_none_for_opaque_objects():
 
     assert serialize_value(Opaque()) is None
     assert serialize_value(lambda: 1) is None
+
+
+def test_serialize_value_degrades_model_with_unserializable_field():
+    """A model holding an opaque field (e.g. an LLM client / model class) must not raise.
+
+    Reproduces the HITL-with-Agent crash: pausing serialized the agent's state and a
+    nested model class blew up ``model_dump(mode="json")``.
+    """
+    from pydantic import BaseModel, ConfigDict
+
+    class Holder(BaseModel):
+        model_config = ConfigDict(arbitrary_types_allowed=True)
+        opaque: type = BaseModel
+
+    assert serialize_value(Holder()) is None