{"record":{"id":"d1d941392fcc69d2","repo":"langflow-ai/langflow","slug":"task-params-id-has-no-active-stream-to-resubscri","errorCode":null,"errorMessage":"Task {params.id} has no active stream to resubscribe to","messagePattern":"Task (.+?) has no active stream to resubscribe to","errorType":"error_code","errorClass":"UnsupportedOperationError","httpStatus":null,"severity":"warning","filePath":"src/backend/base/langflow/api/v1/a2a.py","lineNumber":611,"sourceCode":"        await A2ACheckpointStore().delete_by_run_id(params.id)\n        return task\n\n    async def on_subscribe_to_task(self, params, context: ServerCallContext):\n        # Reattach to a still-streaming run, but only when there is genuinely a live producer to tail\n        # in THIS worker. Two gates, both required:\n        #   1. Flow-scoped store: a task this flow can't see is \"not found\" (same as on_cancel_task);\n        #      never reveal that it exists under another flow, and keep the store off the delegate path.\n        #   2. Live producer: the SDK's subscribe() taps the task's event queue and waits, so for a\n        #      parked (input-required), terminal, or run-on-another-worker task it blocks forever and\n        #      leaks an ActiveTask. Require both a WORKING durable state and a live registry entry\n        #      before delegating; otherwise return the spec error and let tasks/get read it back.\n        stored = await _TASK_STORE.get(params.id, context)\n        if stored is None:\n            raise TaskNotFoundError\n        active = await self._active_task_registry.get(params.id)\n        if stored.status.state != pb.TaskState.TASK_STATE_WORKING or active is None:\n            msg = f\"Task {params.id} has no active stream to resubscribe to\"\n            raise UnsupportedOperationError(message=msg)\n        async for event in super().on_subscribe_to_task(params, context):\n            yield event\n\n\n# One shared httpx client sends webhooks; a short timeout so a slow/hostile webhook\n# can't tie up the run. Created at import (no I/O) and reused across requests; closed\n# from the app lifespan via close_push_client(). The sender re-validates and DNS-pins\n# each webhook at dispatch (per-dispatch client), so this shared client only carries\n# the no-pin path (private webhooks allowed / allowlisted host / SSRF protection off).\n_PUSH_TIMEOUT = 10.0\n_PUSH_HTTP_CLIENT = httpx.AsyncClient(timeout=_PUSH_TIMEOUT)\n_PUSH_CONFIG_STORE = _SafePushConfigStore(owner_resolver=_push_config_scope)\n_PUSH_SENDER = _SafePushNotificationSender(_PUSH_HTTP_CLIENT, _PUSH_CONFIG_STORE)\n\n\nasync def close_push_client() -> None:\n    \"\"\"Close the shared push-notification webhook client. Wired into the app lifespan.\"\"\"\n    await _PUSH_HTTP_CLIENT.aclose()","sourceCodeStart":593,"sourceCodeEnd":629,"githubUrl":"https://github.com/langflow-ai/langflow/blob/976ec789d2886a86de109c044d089d68e96c9a35/src/backend/base/langflow/api/v1/a2a.py#L593-L629","documentation":"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.","triggerScenarios":"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.","commonSituations":"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.","solutions":["Poll tasks/get for task state/output instead of resubscribing when the task is parked or terminal","For input-required tasks, submit the requested input (message/send with the task continuation), then stream again","In multi-worker deployments, use sticky routing for the task's subsequent requests or rely on durable state reads rather than resubscribe"],"exampleFix":"# before\nevents = await client.tasks.resubscribe(task_id)\n# after\nif (task := await client.tasks.get(task_id)).status.state in {\"completed\",\"failed\",\"canceled\",\"input-required\"}:\n    return task  # read state, do not stream\nraise UnsupportedOperationError","handlingStrategy":"type-guard","validationCode":"async def task_streamable(client, task_id: str) -> bool:\n    task = await client.tasks.get(task_id)\n    return task.status.state == \"working\"  # parked/terminal tasks must be polled, not streamed","typeGuard":"def is_streamable_task(task) -> bool:\n    state = getattr(getattr(task, \"status\", None), \"state\", None)\n    return state is not None and getattr(state, \"name\", str(state)).lower() == \"working\"","tryCatchPattern":"try:\n    async for ev in client.tasks.resubscribe(task_id):\n        handle(ev)\nexcept ServerError as e:\n    if \"no active stream\" in str(e):\n        task = await client.tasks.get(task_id)   # fall back to state polling\n        handle_terminal(task)\n    else:\n        raise","preventionTips":["Always check tasks/get state before resubscribing; only WORKING tasks stream","Treat resubscribe as best-effort: pair every stream with a poll fallback","Expect parked (input-required) tasks to require input submission, not resubscription"],"tags":["a2a","streaming","json-rpc","task-state"],"backgroundTag":null,"analyzedSha":"976ec789d2886a86de109c044d089d68e96c9a35","analyzedAt":"2026-08-14T18:23:12.227Z","schemaVersion":2},"datasetVersion":"2026-08-15T22:17:37.221Z"}