fix(hitl): rename inputs, advanced timeout + HITL palette and control polish (#14090)

* fix(hitl): rename inputs, advanced timeout (backend) + fix(frontend): HITL palette and control polish

* [autofix.ci] apply automated fixes

* fix(ci): refresh generated assets

* fix(hitl): atomic supersede + close stale cards

* remove unecessary files

* [autofix.ci] apply automated fixes

* add hitl metadata

* [autofix.ci] apply automated fixes

* [autofix.ci] apply automated fixes

---------

Co-authored-by: autofix-ci[bot] <114827586+autofix-ci[bot]@users.noreply.github.com>
Co-authored-by: Eric Hare <ericrhare@gmail.com>

Author: 3 people

.secrets.baseline

@@ -127,6 +127,18 @@ The pause probe is a no-op unless a component requests it, so normal flows are b
 - **Then** the run finalizes once — no loop back to the first node
 - **And** only the chosen branch of each node runs (the first node's non-chosen branches stay dead across the second pause's resume, not just within the build that answered them)
 
+### Scenario: Rerunning a paused flow supersedes the stale pause
+- **Given** a flow with a `SUSPENDED` run awaiting a decision
+- **When** the same user submits a new background run of that flow
+- **Then** the stale suspended run is cancelled (`CANCELLED`, pending request cleared) before the new run starts, so only the new pause is offered on every surface
+- **And** the persisted chat card of the superseded pause is stamped `superseded` — after a history reload it renders closed ("Superseded by a new run") instead of turning interactive again and 409ing on every click
+- **And** the scope is flow + user on the langflow backend (another user's pause on the same flow is untouched) and flow on `lfx serve` (single API-key identity); running jobs are never superseded, so parallel runs stay supported
+
+### Scenario: Supersede loses the race to a resume
+- **Given** a `SUSPENDED` run whose resume arrives while a rerun's supersede is in flight
+- **When** the resume wins the atomic `SUSPENDED → IN_PROGRESS` claim between supersede's select and its cancel
+- **Then** supersede skips that job entirely — no `CANCELLED` write, no checkpoint delete, no metadata clear (backend) and no lingering STOP signal (`lfx serve`) — and the resumed run finishes normally
+
 ### Scenario: Single-flight resume
 - **Given** two resume requests for the same `SUSPENDED` job
 - **When** both arrive
@@ -225,12 +237,19 @@ Drain the queue **inline** after cancelling the worker (`service.py::_stop`), an
 
 ## 6. Technical Specification
 
-### 6.1 Pause → suspend (initial run)
+### 6.1 Submit supersedes stale pauses
+- `BackgroundExecutionService.submit` calls `supersede_suspended_runs(flow_id, user_id)` after `create_job` (so idempotent retries still return the existing job) and before enqueue: every `SUSPENDED` workflow job of the same flow + user is cancelled through the `_cancel_suspended` path (status `CANCELLED`, card stamped superseded, checkpoint deleted, pending cleared, `run_cancelled` event, bus closed).
+- The cancel is an **atomic claim**: `JobService.claim_suspended_for_cancel` (a conditional `UPDATE … WHERE status = SUSPENDED`, the mirror of `claim_suspended_for_resume`) flips the row, and cleanup runs only for rows this caller actually claimed. A resume that won the flip first keeps its run — supersede can never cancel a resumed job or destroy its checkpoint mid-flight. `stop_job` uses the same claim and falls through to the running-job STOP path when it loses.
+- Before metadata is cleared, `mark_card_superseded` (`api/v2/hitl.py`) patches the persisted card's `human_input` content with `superseded: true` (skipping already-answered cards); `HumanInputCard` renders that state closed, so a reloaded history cannot re-offer a decision that would only 409.
+- `DurableServeWorkflowHost.submit_background` mirrors it flow-scoped: `SqliteDurableJobStore.claim_suspended_for_cancel` first, then (only for claimed rows) STOP signal, task cancel, pending metadata cleared — a lost claim also means no stray STOP signal for the resumed continuation to consume.
+- Frontend: the canvas badge auto-open is keyed by `request_id`, so the superseding run's new pause reopens a dismissed popover; the 5s pending poll converges every surface onto the single remaining pause.
+
+### 6.2 Pause → suspend (initial run)
 - The graph consults a pause probe at each layer boundary; on a pending pause it snapshots itself, writes a `checkpoint` row (always-writable schema), and raises `GraphPausedException`.
 - The build seam (`api/build.py`) catches it, persists the human-input card to history, emits `human_input_required`, and ends the stream without `on_end`.
 - The runner marks the job `SUSPENDED` (`services/background_execution/runner.py`).
 
-### 6.2 Resume (single-flight)
+### 6.3 Resume (single-flight)
 - `POST /api/v2/workflows/{job_id}/resume` with `{request_id, decision}`.
 - The route re-enforces `FlowAction.EXECUTE` (`ensure_resume_execute_permission`) before applying the decision — job ownership alone is not enough once shared access has been revoked.
 - `claim_suspended_for_resume` performs the atomic status claim; failure → `NOT_RESUMABLE`.
@@ -253,7 +272,7 @@ Drain the queue **inline** after cancelling the worker (`service.py::_stop`), an
   # restore built vertices; re-run only non-input predecessors of the gated vertex
   ```
 
-### 6.3 Trace continuity (initial run)
+### 6.4 Trace continuity (initial run)
 - `build_graph_from_data` (`api/utils/flow_utils.py`) pins the run id before tracing starts:
   ```python
   if (caller_run_id := kwargs.get("run_id")) is not None:
@@ -262,15 +281,15 @@ Drain the queue **inline** after cancelling the worker (`service.py::_stop`), an
   ```
 - `create_graph` forwards `run_id=str(job_id) if job_id is not None else run_id` to both `build_graph_from_data` and `build_graph_from_db`.
 
-### 6.4 Gate span recording
+### 6.5 Gate span recording
 - `TracingService.record_event_span(span_id, name, outputs, span_type="chain")` writes a standalone start+end span into the active trace for every ready tracer; deactivated/contextless calls are no-ops.
 
-### 6.5 Frontend (reads backend spans)
+### 6.6 Frontend (reads backend spans)
 - `useGetTraceQuery` → `GET /api/v1/monitor/traces/{id}` returns the span tree.
 - `TraceDetailView.tsx`: renders the backend gate span; `backendHasGate` dedup prevents a duplicate synthetic node; `executedOutputSpans` bridges Chat Output from live `flowPool` only during the resume window (deduped by name).
 - `hitlStore.ts`: in-memory only (`pending` slot); `localStorage`/`persist` removed.
 
-### 6.6 lfx serve durable mode (LE-1695)
+### 6.7 lfx serve durable mode (LE-1695)
 
 Bare `lfx serve` gains the same background + HITL contract without a database, backed by the single-node SQLite substrate:
 
@@ -282,7 +301,7 @@ Bare `lfx serve` gains the same background + HITL contract without a database, b
 - **Opt-in**: `LFX_SERVE_DURABLE_DB=<path>`. Unset → serve stays stateless and `mode: background` keeps its 422; workers in multi-worker mode inherit the env var and share the DB file (a stop cancels the run only in the worker executing it).
 - **Proof**: `src/lfx/tests/unit/cli/test_serve_durable.py` — background echo completes with the sync-shaped response; a HITL flow suspends, exposes its pending request, resumes only the approved branch, and resumes to completion on a fresh app instance over the same DB file (restart equivalence). The `{job_id}/events` SSE stream replays a completed run, surfaces the `human_input_request` on a suspend, and — reconnected with `Last-Event-ID` after a resume — delivers only the post-resume `human_input_decision` + `end` frames.
 
-### 6.7 Component map (quick index)
+### 6.8 Component map (quick index)
 
 | Concern | Where | Key symbols |
 |---------|-------|-------------|

docs/features/durable-execution-hitl.md

@@ -166,6 +166,54 @@ def _set_submitted_action(content_blocks: list, action_id: str | None, request_i
     return changed
 
 
+def _mark_superseded(content_blocks: list) -> bool:
+    """Flag every still-open human_input content as superseded, in place."""
+    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 []:
+            is_dict = isinstance(content, dict)
+            ctype = content.get("type") if is_dict else getattr(content, "type", None)
+            if ctype != "human_input":
+                continue
+            answered = content.get("submitted_action") if is_dict else getattr(content, "submitted_action", None)
+            if answered is not None:
+                continue
+            if is_dict:
+                content["superseded"] = True
+            else:
+                content.superseded = True
+            changed = True
+    return changed
+
+
+async def mark_card_superseded(job_id: UUID) -> None:
+    """Close the persisted card of a run cancelled by supersede/stop.
+
+    Without this the reloaded card (no ``submitted_action``) turns interactive
+    again and every click 409s against the cancelled job. Patched in place like
+    ``mark_card_answered`` so prompt/options are preserved for the history.
+    """
+    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:
+        return
+    try:
+        async with session_scope() as session:
+            message = await session.get(MessageTable, UUID(str(card_message_id)))
+            if message is None:
+                return
+            if _mark_superseded(message.content_blocks):
+                flag_modified(message, "content_blocks")
+                session.add(message)
+    except Exception as _e:  # noqa: BLE001
+        await logger.awarning("Failed to mark human-input card superseded for job %s: %r", job_id, _e)
+
+
 async def mark_card_answered(job_id: UUID, request_id: str, decision: dict, *, card_message_id=None) -> None:
     """Record the chosen action on the persisted card message for ``job_id``.
 

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

@@ -700,7 +700,7 @@
                   },
                   {
                     "name": "cryptography",
-                    "version": "48.0.1"
+                    "version": "49.0.0"
                   },
                   {
                     "name": "langchain_chroma",

src/backend/base/langflow/initial_setup/starter_projects/Document Q&A.json

@@ -706,7 +706,7 @@
                   },
                   {
                     "name": "cryptography",
-                    "version": "48.0.1"
+                    "version": "49.0.0"
                   },
                   {
                     "name": "langchain_chroma",

src/backend/base/langflow/initial_setup/starter_projects/Knowledge Retrieval.json

@@ -1169,7 +1169,7 @@
                   },
                   {
                     "name": "cryptography",
-                    "version": "48.0.1"
+                    "version": "49.0.0"
                   },
                   {
                     "name": "langchain_chroma",

src/backend/base/langflow/initial_setup/starter_projects/Vector Store RAG.json

@@ -3925,13 +3925,13 @@
   "components.huggingfacemodel.inputs.typical_p.info.ed0e56c7": "Typical Decoding mass.",
   "components.huggingfacemodel.outputs.model_output.display_name.698eabeb": "Language Model",
   "components.huggingfacemodel.outputs.text_output.display_name.2544139a": "Model Response",
-  "components.humaninput.description.e35d40a8": "Pause the flow to collect a human decision and route on the chosen action.",
+  "components.humaninput.description.548359b3": "Pause the flow for a human-in-the-loop (HITL) decision and route on the chosen action.",
   "components.humaninput.display_name.d7c8a913": "Human Input",
-  "components.humaninput.inputs.decisions.display_name.15670189": "User Actions",
-  "components.humaninput.inputs.decisions.info.5126a1fd": "Actions the human can choose; each becomes a branch output. Pick from the list or type a custom one.",
+  "components.humaninput.inputs.decisions.display_name.8ce3a776": "User Choices",
+  "components.humaninput.inputs.decisions.info.ad649edf": "Choices the human can pick; each becomes a branch output. Pick from the list or type a custom one.",
   "components.humaninput.inputs.enable_fallback.display_name.109c8318": "Enable Fallback",
   "components.humaninput.inputs.enable_fallback.info.72174e80": "Add a 'fallback' output taken when the answer arrives after the timeout window.",
-  "components.humaninput.inputs.prompt.display_name.6a8db532": "Form Content",
+  "components.humaninput.inputs.prompt.display_name.36ecb4f8": "Input",
   "components.humaninput.inputs.prompt.info.e40f2bd4": "Content shown to the human for review.",
   "components.humaninput.inputs.timeout.display_name.70594d93": "Timeout",
   "components.humaninput.inputs.timeout.info.866d9f42": "A response received after this window is rerouted to the fallback path (when enabled) instead of the chosen action. The run stays paused until a response arrives or the server's suspended-run deadline expires it. Set to 0 to wait indefinitely.",

src/backend/base/langflow/locales/en.json

@@ -162,6 +162,39 @@ async def teardown(self) -> None:
 
     # ------------------------------------------------------------------ submit
 
+    async def supersede_suspended_runs(self, *, flow_id: UUID, user_id: UUID) -> list[UUID]:
+        """Cancel this user's SUSPENDED runs of the flow so a rerun replaces the stale pause.
+
+        A suspended run holds a pause nobody will answer once its owner reruns the flow;
+        left alive it piles up in every pending surface (badge, cards, trace bar). Scope is
+        flow + user: another user's pause on the same flow is never touched, and running
+        jobs are untouched so parallel runs stay supported.
+        """
+        from sqlmodel import select
+
+        from langflow.services.database.models.jobs.model import Job
+        from langflow.services.deps import session_scope
+
+        job_service = get_job_service()
+        async with session_scope() as session:
+            result = await session.exec(
+                select(Job.job_id).where(
+                    Job.flow_id == flow_id,
+                    Job.user_id == user_id,
+                    Job.status == JobStatus.SUSPENDED,
+                    Job.type == JobType.WORKFLOW,
+                )
+            )
+            stale_job_ids = list(result.all())
+        cancelled: list[UUID] = []
+        for stale_job_id in stale_job_ids:
+            if self._scaled:
+                await self._backend.stop(str(stale_job_id))
+                cancelled.append(stale_job_id)
+            elif await self._cancel_suspended(stale_job_id, job_service):
+                cancelled.append(stale_job_id)
+        return cancelled
+
     async def submit(self, *, flow_id: UUID, request: dict[str, Any], user: UserRead) -> UUID:
         # Lazy-start the executor so the facade works whether or not the app
         # lifespan called start() first. start() is idempotent.
@@ -199,6 +232,9 @@ async def submit(self, *, flow_id: UUID, request: dict[str, Any], user: UserRead
         # variables by name for background runs rather than passing secrets inline.
         # The live in-memory run below still uses the full ``request``.
         await job_service.update_job_metadata(job_id, {"request": self._redact_request(request)})
+        # After create_job so an idempotent retry returns the existing job instead of
+        # cancelling it; the new job is QUEUED, so the suspended-only query skips it.
+        await self.supersede_suspended_runs(flow_id=flow_id, user_id=user.id)
         if self._scaled:
             # Scaled mode: hand the QUEUED job id to a worker via the redis claim
             # queue; the worker hydrates the request from the job row.
@@ -366,8 +402,9 @@ async def stop_job(self, job_id: UUID, user: UserRead) -> None:
             await self._backend.stop(str(job_id))
             return
         job_service = get_job_service()
-        if job.status == JobStatus.SUSPENDED:
-            await self._cancel_suspended(job_id, job_service)
+        # A lost claim means a resume flipped it IN_PROGRESS mid-call — fall
+        # through to the running-job stop path instead of doing nothing.
+        if job.status == JobStatus.SUSPENDED and await self._cancel_suspended(job_id, job_service):
             return
         await job_service.write_signal(job_id, SignalType.STOP)
         await self._executor.cancel(str(job_id))
@@ -408,14 +445,25 @@ async def resume_job(self, job_id: UUID, user: UserRead, *, request_id: str, dec
             raise
         return True
 
-    async def _cancel_suspended(self, job_id: UUID, job_service) -> None:
-        await job_service.update_job_status(job_id, JobStatus.CANCELLED, finished_timestamp=True)
+    async def _cancel_suspended(self, job_id: UUID, job_service) -> bool:
+        """Cancel a SUSPENDED job; False when a concurrent resume already claimed it.
+
+        The conditional claim is what makes supersede-vs-resume safe: cleanup
+        (checkpoint delete, metadata clear, terminal event) runs only for the row
+        this caller actually flipped, never for a run a resume just revived.
+        """
+        if not await job_service.claim_suspended_for_cancel(job_id):
+            return False
+        from langflow.api.v2.hitl import mark_card_superseded
+
+        await mark_card_superseded(job_id)
         with contextlib.suppress(Exception):
             await job_service.delete_checkpoint(job_id, "graph")
         await job_service.update_job_metadata(job_id, {"pending_request_id": None})
         await job_service.append_event(job_id, "run_cancelled", {"type": "cancelled"})
         await job_service.consume_signals(job_id, SignalType.STOP)
         await self._bus.close(str(job_id))
+        return True
 
     # ----------------------------------------------------------- startup sweep
 

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

@@ -615,6 +615,26 @@ async def claim_suspended_for_resume(self, job_id: UUID, *, owner: str | None =
             await session.flush()
             return result.rowcount == 1
 
+    async def claim_suspended_for_cancel(self, job_id: UUID) -> bool:
+        """Atomically flip SUSPENDED->CANCELLED; True iff this caller won.
+
+        Mirror of ``claim_suspended_for_resume`` for the supersede/stop side of the
+        race: a resume that already flipped the row to IN_PROGRESS keeps its run —
+        an unconditional CANCELLED write here would clobber it and destroy its
+        checkpoint mid-flight.
+        """
+        from sqlmodel import update
+
+        async with session_scope() as session:
+            stmt = (
+                update(Job)
+                .where(Job.job_id == job_id, Job.status == JobStatus.SUSPENDED)
+                .values(status=JobStatus.CANCELLED, finished_timestamp=datetime.now(timezone.utc))
+            )
+            result = await session.exec(stmt)  # type: ignore[call-overload]
+            await session.flush()
+            return result.rowcount == 1
+
     async def queued_workflow_job_ids(self) -> list[UUID]:
         """Return the ids of every QUEUED workflow job (for strand recovery)."""
         async with session_scope() as session: