diff --git a/src/backend/base/langflow/api/build.py b/src/backend/base/langflow/api/build.py index 920577e810..0d1a525c25 100644 --- a/src/backend/base/langflow/api/build.py +++ b/src/backend/base/langflow/api/build.py @@ -350,43 +350,6 @@ 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, @@ -827,7 +790,9 @@ async def generate_flow_events( 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) + from langflow.api.v2.hitl import persist_human_input_card + + 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 diff --git a/src/backend/base/langflow/api/v2/hitl.py b/src/backend/base/langflow/api/v2/hitl.py index dcf350e72d..a690d7cb16 100644 --- a/src/backend/base/langflow/api/v2/hitl.py +++ b/src/backend/base/langflow/api/v2/hitl.py @@ -1,4 +1,4 @@ -"""Human-in-the-loop persistence helpers for the v2 workflows API. +"""Human-in-the-loop persistence + decision validation 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 @@ -7,12 +7,64 @@ session renders it as resolved instead of re-offering the decision. from __future__ import annotations +import uuid from uuid import UUID +from lfx.log import logger from sqlalchemy.orm.attributes import flag_modified from langflow.services.deps import get_job_service -from lfx.log import logger + + +async def is_decision_allowed(job_id: UUID, decision: dict) -> bool: + """Whether ``decision.action_id`` is one of the pause's allowed decisions. + + Returns True when there is no pending request constraining the choice (nothing + to enforce); otherwise the chosen action must be in ``allowed_decisions``. + """ + pending = await get_job_service().get_pending_human_request(job_id) + allowed = (pending or {}).get("allowed_decisions") or [] + if not allowed: + return True + return decision.get("action_id") in allowed + + +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. + Records the card's message id in job metadata so resume can mark it answered. + """ + from lfx.schema.content_block import ContentBlock + from lfx.schema.content_types import HumanInputContent + from lfx.schema.message import Message + + from langflow.memory import astore_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) + 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) def _set_submitted_action(content_blocks: list, action_id: str | None) -> bool: @@ -38,9 +90,10 @@ async def mark_card_answered(job_id: UUID, request_id: str, decision: dict) -> N 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 + from langflow.services.database.models.message.model import MessageTable + 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: diff --git a/src/backend/base/langflow/api/v2/workflow.py b/src/backend/base/langflow/api/v2/workflow.py index 4e718e822c..5b5033033b 100644 --- a/src/backend/base/langflow/api/v2/workflow.py +++ b/src/backend/base/langflow/api/v2/workflow.py @@ -1191,6 +1191,19 @@ async def resume_workflow( if job is None or job.type != JobType.WORKFLOW or not (is_owner or current_user.is_superuser): raise _not_found() + from langflow.api.v2.hitl import is_decision_allowed, mark_card_answered + + if not await is_decision_allowed(parsed_job_id, request.decision or {}): + raise HTTPException( + status_code=status.HTTP_422_UNPROCESSABLE_ENTITY, + detail={ + "error": "Invalid decision", + "code": "INVALID_DECISION", + "message": "decision.action_id is not one of the pending request's allowed_decisions.", + "job_id": job_id, + }, + ) + service = get_background_execution_service() if service._frame_source_factory is None: # noqa: SLF001 service._frame_source_factory = _default_frame_source_factory # noqa: SLF001 @@ -1210,8 +1223,6 @@ async def resume_workflow( "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") diff --git a/src/backend/base/langflow/schema/content_block.py b/src/backend/base/langflow/schema/content_block.py index bbe266cc41..94a3f0e41d 100644 --- a/src/backend/base/langflow/schema/content_block.py +++ b/src/backend/base/langflow/schema/content_block.py @@ -20,7 +20,7 @@ def _get_type(d: dict | BaseModel) -> str | None: return getattr(d, "type", None) -# Create a union type of all content types +# Mirror of the lfx ContentType union — keep both in sync (a missing tag breaks MessageRead validation). ContentType = Annotated[ Annotated[ToolContent, Tag("tool_use")] | Annotated[ErrorContent, Tag("error")] diff --git a/src/backend/tests/unit/api/v2/test_resume_route.py b/src/backend/tests/unit/api/v2/test_resume_route.py index 8fc6dc0ddc..24b6b86848 100644 --- a/src/backend/tests/unit/api/v2/test_resume_route.py +++ b/src/backend/tests/unit/api/v2/test_resume_route.py @@ -11,7 +11,7 @@ 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 langflow.services.database.models.jobs.model import Job, JobEvent, JobStatus, JobType from lfx.services.deps import session_scope pytestmark = pytest.mark.usefixtures("client") @@ -75,7 +75,9 @@ async def test_resume_non_suspended_job_409(client, created_api_key): 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)) + 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": {}} @@ -93,3 +95,67 @@ 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 + + +@pytest.fixture +async def suspended_job_with_decisions(created_api_key): + """A SUSPENDED job whose pending request constrains the choice to approve/reject.""" + 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}, + ) + ) + session.add( + JobEvent( + job_id=job_id, + seq=1, + event_type="human_input_required", + payload={"request_id": "req-1", "allowed_decisions": ["approve", "reject"]}, + ) + ) + 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_action_id_not_allowed_422(client, created_api_key, suspended_job_with_decisions): + """B1: a decision whose action_id is not in allowed_decisions is rejected with 422.""" + body = {"request_id": "req-1", "decision": {"action_id": "delete_everything"}} + resp = await client.post( + f"api/v2/workflows/{suspended_job_with_decisions}/resume", json=body, headers=_headers(created_api_key) + ) + assert resp.status_code == 422, resp.text + assert resp.json()["detail"]["code"] == "INVALID_DECISION" + + +async def test_resume_allowed_action_id_accepted_200(client, created_api_key, suspended_job_with_decisions): + """B1: a decision whose action_id is in allowed_decisions passes validation.""" + body = {"request_id": "req-1", "decision": {"action_id": "approve"}} + resp = await client.post( + f"api/v2/workflows/{suspended_job_with_decisions}/resume", json=body, headers=_headers(created_api_key) + ) + assert resp.status_code == 200, resp.text + + +async def test_mark_card_answered_is_noop_without_card_message(client): # noqa: ARG001 + """mark_card_answered degrades gracefully when no card message id is recorded.""" + from langflow.api.v2.hitl import mark_card_answered + + # Unknown job → no job_metadata.card_message_id → returns without raising. + await mark_card_answered(uuid4(), "req-1", {"action_id": "approve"}) diff --git a/src/backend/tests/unit/schema/test_human_input_content_block.py b/src/backend/tests/unit/schema/test_human_input_content_block.py new file mode 100644 index 0000000000..c0a50104d2 --- /dev/null +++ b/src/backend/tests/unit/schema/test_human_input_content_block.py @@ -0,0 +1,39 @@ +"""Contract: the `human_input` content type must validate in BOTH ContentType unions. + +langflow-base and lfx each define a mirrored `ContentType` union. MessageRead uses +one of them; the durable card uses the other. If a tag is added to one union but not +the other, persisting the card fails at read-back with `union_tag_invalid`. This test +guards that drift for the HITL card. +""" + +from __future__ import annotations + +import pytest +from langflow.schema.content_block import ContentBlock as LangflowContentBlock +from lfx.schema.content_block import ContentBlock as LfxContentBlock + +_RAW = { + "type": "human_input", + "request_id": "HumanInput-x:job-1", + "job_id": "job-1", + "prompt": "Approve refund?", + "options": [{"action_id": "approve", "label": "Approve"}], + "allowed_decisions": ["approve"], +} + + +@pytest.mark.parametrize("content_block_cls", [LangflowContentBlock, LfxContentBlock]) +def test_human_input_content_validates_in_both_unions(content_block_cls): + block = content_block_cls(title="Human input required", contents=[dict(_RAW)]) + content = block.contents[0] + assert content.type == "human_input" + assert content.request_id == "HumanInput-x:job-1" + assert content.options[0]["action_id"] == "approve" + + +@pytest.mark.parametrize("content_block_cls", [LangflowContentBlock, LfxContentBlock]) +def test_human_input_content_round_trips_through_json(content_block_cls): + block = content_block_cls(title="Human input required", contents=[dict(_RAW)]) + restored = content_block_cls.model_validate(block.model_dump()) + assert restored.contents[0].type == "human_input" + assert restored.contents[0].submitted_action is None diff --git a/src/frontend/src/components/core/chatComponents/HumanInputCard.tsx b/src/frontend/src/components/core/chatComponents/HumanInputCard.tsx index ce11072cce..63a044f0ea 100644 --- a/src/frontend/src/components/core/chatComponents/HumanInputCard.tsx +++ b/src/frontend/src/components/core/chatComponents/HumanInputCard.tsx @@ -3,10 +3,10 @@ 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"; +} from "@/controllers/API/agui/human-input-card"; +import { consumeBackgroundEvents } 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"; diff --git a/src/frontend/src/components/core/chatComponents/__tests__/HumanInputCard.test.tsx b/src/frontend/src/components/core/chatComponents/__tests__/HumanInputCard.test.tsx index a58fb9a7b5..0d2720dd9a 100644 --- a/src/frontend/src/components/core/chatComponents/__tests__/HumanInputCard.test.tsx +++ b/src/frontend/src/components/core/chatComponents/__tests__/HumanInputCard.test.tsx @@ -16,6 +16,8 @@ jest.mock("@/controllers/API/queries/workflows/use-resume-workflow", () => ({ })); jest.mock("@/controllers/API/agui/run-flow-bridge", () => ({ consumeBackgroundEvents: (...args: unknown[]) => mockConsume(...args), +})); +jest.mock("@/controllers/API/agui/human-input-card", () => ({ getResumeContext: () => ({ jobId: "job-1", opts: { flowId: "f1" } }), markHumanInputSubmitted: jest.fn(), })); @@ -142,6 +144,23 @@ describe("HumanInputCard", () => { expect(screen.getByTestId("human-input-decision-approve")).toBeDisabled(); }); + it("renders resolved on reload when content carries submitted_action", () => { + const content: InteractiveContent = { + ..._approval, + job_id: "job-1", + submitted_action: "approve", + }; + render(); + // Only the chosen option is shown; the others are gone; no resume is fired. + expect( + screen.getByTestId("human-input-decision-approve"), + ).toBeInTheDocument(); + expect( + screen.queryByTestId("human-input-decision-reject"), + ).not.toBeInTheDocument(); + expect(mockResume).not.toHaveBeenCalled(); + }); + it("keeps only the chosen option and removes the others after selecting", () => { const content: InteractiveContent = { ..._approval, job_id: "job-1" }; render(); diff --git a/src/frontend/src/components/core/playgroundComponent/chat-view/chat-messages/components/bot-message.tsx b/src/frontend/src/components/core/playgroundComponent/chat-view/chat-messages/components/bot-message.tsx index ca175b9e2b..c22152d94c 100644 --- a/src/frontend/src/components/core/playgroundComponent/chat-view/chat-messages/components/bot-message.tsx +++ b/src/frontend/src/components/core/playgroundComponent/chat-view/chat-messages/components/bot-message.tsx @@ -7,8 +7,8 @@ import IconComponent, { import MessageMetadata from "@/components/common/messageMetadataComponent"; import { ContentBlockDisplay } from "@/components/core/chatComponents/ContentBlockDisplay"; import HumanInputCard from "@/components/core/chatComponents/HumanInputCard"; +import { findHumanInputContent } from "@/controllers/API/agui/human-input-card"; 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"; @@ -47,11 +47,11 @@ 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 humanInputContent = findHumanInputContent(chat.content_blocks); + const showThinkingDots = + (chatMessage === "" || (isEmpty && !isStreaming)) && + isBuilding && + lastMessage; const { mutate: updateMessageMutation } = useUpdateMessage(); const handleEditMessage = (message: string) => { @@ -228,10 +228,7 @@ export const BotMessage = memo( data-testid={`chat-message-${chat.sender_name}-${chatMessage}`} className="flex w-full flex-col" > - {humanInputContent ? null : (chatMessage === "" || - (isEmpty && !isStreaming)) && - isBuilding && - lastMessage ? ( + {humanInputContent ? null : showThinkingDots ? ( block.contents ?? []) + .find((content) => content?.type === "human_input") as + | InteractiveContent + | undefined; +} + +function toInteractiveContent( + payload: Record, + jobId: string, +): InteractiveContent { + 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: (payload.allowed_decisions as string[]) ?? [], + 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 forEachMessageCache( + fn: (key: unknown[], messages: Message[]) => void, +): void { + for (const query of queryClient.getQueryCache().getAll()) { + const key = query.queryKey; + if (!Array.isArray(key) || key[0] !== MESSAGES_QUERY_KEY) continue; + const messages = query.state.data as Message[] | undefined; + if (Array.isArray(messages)) fn(key, messages); + } +} + +function cardAlreadyAnswered(messageId: string): boolean { + let answered = false; + forEachMessageCache((_key, messages) => { + const msg = messages.find((m) => m.id === messageId); + const content = findHumanInputContent(msg?.content_blocks); + if (content?.submitted_action) answered = true; + }); + return answered; +} + +/** + * 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}`; + forEachMessageCache((key, messages) => { + if (!messages.some((m) => m.id === messageId)) return; + 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. */ +export function injectHumanInputCard( + payload: Record, + 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); +} diff --git a/src/frontend/src/controllers/API/agui/run-flow-bridge.ts b/src/frontend/src/controllers/API/agui/run-flow-bridge.ts index 03b84c21d2..6fa8777cd4 100644 --- a/src/frontend/src/controllers/API/agui/run-flow-bridge.ts +++ b/src/frontend/src/controllers/API/agui/run-flow-bridge.ts @@ -10,21 +10,17 @@ 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 { injectHumanInputCard } from "./human-input-card"; import { buildWorkflowRunRequest, createWorkflowAgent, @@ -355,138 +351,6 @@ function buildBackgroundRunRequest(opts: WorkflowRunOptions) { return body; } -function toInteractiveContent( - payload: Record, - 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, - 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, diff --git a/src/lfx/src/lfx/graph/checkpoint/schema.py b/src/lfx/src/lfx/graph/checkpoint/schema.py index 007368ed68..d6c6b2bfdd 100644 --- a/src/lfx/src/lfx/graph/checkpoint/schema.py +++ b/src/lfx/src/lfx/graph/checkpoint/schema.py @@ -13,6 +13,8 @@ from uuid import uuid4 from pydantic import BaseModel, Field +from lfx.log.logger import logger + _WIRE_KIND = "__lfx_ser__" @@ -55,6 +57,10 @@ def serialize_value(value: Any) -> dict[str, Any] | None: Opaque objects (no faithful JSON form) degrade to None rather than raising: a checkpoint must still be writable when one vertex holds e.g. a client handle, and resume re-derives such objects from the rebuilt component. + + Limitation: paused vertex state that is not JSON-serializable is NOT restored + on resume (it returns as None). A flow that depends on such state downstream of + the pause must re-derive it from the rebuilt node, not from the checkpoint. """ if value is None or isinstance(value, (str, int, float, bool)): return {_WIRE_KIND: "raw", "value": value} @@ -62,7 +68,8 @@ def serialize_value(value: Any) -> dict[str, Any] | None: 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. + # Opaque field (LLM client / model class) can't round-trip; degrade to None. + logger.debug("checkpoint: dropping non-serializable model %s", type(value).__qualname__) return None return { _WIRE_KIND: "model", diff --git a/src/lfx/src/lfx/schema/content_block.py b/src/lfx/src/lfx/schema/content_block.py index bbe266cc41..a101c05add 100644 --- a/src/lfx/src/lfx/schema/content_block.py +++ b/src/lfx/src/lfx/schema/content_block.py @@ -20,7 +20,7 @@ def _get_type(d: dict | BaseModel) -> str | None: return getattr(d, "type", None) -# Create a union type of all content types +# Mirror of the langflow-base ContentType union — keep both in sync (a missing tag breaks MessageRead validation). ContentType = Annotated[ Annotated[ToolContent, Tag("tool_use")] | Annotated[ErrorContent, Tag("error")]