Add arun_deployment and replace @sync_compatible with @async_dispatch

[tomerqodo]

Benchmark PR from agentic-review-benchmarks#6

[claude]

🔴 run_deployment is set as a bare alias (run_deployment = arun_deployment) on line 241, but it needs to be wrapped with @async_dispatch(arun_deployment) to provide sync/async context detection, _sync parameter support, and the .aio attribute. Without this decorator, sync callers will receive a coroutine instead of a FlowRun, tests passing _sync=True will raise TypeError, and run_deployment.aio will not exist.

Extended reasoning...

What the bug is

The PR renames the original run_deployment to arun_deployment (the async implementation) and then sets run_deployment = arun_deployment as a plain alias on line 241. The PR title says "replace @sync_compatible with @async_dispatch", but the @async_dispatch decorator is never actually applied to run_deployment.

How async_dispatch works

The async_dispatch decorator (in src/prefect/_internal/compatibility/async_dispatch.py) creates a wrapper function that:

1. Pops _sync from kwargs (line 89) to determine execution context
2. Uses is_in_async_context() to auto-detect whether to run sync or async (line 91)
3. Dispatches to either the sync function or the async implementation (line 93)
4. Sets a .aio attribute on the wrapper pointing to the async implementation (line 98)

The correct pattern is:

@async_dispatch(arun_deployment)
def run_deployment(...):
    # sync implementation
    ...

Step-by-step proof of failure

Failure 1 — _sync=True causes TypeError:
TestRunDeploymentSyncContext.test_run_deployment_sync_basic calls run_deployment(..., _sync=True). Since run_deployment is just arun_deployment, and arun_deployment has no _sync parameter in its signature, this passes _sync=True directly into the async function. The function will either raise a TypeError: unexpected keyword argument _sync or silently ignore it (depending on how the function handles kwargs), but either way it returns a coroutine object, not a FlowRun.

Failure 2 — .aio attribute missing:
TestAsyncDispatchBehavior.test_run_deployment_aio_attribute asserts hasattr(run_deployment, "aio") and run_deployment.aio is arun_deployment. A plain async function does not have an .aio attribute, so hasattr(run_deployment, "aio") returns False and the test fails with AssertionError.

Failure 3 — Sync callers always get a coroutine:
Without the async_dispatch wrapper, calling run_deployment(...) in a synchronous context will always return a coroutine object (since arun_deployment is an async def). Sync callers would need to manually run the event loop to get a result, which defeats the purpose of having a sync-compatible API.

Impact

This breaks backward compatibility for all sync callers of run_deployment. Any existing code that calls run_deployment(...) synchronously (which was the original API) will receive an unawaited coroutine instead of a FlowRun object. Additionally, the new tests added by this PR (TestRunDeploymentSyncContext and TestAsyncDispatchBehavior) will fail.

How to fix

Replace the bare alias with a properly decorated sync function:

@async_dispatch(arun_deployment)
def run_deployment(
    name: Union[str, UUID],
    client: Optional["PrefectClient"] = None,
    parameters: Optional[dict[str, Any]] = None,
    scheduled_time: Optional[datetime] = None,
    flow_run_name: Optional[str] = None,
    timeout: Optional[float] = None,
    poll_interval: Optional[float] = 5,
    tags: Optional[Iterable[str]] = None,
    idempotency_key: Optional[str] = None,
    work_queue_name: Optional[str] = None,
    as_subflow: Optional[bool] = True,
    job_variables: Optional[dict[str, Any]] = None,
) -> "FlowRun":
    return arun_deployment(
        name=name, client=client, parameters=parameters,
        scheduled_time=scheduled_time, flow_run_name=flow_run_name,
        timeout=timeout, poll_interval=poll_interval, tags=tags,
        idempotency_key=idempotency_key, work_queue_name=work_queue_name,
        as_subflow=as_subflow, job_variables=job_variables,
    )

This will also require importing async_dispatch at the top of the file.

[claude]

