NVIDIANVIDIA-Certified Professional: Agentic AI2026-104.1 Deploy and orchestrate multi-agent systems at production scale.
Recover distributed tasks from persisted progress
Completed by the knowledge check
Learning objective
Coordinate distributed agent tasks with queues, timeouts, and durable state.
Core idea
A background job needs more than a detached HTTP request. A scheduler decides when workers run it; a deadline decides whether it is still useful; persisted state records what the system can recover after a process fails. Design all three together.
NeMo Agent Toolkit’s asynchronous endpoint combines Dask execution with SQL-backed job history and metadata. That is a concrete starting point, but job records and resumable agent execution are different responsibilities. Here the question is how distributed work remains bounded and recoverable when workers and schedulers fail independently.
How it works
Submit an identified job, then schedule its work
Install nvidia-nat[async_endpoints] to enable the asynchronous workflow endpoint. The API server documents /v1/workflow/async, with legacy /generate/async. Versioned workflow paths are marked experimental because of HTTP human-in-the-loop and OAuth support; the documented compatibility with legacy paths covers workflows that do not rely on those behaviors.
A request can supply job_id. If omitted, the server generates a UUID; if supplied, uniqueness is the caller’s responsibility. When that identifier already exists, the endpoint returns the job’s status. Reusing a known identifier after a lost submission response is therefore different from submitting another identifier and potentially starting another workflow. This identifier behavior does not make every external action inside the workflow execute exactly once; see 2.4.4 for side-effect recovery.
The endpoint uses Dask for asynchronous jobs. Connect to an external scheduler with nat serve --scheduler_address SCHEDULER_ADDRESS, replacing SCHEDULER_ADDRESS with its network address. This separates API service replacement from the scheduler’s lifecycle and creates a separately owned dependency.
Dask distinguishes tasks waiting on dependencies from runnable root tasks queued because workers are full. Its scheduler state documentation calls those internal states waiting and queued, respectively. Treat these as diagnostic concepts, not a stable application API: that page explicitly identifies scheduler internals as subject to change. A queue can absorb a burst, but it cannot increase the rate at which worker and inference resources complete jobs.
For local clusters, --max_concurrent_jobs is documented as a misleading name for the maximum worker count. It and --dask_threads_per_worker do not apply when scheduler_address is set. Configure and measure the external cluster’s capacity instead.
Separate response waiting from useful execution time
The request field sync_timeout controls how long the submission waits for completion. Its default is 0, returning immediately; its maximum is 300 seconds. If work finishes inside the waiting interval, the response includes the result; otherwise it returns job_id and status. A short wait changes the client interaction, not the job’s allowed execution duration. The submission returns the result only when work finishes inside the waiting interval, and reusing a known job_id returns the job’s status.
Likewise, expiry_seconds controls preservation of output files after completion. It defaults to 3600, has a minimum of 600 and a maximum of 86400, and does not affect the response’s text output. It is neither queue expiration nor execution cancellation.
Define an application deadline covering queue wait and execution. Store it with the job request, and check remaining time before expensive work. Apply that remaining budget with asyncio.timeout() inside the Python worker; see 1.8.2 for cancellation and timeout handling. A rescheduled task inherits the accepted deadline instead of receiving a fresh execution budget.
An execution timeout does not undo a remote action already performed. Keep the timeout outcome distinct from confirmed failure of that action. The recovery rules in 2.4.4 determine whether an uncertain side effect can be replayed; adding another queue delivery does not resolve that uncertainty.
Persist both job records and resumable computation
See 1.8.3 for persistent job storage through --db_url and external scheduling through --scheduler_address. Those settings separate job metadata from the scheduler’s computation records; recovery must account for both.
Dask documents two distinct failure cases. After worker loss, it can reroute pending computations and recompute missing results from the task graph. If the scheduler itself dies, its ongoing computation records are lost: the resilience documentation states that scheduler state has no persistence-and-recovery mechanism. A surviving SQL job row is therefore insufficient proof that its computation continues after scheduler restart.
See 1.8.3 for graph checkpointers, configurable.thread_id, and the "sync" versus "async" durability trade-off. For long jobs, durability="exit" cannot preserve intermediate progress through a process crash because it saves only when execution exits.
Account for the invocation boundary: the Toolkit v1.8.0 implementation of langgraph_wrapper calls graph.ainvoke() and graph.astream() without passing a RunnableConfig. Its 1.8 documentation’s State Management section warns that complex state management may require code modifications. For the recovery path here, the application worker invokes the checkpointed graph directly with an explicit configurable.thread_id; it does not use the unmodified wrapper or assume that async job_id supplies thread identity.
Make reconciliation an application procedure with explicit records:
- Read the application’s durable admission record, containing
job_id, the chosenthread_id, the accepted deadline, and its terminal outcome if one exists. These are application fields, not additional Toolkit request fields. Compare that record with SQL job history to identify submissions without a confirmed terminal result. - For unfinished work, check the deadline before scheduling recovery. If it has passed, record an application-level expired outcome and notify the caller; do not enqueue more inference under the original commitment.
- For eligible work, load the saved graph state for that
thread_idand check completed stages and unresolved external effects. Apply 2.4.4’s replay rules before allowing another attempt; a scheduler’s missing task record does not establish that an external action failed. - Have one recovery owner resubmit eligible work, invoke the graph with the same recorded thread identifier, and retain the resulting outcome in the admission record. Verify completion before closing the job. Reusing
job_idat the async endpoint retrieves status; it is not a command to restart an existing unfinished job.
Reconciliation joins application admission, job history, and graph progress; restarting the scheduler does not perform it.
Key distinctions and trade-offs
The mechanisms answer different questions:
| Concern | Mechanism | What it establishes | Remaining cost or risk |
|---|---|---|---|
| Work admission and scheduling | Dask scheduler and workers | Which available worker executes runnable work | Queue delay and finite worker capacity |
| HTTP response wait | sync_timeout | How long submission waits for a result | Background work can continue afterward |
| Useful execution budget | Application deadline and asyncio.timeout() | When a worker stops waiting on an operation | Remote effects can remain uncertain |
| Job history | --db_url | Persistent job metadata | Scheduler computation is separate |
| Agent progress | Persistent checkpointer, thread_id, durability mode | Saved graph state for recovery | Database latency and safe replay requirements |
Architecture / decision scenario
A legal research service receives 24 document-analysis jobs together. Its measured workload takes 120 seconds per job with eight execution slots, and users require completion within 300 seconds of acceptance. Once the 90-second extraction stage is checkpointed, recovery must preserve it rather than repeat it. Under these simplified, measured conditions, the third wave finishes at 360 seconds. Returning immediately with sync_timeout: 0 cannot repair that missed completion target.
The service owner admits 16 jobs for the five-minute commitment and offers the remaining eight a separately agreed later window. Its durable admission record stores job_id, the graph thread_id selected by the application, and the accepted deadline. Workers check it before analysis, and bound awaited calls with the remaining execution budget. The platform owner operates the external Dask scheduler; the API uses --scheduler_address, while job history uses persistent --db_url storage.
The tempting recovery design is to restart the scheduler and trust PostgreSQL job rows to resume everything. That fails the requirement to preserve completed analysis stages. The workflow team instead uses its own worker invocation of the PostgreSQL-checkpointed graph with durability="sync", accepting its measured write latency. The worker explicitly passes the configurable.thread_id from the admission record. After scheduler loss, the recovery owner reads unfinished admission and job records, expires overdue work, and resumes eligible graph threads from their saved state. The team tests scheduler loss separately from worker loss because Dask’s guarantees differ for the two.
Common exam traps
- Treating immediate asynchronous acceptance as a completion-time guarantee.
- Reading
sync_timeoutas an execution deadline orexpiry_secondsas queue expiration. - Applying local Dask worker flags to an external scheduler.
- Treating a persistent SQL job row as persistent scheduler state.
- Assuming worker recomputation makes arbitrary business actions safe to repeat.
- Assuming Toolkit 1.8’s unmodified
langgraph_wrapperpasses the thread identity required by a checkpointer.
What to remember
Identify the job, bound queue wait and execution, and persist the state needed for recovery. Dask schedules work, SQL records job metadata, and a persistent graph checkpointer preserves resumable agent progress. Response waiting, artifact retention, and execution deadlines remain separate controls.
Knowledge check
A contract-review service accepts jobs through /v1/workflow/async and gives each job a 10-minute budget from acceptance. During bursts, jobs wait in the Dask queue before a worker picks them up. Workers must reject overdue jobs and limit execution to the remaining budget. Which control implements this rule?
sync_timeout of 300 seconds on every submissionexpiry_seconds of 600 seconds on every submission--max_concurrent_jobs value on the API serversync_timeout only sets how long the submission waits before returning, and expiry_seconds how long output files are kept after completion; neither stops queued or running work. A deadline stored with the job covers queue wait and execution, lets a worker check the remaining time before expensive work and bound it with asyncio.timeout() , and passes to a rescheduled task instead of granting a fresh budget.A client submits a long job to /v1/workflow/async with its own job_id of inv-2291, and the connection drops before the response arrives. The client must not start the workflow a second time. What should it do?
job_id and cancel whichever job finishes secondjob_id inv-2291 and read the status it returnsjob_id so the server generates a UUIDexpiry_seconds to pass, then resubmit with inv-2291job_id already exists, the endpoint returns that job’s latest status instead of starting another run, so reusing the known identifier is safe after a lost response. A new or generated identifier submits another job, and expiry_seconds governs output files after completion. The identifier prevents a second submission; it does not make external actions inside the workflow run exactly once.A research service on NeMo Agent Toolkit 1.8 runs async jobs on an external Dask scheduler with a persistent --db_url. Each job’s graph uses a persistent checkpointer and spends 90 seconds on an extraction stage. Recovery must preserve that stage after execution has advanced to the next graph step. Which invocation supports recovery after scheduler and worker replacement?
langgraph_wrapper with durability="sync"durability="exit" and an explicit thread_iddurability="sync" and an explicit thread_idthread_id , and "sync" writes each checkpoint before the next step, while "exit" saves only when execution exits and loses the finished stage if the run dies mid-graph. The Toolkit 1.8 langgraph_wrapper invokes the graph without a RunnableConfig , so it passes no thread identity to the checkpointer.A team moves its NeMo Agent Toolkit async endpoint from a local Dask cluster to nat serve --scheduler_address pointing at a shared scheduler. To run more jobs at once it raises --max_concurrent_jobs from 10 to 40, but throughput does not change. What explains the result?
sync_timeout caps throughput--max_concurrent_jobs is documented as the maximum number of workers of the local Dask cluster — its name is misleading — and, like --dask_threads_per_worker , it is not used when scheduler_address is set. Capacity then belongs to the external cluster, which the team configures and measures there; sync_timeout changes only how long a submission waits for its response.References
- NVIDIA, NVIDIA NeMo Agent Toolkit API Server Endpoints
- NVIDIA, Running Existing LangGraph Agents in NVIDIA NeMo Agent Toolkit
- NVIDIA, LangGraph wrapper implementation, v1.8.0
- Dask, Scheduler State Machine
- Dask, Resilience
- LangChain, Checkpointers
- Python Software Foundation, Coroutines and Tasks