feat(workflow): serve orchestrators through model endpoints

[furionw]

Based on #13081.

Why

The generic frontend should remain independent of workflow authoring and deployment. A workflow orchestrator running as an ordinary discovered model worker still needs a small boundary between Dynamo's endpoint generator ABI and the orchestrator's unary request/chunk contract.

Experimental: This workflow API is under active development and may evolve based on feedback. It is an optional orchestration layer over Dynamo's discoverable workers and endpoints; applications may compose those primitives directly when they prefer to own the orchestration logic.

What Change

• Add WorkflowEndpointHandler for bound orchestrators.
• Validate the endpoint ABI from WorkflowIR.
• Propagate request identity, cancellation, and terminal chunks.

classDiagram
    WorkflowEndpointHandler --> WorkflowOrchestrator : executes request/chunk ABI
    WorkflowOrchestrator --> WorkflowIR : exposes logical endpoint contract
    WorkflowOrchestrator --> GraphScheduler : schedules bound graph
    WorkflowEndpointHandler ..> Endpoint : served by worker

    note for WorkflowEndpointHandler "Adapts discovered endpoint requests"
    note for WorkflowOrchestrator "Owns one bound workflow"

Loading

Test Plan

• Focused CPU workflow and endpoint-handler tests pass.
• Applicable pre-commit hooks pass.

[github-actions]

🌿 Fern Docs Preview: https://nvidia-preview-01a022ae-95a7-703b-b6b8-28963bf83f0d.docs.buildwithfern.com/dynamo/dev

[devin-ai-integration]

Devin Review found 1 potential issue.

[devin-ai-integration]

🟡 Cancellation test can hang CI forever without a timeout

This test drives BlockingRunner, which awaits asyncio.Event().wait() and only unblocks when the handler cancels the workflow. It carries no @pytest.mark.timeout. If the cancellation path regresses, the async-for never completes and the run hangs indefinitely. The pytest guidelines require a timeout on any test with an unbounded wait.

Suggested change

	async def test_endpoint_handler_cancels_workflow_when_context_stops() -> None:
	@pytest.mark.timeout(30)
	async def test_endpoint_handler_cancels_workflow_when_context_stops() -> None:

---

Was this helpful? React with 👍 or 👎 to provide feedback.