Facio's Async Processing Discipline: How AI Agents Handle Long-Running Tasks Without Blocking Conversations, Losing State, or Breaking the Customer Experience
AI agents do more than respond in seconds. The agents process long-running tasks: data analysis jobs, file transformations, batch operations, model training runs, multi-stage workflows, third-party integrations that take hours. The naive approach blocks the conversation until the task completes. The customer waits. The connection times out. The customer refreshes. The state is lost. The work is gone.
Facio's async processing discipline gives AI agents structured mechanisms to handle long-running tasks without blocking the customer experience. The agent submits the task and returns immediately. The agent polls for status. The agent notifies when complete. The agent recovers from interruptions. The customer gets a great experience even for hours-long work.
Here's how the discipline works, what async processing it includes, and why async processing discipline is what makes AI agents capable of doing real, long-running work in production without frustrating customers.
The Async Processing Reality
Production AI agents face an async processing problem:
Problem 1: Blocking the conversation. The agent starts a long-running task. The customer waits. The connection times out. The customer closes the chat. The work is interrupted.
Problem 2: Lost state on interruption. The agent processes a task across multiple turns. The customer's network drops. The session ends. The state is lost. The work starts over.
Problem 3: No progress indication. The agent starts a task. The customer has no idea how long it will take. The customer assumes the agent is broken. The customer gives up.
Problem 4: Resource exhaustion. The agent holds resources while waiting for the task. The agent can't take other work. The agent is blocked. The throughput is limited.
Problem 5: No retry mechanism. The long-running task fails. The agent has no way to retry. The work is lost. The customer has to start over.
Problem 6: No completion notification. The task completes in the background. The customer has no idea. The customer moves on. The result is unused.
The naive approach — block and wait — fails every test. The team ends up with customers who can't use agents for real work.
The Async Processing Discipline
Facio's async processing discipline has five pillars. Each addresses a different aspect of handling long-running tasks well.
Pillar 1: Task Submission (Start Without Blocking)
Long-running tasks are submitted without blocking:
# Task submission
task_submission = {
"submission_patterns": {
"fire_and_forget": {
"description": "Submit task and return immediately",
"use_case": "tasks_where_customer_does_NOT_need_to_wait",
"implementation": "submit_to_queue_AND_return_immediately",
"customer_response": "Got it, I'm working on that.",
},
"background_with_notification": {
"description": "Submit task, return immediately, notify on completion",
"use_case": "tasks_where_customer_needs_result_later",
"implementation": "submit_to_queue_AND_register_notification_handler",
"customer_response": "I'll let you know when this is done.",
},
"polled_status": {
"description": "Submit task, return immediately, allow status polling",
"use_case": "tasks_where_customer_may_check_progress",
"implementation": "submit_to_queue_AND_provide_status_endpoint",
"customer_response": "Started! You can check status anytime.",
},
"streaming_progress": {
"description": "Submit task, return immediately, stream progress updates",
"use_case": "tasks_where_customer_wants_live_progress",
"implementation": "submit_to_queue_AND_stream_progress_events",
"customer_response": "Started! I'll show you progress as it happens.",
},
},
"submission_metadata": {
"task_id": "task-abc-123",
"task_type": "data_analysis",
"submitted_at": "2026-08-07T10:00:00Z",
"expected_duration_seconds": 3600,
"tenant_id": "tenant-acme-corp",
"submitted_by": "agent-customer-support-007",
"priority": "normal",
"notification_channel": "customer-chat-001",
},
"submission_validation": {
"task_type_allowed": "verify_task_type_in_agent_capabilities",
"arguments_validated": "verify_arguments_against_schema",
"priority_validated": "verify_priority_within_allowed_range",
"tenant_validated": "verify_tenant_allowed_to_submit_this_task",
},
}
def submit_long_running_task(task_type, arguments, notification_config):
task_id = generate_task_id()
task_request = {
"task_id": task_id,
"task_type": task_type,
"arguments": arguments,
"submitted_at": time.now(),
"tenant_context": get_tenant_context(),
"notification_config": notification_config,
}
validate_submission(task_request)
task_queue.enqueue(task_request)
register_notification_handler(task_id, notification_config)
return {"task_id": task_id, "status": "submitted"}
The task submission is non-blocking. The agent returns immediately.
Pillar 2: Task State Persistence (Survive Interruptions)
Long-running task state is persisted:
# Task state persistence
task_state_persistence = {
"state_structure": {
"task_id": "task-abc-123",
"task_type": "data_analysis",
"current_step": "step-3-of-10",
"completed_steps": ["step-1", "step-2"],
"partial_outputs": {
"step-1": {"records_processed": 1000, "errors": 0},
"step-2": {"records_processed": 5000, "errors": 2},
},
"execution_state": {
"started_at": "2026-08-07T10:00:00Z",
"last_checkpoint_at": "2026-08-07T10:30:00Z",
"duration_so_far_seconds": 1800,
"estimated_remaining_seconds": 1800,
},
"interruption_history": [
{"at": "2026-08-07T10:15:00Z", "reason": "worker_restart", "resumed_at": "2026-08-07T10:16:00Z"},
],
},
"persistence_strategies": {
"checkpoint_after_each_step": "save_state_after_each_completed_step",
"periodic_checkpoint": "save_state_every_30_seconds",
"incremental_state": "save_only_state_changes_since_last_checkpoint",
"full_state_snapshot": "save_complete_state_at_intervals",
},
"state_storage": {
"primary_storage": "task_state_database_with_durability",
"backup_storage": "replicated_for_disaster_recovery",
"state_encryption": "state_encrypted_at_rest",
"state_retention": "state_retained_for_30_days_after_completion",
},
}
def persist_task_state(task_id, state):
task_state_store.put(task_id, state)
checkpoint_state(task_id, state)
return state
The state persistence ensures tasks survive interruptions. The work resumes from the latest checkpoint.
Pillar 3: Progress Indication (Keep Customer Informed)
Long-running tasks provide progress updates:
# Progress indication
progress_indication = {
"progress_types": {
"percentage_progress": {
"description": "Percentage of task complete",
"example": "47% complete (estimated 18 minutes remaining)",
"use_case": "long_tasks_with_clear_steps",
},
"step_progress": {
"description": "Which step is being executed",
"example": "Currently processing step 5 of 10: validating customer records",
"use_case": "tasks_with_named_steps",
},
"record_progress": {
"description": "Records processed vs total",
"example": "Processed 50,000 of 100,000 records",
"use_case": "batch_processing_tasks",
},
"bytes_progress": {
"description": "Bytes processed vs total",
"example": "Uploaded 750 MB of 2 GB",
"use_case": "data_upload_or_download_tasks",
},
"indeterminate_progress": {
"description": "Progress not measurable",
"example": "Still processing, please wait",
"use_case": "tasks_where_progress_NOT_measurable",
},
},
"progress_update_frequency": {
"high_progress_tasks": "update_every_5_seconds",
"medium_progress_tasks": "update_every_30_seconds",
"low_progress_tasks": "update_every_5_minutes",
"long_silent_tasks": "send_at_least_one_update_every_15_minutes",
},
"progress_communication": {
"to_customer": "polite_update_with_estimated_remaining_time",
"to_log": "detailed_progress_with_timing_data",
"to_dashboard": "real_time_progress_for_operators",
},
}
def emit_progress_update(task_id, progress_data):
progress_message = format_progress_message(progress_data)
if should_notify_customer(progress_data):
send_progress_to_customer(task_id, progress_message)
update_progress_log(task_id, progress_data)
update_dashboard(task_id, progress_data)
The progress indication keeps the customer informed. The customer knows what's happening.
Pillar 4: Task Recovery (Resume After Failure)
Long-running tasks can recover from failures:
# Task recovery
task_recovery = {
"failure_types": {
"transient_failure": {
"description": "Temporary failure that's likely to resolve",
"examples": ["network_timeout", "rate_limit", "service_503"],
"recovery": "retry_with_exponential_backoff",
},
"worker_failure": {
"description": "Worker process crashed",
"examples": ["oom_killed", "worker_process_died"],
"recovery": "restart_on_different_worker_with_state_restore",
},
"dependency_failure": {
"description": "External dependency failed",
"examples": ["external_api_down", "database_unavailable"],
"recovery": "wait_for_dependency_recovery_AND_retry",
},
"permanent_failure": {
"description": "Failure that won't resolve automatically",
"examples": ["invalid_arguments", "data_corruption"],
"recovery": "mark_task_failed_AND_notify_customer",
},
},
"recovery_strategies": {
"automatic_retry": {
"description": "System retries automatically",
"implementation": "exponential_backoff_up_to_max_retries",
"use_case": "transient_failures_likely_to_resolve",
},
"worker_replacement": {
"description": "Different worker takes over task",
"implementation": "task_requeued_for_different_worker",
"use_case": "specific_worker_problem",
},
"dependency_wait": {
"description": "Wait for dependency to recover",
"implementation": "task_paused_AND_polled_until_dependency_available",
"use_case": "external_dependency_issues",
},
"manual_intervention": {
"description": "Human handles the failure",
"implementation": "task_paused_AND_human_notified",
"use_case": "complex_failures_needing_judgment",
},
},
"recovery_state": {
"retry_count": "tracked_per_task",
"last_failure": "logged_with_details",
"recovery_history": "all_recovery_attempts_recorded",
"final_outcome": "either_success_or_failure_with_reason",
},
}
def recover_task(task_id, failure_context):
task_state = load_task_state(task_id)
if failure_context["type"] == "transient_failure":
return retry_task_with_backoff(task_id, task_state, failure_context)
elif failure_context["type"] == "worker_failure":
return requeue_task_for_different_worker(task_id, task_state)
elif failure_context["type"] == "dependency_failure":
return pause_task_AND_wait_for_dependency(task_id, task_state)
elif failure_context["type"] == "permanent_failure":
return mark_task_failed_AND_notify(task_id, task_state, failure_context)
The task recovery ensures long work survives failures. The work resumes from where it was.
Pillar 5: Completion Notification (Tell Customer When Done)
Long-running tasks notify on completion:
# Completion notification
completion_notification = {
"notification_types": {
"success_notification": {
"description": "Task succeeded",
"example": "Your analysis is complete. Here are the results...",
"delivery": "via_notification_channel_registered_at_submission",
},
"failure_notification": {
"description": "Task failed",
"example": "I couldn't complete the analysis. Here's what went wrong...",
"delivery": "via_notification_channel_with_failure_details",
},
"partial_notification": {
"description": "Task partially succeeded",
"example": "I completed 8 of 10 steps. The last 2 steps failed...",
"delivery": "via_notification_channel_with_partial_results",
},
"timeout_notification": {
"description": "Task exceeded time limit",
"example": "The task is taking longer than expected. I'll continue processing...",
"delivery": "via_notification_channel_with_timeout_status",
},
},
"notification_delivery": {
"channel_options": ["websocket", "email", "sms", "push_notification", "polling"],
"delivery_retry": "retry_notification_if_delivery_fails",
"delivery_audit": "every_notification_delivery_logged",
"customer_acknowledgment": "verify_customer_received_notification",
},
"notification_content": {
"result_summary": "concise_summary_of_outcome",
"result_data": "relevant_data_or_links_to_data",
"next_steps": "what_customer_should_do_next",
"support_contact": "how_to_get_help_if_needed",
},
}
def notify_task_completion(task_id, outcome):
notification = format_completion_notification(task_id, outcome)
delivery_channel = load_notification_channel(task_id)
deliver_notification(delivery_channel, notification)
log_notification_delivery(task_id, outcome, delivery_channel)
if outcome["status"] == "success":
mark_task_complete(task_id)
else:
mark_task_failed(task_id, outcome["reason"])
The completion notification tells the customer when the work is done. The customer gets the result.
The Async Processing Patterns
Several patterns emerge from disciplined async processing.
Pattern 1: Async Task Queue Architecture
Long-running tasks use a queue architecture:
# Async task queue architecture
async_queue_architecture = {
"queue_components": {
"submission_api": {
"description": "Endpoint where tasks are submitted",
"implementation": "HTTP_endpoint_with_validation_AND_acknowledgment",
"responsibility": "accept_tasks_AND_persist_them",
},
"task_queue": {
"description": "Queue holding tasks waiting for execution",
"implementation": "durable_queue_with_priority_AND_fairness",
"responsibility": "store_tasks_until_worker_picks_up",
},
"task_workers": {
"description": "Process that executes tasks",
"implementation": "horizontal_pool_of_workers_pulling_from_queue",
"responsibility": "execute_tasks_AND_report_progress",
},
"result_storage": {
"description": "Store task results",
"implementation": "durable_storage_for_results_AND_metadata",
"responsibility": "preserve_results_for_customer_access",
},
"notification_system": {
"description": "Notify customers of completion",
"implementation": "notification_dispatcher_with_multiple_channels",
"responsibility": "deliver_results_to_customers",
},
},
"queue_properties": {
"durability": "tasks_survive_system_restarts",
"priority": "high_priority_tasks_processed_first",
"fairness": "all_tenants_get_fair_share",
"observability": "queue_state_observable_for_operators",
"scalability": "queue_scales_with_demand",
},
}
def submit_to_async_queue(task_request):
validated = validate_submission(task_request)
task_id = task_queue.enqueue(validated)
result_storage.initialize(task_id)
notification_system.register(task_id, validated["notification_config"])
return {"task_id": task_id, "status": "queued"}
The queue architecture handles scale. The tasks flow through the system.
Pattern 2: Worker Pool and Scaling
Workers are pooled and scaled:
# Worker pool and scaling
worker_pool = {
"pool_configuration": {
"min_workers": "minimum_workers_for_baseload",
"max_workers": "maximum_workers_to_prevent_overload",
"auto_scaling": "scale_workers_based_on_queue_depth",
"worker_specialization": "specialized_workers_for_different_task_types",
},
"worker_lifecycle": {
"startup": "worker_registers_AND_starts_polling_queue",
"execution": "worker_pulls_task_AND_executes",
"heartbeat": "worker_reports_heartbeat_to_orchestrator",
"shutdown": "worker_finishes_current_task_AND_exits",
"failure_handling": "worker_death_detected_AND_task_requeued",
},
"scaling_strategies": {
"scale_up_on_queue_growth": "add_workers_when_queue_depth_grows",
"scale_down_on_idle": "remove_workers_when_idle_to_save_costs",
"scale_for_task_type": "specialized_workers_for_heavy_task_types",
"predictive_scaling": "scale_based_on_historical_patterns",
},
"worker_health": {
"health_check": "periodic_health_check_for_each_worker",
"heartbeat_monitoring": "detect_stuck_workers_via_missing_heartbeat",
"task_timeout": "task_killed_if_worker_unresponsive",
"automatic_replacement": "dead_workers_replaced_automatically",
},
}
def manage_worker_pool(queue_depth, task_distribution):
desired_workers = calculate_desired_workers(queue_depth, task_distribution)
current_workers = get_current_worker_count()
if desired_workers > current_workers:
scale_up_workers(desired_workers - current_workers)
elif desired_workers < current_workers:
scale_down_workers(current_workers - desired_workers)
The worker pool scales with demand. The system handles load.
Pattern 3: Long-Running Task Patterns
Common long-running patterns are reusable:
# Long-running task patterns
long_running_patterns = {
"batch_processing": {
"description": "Process many records efficiently",
"implementation": "chunked_processing_with_progress_reporting",
"example": "process_1M_records_in_batches_of_1000",
"progress_indicator": "records_processed / total_records",
},
"data_pipeline": {
"description": "Multi-stage data transformation",
"implementation": "stages_with_dependencies_AND_state_persistence",
"example": "extract -> transform -> load -> validate",
"progress_indicator": "current_stage / total_stages",
},
"model_training": {
"description": "Train machine learning model",
"implementation": "iterative_training_with_checkpoint_saves",
"example": "train_model_for_N_epochs_with_validation",
"progress_indicator": "epoch / total_epochs_AND_loss_metrics",
},
"external_integration": {
"description": "Wait for external system to complete work",
"implementation": "poll_external_status_AND_wait",
"example": "wait_for_third_party_job_completion",
"progress_indicator": "external_status_polling",
},
"report_generation": {
"description": "Generate complex report",
"implementation": "stages_with_caching_AND_progress_reporting",
"example": "gather_data -> analyze -> format -> render",
"progress_indicator": "stage / total_stages",
},
}
def execute_batch_processing_task(task_request):
records = task_request["records"]
chunk_size = task_request["chunk_size"]
for i in range(0, len(records), chunk_size):
chunk = records[i:i+chunk_size]
process_chunk(chunk)
update_progress(task_request["task_id"], i + len(chunk), len(records))
The patterns provide proven approaches. The work uses established techniques.
Pattern 4: Task Observability
Long-running tasks are observable:
# Task observability
task_observability = {
"metrics": {
"tasks_submitted_per_hour": "tracks_submission_volume",
"tasks_completed_per_hour": "tracks_throughput",
"tasks_failed_per_hour": "tracks_failure_rate",
"task_duration_distribution": "tracks_p50/p95/p99_durations",
"queue_depth": "tracks_pending_tasks",
"worker_utilization": "tracks_worker_efficiency",
"task_retry_rate": "tracks_how_often_retries_needed",
"notification_delivery_rate": "tracks_notification_success",
},
"alerting": [
{"level": "critical", "message": "Task failure rate spike"},
{"level": "warning", "message": "Task queue depth growing"},
{"level": "warning", "message": "Task duration exceeding SLA"},
{"level": "warning", "message": "Worker pool undersized for load"},
{"level": "info", "message": "New long-running task type added"},
],
"dashboard": {
"active_tasks": "shows_currently_running",
"queue_overview": "shows_pending_AND_processing",
"worker_health": "shows_worker_status",
"task_history": "shows_completed_tasks",
},
"task_audit": {
"every_task_recorded": "every_submission_AND_outcome_logged",
"state_history": "every_state_change_recorded",
"retry_history": "every_retry_recorded",
"audit_searchable": "logs_searchable_for_compliance",
},
}
def emit_task_observability(task_id, task_outcome):
metrics = collect_all_task_metrics(task_id, task_outcome)
publish_to_dashboard(metrics)
check_alert_thresholds(metrics)
update_audit_log(task_id, task_outcome)
The observability surfaces task patterns. The team understands long-running work.
Pattern 5: Customer-Facing Communication
Customers get clear communication about long-running tasks:
# Customer-facing communication
customer_communication = {
"communication_patterns": {
"submission_acknowledgment": {
"description": "Acknowledge the task was submitted",
"example": "I've started processing your data. This usually takes about 30 minutes.",
"timing": "immediately_at_submission",
},
"progress_updates": {
"description": "Periodic progress updates",
"example": "Progress: 47% complete, about 15 minutes remaining.",
"timing": "at_declared_intervals",
},
"milestone_updates": {
"description": "Updates at significant milestones",
"example": "Halfway done! I'm now starting the analysis phase.",
"timing": "at_major_progress_milestones",
},
"completion_notification": {
"description": "Notify when task is done",
"example": "Done! Here are your results: ...",
"timing": "at_task_completion",
},
"delay_notification": {
"description": "Notify if task takes longer than expected",
"example": "This is taking a bit longer than usual. I'm still working on it.",
"timing": "if_task_exceeds_expected_duration",
},
},
"communication_tone": {
"informative": "communicate_progress_AND_status",
"reassuring": "reassure_customer_task_is_progressing",
"honest": "communicate_failures_AND_delays_honestly",
"actionable": "tell_customer_what_to_do_next",
},
}
def communicate_async_task_state(task_id, communication_event):
task_state = load_task_state(task_id)
customer_context = get_customer_context(task_id)
message = generate_communication(communication_event, task_state, customer_context)
send_to_customer(customer_context, message)
log_communication(task_id, communication_event, message)
The communication keeps the customer informed. The customer experience is good even for long work.
The Async Processing Discipline Doesn't Do
Honest limitations:
- It can't make tasks instant. Long tasks are still long. The discipline manages them better.
- It adds complexity. Queue, workers, state persistence require infrastructure. The discipline requires engineering.
- It depends on notification channels. If channels are down, notifications fail. The discipline requires redundancy.
- It can mask task failures. Automatic retry can hide real problems. The discipline requires monitoring.
- It requires careful state design. Bad state design causes work loss. The discipline requires careful engineering.
The Async Processing Discipline as Operational Practice
Async processing discipline is operational practice:
Queue monitoring. The team monitors queue depth. They scale workers proactively.
Task review. The team reviews task patterns. They identify slow tasks.
Notification testing. The team tests notification delivery. They verify customers receive notifications.
State validation. The team validates state persistence. They ensure recovery works.
The practice is what makes the discipline sustainable. Without it, async work degrades. With it, async work is reliable.
The Compound Effect of Async Processing Discipline
Async processing discipline compounds:
- Better customer experience. Customers aren't blocked. The experience is good.
- Higher task reliability. Tasks survive failures. The work is done.
- Better resource utilization. Workers are pooled efficiently. The costs are managed.
- Higher team confidence. The team trusts async work. The deployment includes real work.
- Better scalability. The system scales with demand. The growth is supported.
The undisciplined approach has the opposite trajectory. Blocked customers, lost work, poor resource utilization, low team confidence, limited scalability.
Bottom Line
AI agents handle long-running tasks. Without async processing discipline, the tasks block customers. With async processing discipline, the tasks don't block.
Facio's async processing discipline provides task submission, state persistence, progress indication, task recovery, and completion notification. The discipline makes AI agents capable of real work.
The agent without async processing is limited to quick responses. The agent with it handles real work. The team trusts the capable one. The customers trust the capable one.
Because AI agents in production do long-running work. The question is whether the work blocks or scales. The async processing discipline is what makes the answer scales.
See the async processing documentation for task submission patterns, state persistence schemas, and worker pool configuration.