Workflow Execution Dispatch#

AppMode.WORKFLOW and AppMode.ADVANCED_CHAT apps share a dual-mode dispatch mechanism controlled by a single streaming flag. The entry point is AppGenerateService._dispatch_generate(), which routes each execution onto one of two distinct paths.

streaming=True → serialize AppExecutionParams → Celery task → Redis pub/sub → SSE
streaming=False → call generator directly → return JSON blocking response

Streaming Path (Celery async)#

When streaming=True, the dispatcher 1:

1. Builds an AppExecutionParams object containing app_id, workflow_id, tenant_id, user (discriminated union of _Account or _EndUser), args, invoke_from, and a pre-generated workflow_run_id.
2. Serializes it to JSON via payload.model_dump_json().
3. Dispatches workflow_based_app_execution_task.delay(payload_json) to the workflow_based_app_execution Celery queue.

Task startup is coordinated by _build_streaming_task_on_subscribe(), which gates task start on the first SSE subscriber via the on_subscribe callback:

• Redis Streams: A "prepared subscription" is established before task dispatch using prepare_subscription(), which snapshots the current tail of the stream as a delivery boundary. This checkpoint fixes the stream entry ID before any events are published, ensuring all events from the task are captured even if the subscriber's listener thread starts late. The task itself still waits for the first subscriber via on_subscribe (preventing the race condition where the task could complete before the SSE subscription activates).
• Pub/Sub: The task starts on the first SSE subscriber, with a 200 ms fallback timer to handle clients that never connect. No prepared subscription is used.

In the Celery worker, _AppRunner.run() reconstructs the Account or EndUser from the serialized params, sets up a Flask application context, then calls _AppRunner._run_app(), which dispatches to:

• AdvancedChatAppGenerator().generate(..., streaming=True) for AppMode.ADVANCED_CHAT
• WorkflowAppGenerator().generate(..., streaming=True) for AppMode.WORKFLOW

The returned generator is consumed by _publish_streaming_response(), which publishes each event to the Redis topic that the SSE subscriber is listening on.

Blocking Path (synchronous)#

When streaming=False, the generator is called directly in the request thread 2:

• A PauseStateLayerConfig is constructed from session_factory and workflow.created_by.
• The appropriate generator's .generate() is called with streaming=False.
• The result is a Mapping (JSON dict) returned synchronously to the caller.

The blocking path skips Celery, Redis, and the SSE subscriber loop entirely — the response is returned as a plain JSON dict after the workflow finishes.

File Access Scope#

Both paths establish a FileAccessScope before any file references are resolved. BaseAppGenerator._bind_file_access_scope() is a context manager that constructs FileAccessScope(tenant_id, user_id, user_from, invoke_from) and binds it to a Python ContextVar.

In the blocking path, the scope is established inside AdvancedChatAppGenerator.generate() 3 before file objects are parsed from args["files"]. In the streaming path, the scope is re-established inside _generate() in the Celery worker — the ContextVar does not survive serialization, so it must be re-bound in the worker process before execution proceeds.

DatabaseFileAccessController (a class-level singleton on BaseAppGenerator) then enforces a tenant_id equality filter on every UploadFile and ToolFile database lookup within the scope. 4

Terminal Event Safety#

If execution fails before the generator produces any events, _publish_failed_workflow_terminal_events() synthesizes a workflow_started + workflow_finished(FAILED) pair so that SSE consumers don't hang. If the generator exhausts without emitting a terminal event (workflow_finished or workflow_paused), _publish_streaming_response() emits a fallback failed terminal event to close the stream cleanly.

Key Files#