🔴 Replacing @inject_client with client, _ = get_or_create_client(client) at line 132 drops client lifecycle management. When no client is provided and none exists in context (e.g., standalone await arun_deployment("my-flow/my-deployment")), get_or_create_client returns a new PrefectClient with inferred=False. The old inject_client decorator would use async with client to properly initialize the HTTP transport and close connections; the new code discards the inferred flag and never context-manages the client, leaking HTTP connections.

Extended reasoning...

What the bug is

The @inject_client decorator at src/prefect/client/utilities.py:74-101 does more than just obtain a client — it manages the client lifecycle. When get_or_create_client creates a brand-new client (no client passed, none in context), it returns (new_client, False). The decorator then uses async with client as new_client to properly call PrefectClient.__aenter__() (which sets up the exit stack, HTTP transport, and ephemeral app lifespan) and __aexit__() (which closes connections and cleans up). When a client is inferred from context (inferred=True), it wraps it in asyncnullcontext since the context owner manages that client.

The new code at line 132 does:

client, _ = get_or_create_client(client)

This discards the inferred boolean with _ and proceeds to use the client directly without ever calling __aenter__ or __aexit__.

Step-by-step proof

1. User calls await arun_deployment("my-flow/my-deployment") with no client argument, outside any Prefect flow/task context.
2. get_or_create_client(None) is called. Since client is None, it checks AsyncClientContext, FlowRunContext, and TaskRunContext — all are None in standalone usage.
3. It falls through to get_client() at line 60 which creates a brand-new PrefectClient, returning (new_client, False).
4. The False (meaning "not inferred, needs context management") is discarded via _.
5. The function proceeds to call client.read_deployment(...), client.create_flow_run_from_deployment(...), etc. on an un-entered client.
6. The PrefectClient.__aenter__ method is never called, so the HTTP transport may not be properly initialized.
7. When arun_deployment returns, PrefectClient.__aexit__ is never called, so HTTP connections are never closed.

Addressing the refutation

The refutation argues that client_injector at line 63-71 uses the same client, _ = get_or_create_client() pattern without context management, so this is an established pattern. However, this is a false equivalence:

• client_injector has always had this behavior — it was designed that way (and arguably has the same bug, as noted by verifiers).
• inject_client (the decorator being replaced in this PR) did properly context-manage the client. This PR is removing that context management, which is a regression.
• The fact that another utility has a similar deficiency does not make introducing the same deficiency into arun_deployment acceptable. The old run_deployment function worked correctly via @inject_client; the new arun_deployment does not.
• The get_client() docstring itself states the client must be context managed.

Impact

When arun_deployment is called standalone (the primary documented usage pattern in the new docstring example), every invocation creates an HTTP client that is never properly closed. This causes:

• HTTP connection leaks (file descriptors, sockets)
• Potential ResourceWarning at garbage collection time
• In long-running processes or loops calling arun_deployment, eventual resource exhaustion

Fix

Restore the lifecycle management pattern from inject_client. Use the inferred boolean to decide whether to context-manage:

client, inferred = get_or_create_client(client)
if inferred:
    context = asyncnullcontext(client)
else:
    context = client
async with context as client:
    # ... rest of the function

Alternatively, re-apply @inject_client to arun_deployment.

[claude]

🔴 Bug: collect_task_run_inputs(v) on line 149 is missing await. Since collect_task_run_inputs is an async def function (src/prefect/utilities/engine.py:66), calling it without await stores coroutine objects as dict values instead of the actual set[TaskRunResult]. This breaks subflow dependency tracking when task_inputs is passed to client.create_task_run(). Fix: restore the await keyword.

Extended reasoning...

What the bug is

collect_task_run_inputs is defined as async def collect_task_run_inputs(expr, max_depth=-1) -> set[...] at src/prefect/utilities/engine.py:66. The PR changed line 149 from k: await collect_task_run_inputs(v) to k: collect_task_run_inputs(v), removing the await keyword inside the dict comprehension.

How it manifests

Without await, calling an async function returns a coroutine object rather than executing the function and returning its result. The task_inputs dict will contain {"param_name": } instead of {"param_name": {TaskRunResult(id=...)}}. This dict is then passed directly to client.create_task_run(task_inputs=task_inputs) on line 194.

