langflow-ai/langflow · warning · UnsupportedOperationError

Task {params.id} has no active stream to resubscribe to

Error message

Task {params.id} has no active stream to resubscribe to

What it means

Raised by on_subscribe_to_task (tasks/resubscribe) when the task exists in the durable store but cannot be streamed: its state is not TASK_STATE_WORKING or there is no live in-process ActiveTask registry entry. The SDK's subscribe() taps a live event queue and would block forever for parked (input-required), terminal, or tasks running on another worker, so the guard returns UnsupportedOperationError instead of leaking a blocked subscriber. Clients should fall back to tasks/get polling.

Source

Thrown at src/backend/base/langflow/api/v1/a2a.py:611

        await A2ACheckpointStore().delete_by_run_id(params.id)
        return task

    async def on_subscribe_to_task(self, params, context: ServerCallContext):
        # Reattach to a still-streaming run, but only when there is genuinely a live producer to tail
        # in THIS worker. Two gates, both required:
        #   1. Flow-scoped store: a task this flow can't see is "not found" (same as on_cancel_task);
        #      never reveal that it exists under another flow, and keep the store off the delegate path.
        #   2. Live producer: the SDK's subscribe() taps the task's event queue and waits, so for a
        #      parked (input-required), terminal, or run-on-another-worker task it blocks forever and
        #      leaks an ActiveTask. Require both a WORKING durable state and a live registry entry
        #      before delegating; otherwise return the spec error and let tasks/get read it back.
        stored = await _TASK_STORE.get(params.id, context)
        if stored is None:
            raise TaskNotFoundError
        active = await self._active_task_registry.get(params.id)
        if stored.status.state != pb.TaskState.TASK_STATE_WORKING or active is None:
            msg = f"Task {params.id} has no active stream to resubscribe to"
            raise UnsupportedOperationError(message=msg)
        async for event in super().on_subscribe_to_task(params, context):
            yield event


# One shared httpx client sends webhooks; a short timeout so a slow/hostile webhook
# can't tie up the run. Created at import (no I/O) and reused across requests; closed
# from the app lifespan via close_push_client(). The sender re-validates and DNS-pins
# each webhook at dispatch (per-dispatch client), so this shared client only carries
# the no-pin path (private webhooks allowed / allowlisted host / SSRF protection off).
_PUSH_TIMEOUT = 10.0
_PUSH_HTTP_CLIENT = httpx.AsyncClient(timeout=_PUSH_TIMEOUT)
_PUSH_CONFIG_STORE = _SafePushConfigStore(owner_resolver=_push_config_scope)
_PUSH_SENDER = _SafePushNotificationSender(_PUSH_HTTP_CLIENT, _PUSH_CONFIG_STORE)


async def close_push_client() -> None:
    """Close the shared push-notification webhook client. Wired into the app lifespan."""
    await _PUSH_HTTP_CLIENT.aclose()

View on GitHub (pinned to 976ec789d2)

Solutions

1. Poll tasks/get for task state/output instead of resubscribing when the task is parked or terminal
2. For input-required tasks, submit the requested input (message/send with the task continuation), then stream again
3. In multi-worker deployments, use sticky routing for the task's subsequent requests or rely on durable state reads rather than resubscribe

Example fix

# before
events = await client.tasks.resubscribe(task_id)
# after
if (task := await client.tasks.get(task_id)).status.state in {"completed","failed","canceled","input-required"}:
    return task  # read state, do not stream
raise UnsupportedOperationError

async def task_streamable(client, task_id: str) -> bool:
    task = await client.tasks.get(task_id)
    return task.status.state == "working"  # parked/terminal tasks must be polled, not streamed

def is_streamable_task(task) -> bool:
    state = getattr(getattr(task, "status", None), "state", None)
    return state is not None and getattr(state, "name", str(state)).lower() == "working"

try:
    async for ev in client.tasks.resubscribe(task_id):
        handle(ev)
except ServerError as e:
    if "no active stream" in str(e):
        task = await client.tasks.get(task_id)   # fall back to state polling
        handle_terminal(task)
    else:
        raise

When it happens

Trigger: A2A tasks/resubscribe for (a) a task in input-required state waiting for user input, (b) a COMPLETED/FAILED/CANCELED task, or (c) a WORKING task whose producer lives on a different worker/process than the one handling the resubscribe.

Common situations: Reconnecting a UI after the SSE stream dropped while the agent was waiting on human input; multi-worker deployments where resubscribe lands on a worker that isn't running the task; resubscribing after the task already finished.

Related errors

• str(exc)
• Error processing build events
• Build job not found
• Error in streaming request.
• An error occurred while preparing the flow.

AI-assisted analysis of langflow-ai/langflow@976ec789d2 (2026-08-14). Data as JSON: /api/errors/d1d941392fcc69d2. Report an issue: GitHub.