Step-by-step proof

1. A user calls arun_deployment("my-flow/my-deployment", parameters={"x": some_upstream_result}) from within a parent flow.
2. Execution reaches line 146 where as_subflow is true and flow_run_ctx exists.
3. The dict comprehension on lines 148-149 iterates over parameters.items(). For key "x", it calls collect_task_run_inputs(some_upstream_result) without await.
4. The value stored for key "x" is a coroutine object, not the expected set[TaskRunResult] containing the upstream task run reference.
5. This corrupted task_inputs is passed to client.create_task_run() on line 194, which expects sets of TaskRunResult/FlowRunResult objects.
6. The Prefect server/API will either fail to serialize the coroutine objects or silently drop the dependency information, breaking the execution graph linkage between the parent task and the subflow deployment run.

Why existing code does not prevent it

Python does not raise an error when you call an async function without await - it simply returns a coroutine object. The type checker likely will not catch this either since the comprehension is inside an async def where await is valid but not enforced. Additionally, the unawaited coroutines will generate RuntimeWarning: coroutine was never awaited at garbage collection time, but this will not prevent the code from executing.

Impact

This breaks subflow dependency tracking for any deployment run triggered from within a flow that passes parameters. The execution graph will not correctly show parameter-based dependencies between upstream tasks and the deployment flow run. The existing test test_tracks_dependencies_when_used_in_flow should catch this regression, as it asserts that task_run.task_inputs contains the correct TaskRunResult references.

Fix

Restore the await keyword on line 149:

task_inputs = {
k: await collect_task_run_inputs(v) for k, v in parameters.items()
}

[claude]

🔴 Bug: deployment_id is None when a deployment name string (e.g., "flow/deployment") is passed, causing create_flow_run_from_deployment(None, ...) to fail. Line 212 should use deployment.id instead of deployment_id, since deployment_id is only set for UUID inputs and is never updated after the name-based lookup via read_deployment_by_name.

Extended reasoning...

What the bug is

On line 212 of src/prefect/deployments/flow_runs.py, the code passes deployment_id as the first argument to client.create_flow_run_from_deployment(). However, when the user passes a deployment name string rather than a UUID, deployment_id remains None.

Code path walkthrough

Here is a concrete step-by-step trace when calling arun_deployment("my-flow/my-deployment"):

1. Line 123: deployment_id = None -- initialized to None.
2. Line 125: isinstance(name, UUID) is False (it is a str), so we skip to the else branch.
3. Lines 128-131: UUID("my-flow/my-deployment") raises ValueError, caught by the except -- deployment_id remains None.
4. Line 132: get_or_create_client(client) creates a client.
5. Line 134: if deployment_id: is False (it is None), so we go to the else branch.
6. Line 138: deployment = await client.read_deployment_by_name(name) -- the deployment object is successfully resolved, with a valid .id attribute.
7. Crucially, deployment_id is never updated to deployment.id after this point.
8. Line 212: await client.create_flow_run_from_deployment(deployment_id, ...) passes None as the deployment ID.

The create_flow_run_from_deployment method expects a UUID parameter (deployment_id: UUID per the client signature). Passing None will cause an error.

Why existing code does not prevent it

The original code (before this PR) used deployment.id on the equivalent line, which was always valid because deployment is populated in both the UUID and name-based branches. This PR replaced deployment.id with deployment_id without ensuring deployment_id gets updated in the name-based lookup path.

Impact

This is a regression that breaks all name-based deployment lookups. Any call like arun_deployment("flow-name/deployment-name") or run_deployment("flow-name/deployment-name") will fail at the create_flow_run_from_deployment call. Only UUID-based lookups (where deployment_id is correctly set) will work.

Fix

Revert line 212 to use deployment.id instead of deployment_id:

flow_run = await client.create_flow_run_from_deployment(
deployment.id,
...
)

Alternatively, add deployment_id = deployment.id after line 138 (the read_deployment_by_name call) to keep deployment_id in sync. Either approach resolves the issue.