cleveragents-core/docs/reference/async_architecture.md

Async Architecture

Specification Note

docs/specification.md (line ~23362) states: "Plans are not queued. There is no worker queue or delayed execution model." However, the specification's plan-state tables already include a queued state described as "Waiting for compute/worker", and the specification discusses server-side execution for long-running plans.

Issue #312 (filed by the CTO) explicitly authorises this async infrastructure to bridge the gap for server-mode and long-running plan execution. A formal ADR or specification amendment should be raised to reconcile the "No Plan Queuing" clause with the async subsystem before the next major release.

Overview

The async execution infrastructure allows plan phases (Execute, Apply) to run as background jobs processed by a thread pool of workers. When async.enabled is True, plan phase transitions enqueue jobs instead of executing synchronously.

Execution Flow

1. User runs `agents plan execute <plan>`
2. If async.enabled:
     -> Job enqueued (status=queued, phase=execute)
     -> Poll loop dispatches job to ThreadPoolExecutor
     -> Worker thread picks up job (status=running)
     -> Worker thread executes plan phase
     -> Job completes (status=succeeded/failed/cancelled)
3. If not async.enabled:
     -> Synchronous execution (existing behaviour)

Job States

queued -> running -> succeeded
                  -> failed
                  -> cancelled

State	Description
queued	Job enqueued, waiting for a worker thread
running	Worker thread has picked up the job
succeeded	Job completed successfully
failed	Job failed (error or heartbeat timeout)
cancelled	Job cancelled by user or shutdown

Valid Transitions

• queued -> running (worker pickup)
• queued -> cancelled (user cancellation)
• running -> succeeded (completion)
• running -> failed (error or stuck detection)
• running -> cancelled (user/shutdown cancellation)

Terminal states (succeeded, failed, cancelled) have no outgoing transitions.

Configuration

Key	Type	Default	Description
CLEVERAGENTS_ASYNC_ENABLED	bool	False	Enable async job execution
CLEVERAGENTS_ASYNC_MAX_WORKERS	int	4	Max concurrent worker threads
CLEVERAGENTS_ASYNC_POLL_INTERVAL	float	1.0	Poll interval in seconds
CLEVERAGENTS_ASYNC_JOB_TIMEOUT	int	3600	Job timeout (seconds)
CLEVERAGENTS_ASYNC_JOB_TTL	int	86400	Completed job retention (sec)

Worker Lifecycle

Startup

1. Worker generates a unique worker_id (worker-<uuid8>)
2. A ThreadPoolExecutor(max_workers=N) is created for concurrent job execution
3. Signal handlers (SIGINT, SIGTERM) installed for graceful shutdown
4. Background polling thread started

Polling Loop

1. Query job store for jobs with status=queued
2. Check ThreadPoolExecutor capacity (max_workers - active_futures)
3. Dispatch up to capacity jobs to the thread pool:
   • Each job is submitted as a Future via ThreadPoolExecutor.submit
   • pickup_and_execute runs in its own worker thread
4. Sleep for poll_interval seconds

Graceful Shutdown

1. Signal received (SIGINT/SIGTERM) or stop() called
2. Shutdown event set — polling loop exits
3. All running jobs receive cancellation via their tokens
4. ThreadPoolExecutor.shutdown(wait=True) waits for in-flight jobs
5. Poll thread joined
6. Signal handlers restored

Cancellation

Cancellation uses a cooperative token model with a strict ownership contract to prevent race conditions:

1. CancellationToken created per job when pickup_and_execute starts
2. Token checked periodically during execution by the job executor
3. On cancellation request (cancel_job):
   • For running jobs: only the cancellation token is signalled. The worker thread that owns the job performs the RUNNING -> CANCELLED state transition, preventing a race where two threads attempt conflicting transitions on the same job object.
   • For queued jobs (not yet picked up): cancel_job directly transitions the job since no worker thread owns it.
4. Executor checks token and stops work
5. pickup_and_execute observes token.is_cancelled and transitions the job to cancelled, incrementing jobs_cancelled (not jobs_processed)

Why This Design

The original implementation had a race condition where cancel_job both signalled the token and directly transitioned the job to cancelled. When pickup_and_execute later checked the token and attempted the same transition, it raised InvalidJobTransitionError because the job was already terminal. The fix ensures only one code path (the owning thread) performs state transitions for running jobs.

Stuck Job Detection

Jobs are considered stuck when:

• Status is running
• last_heartbeat is older than job_timeout seconds

Stuck jobs are automatically marked as failed (with error_message set to "Heartbeat timeout — job stuck") by the worker's detect_stuck_jobs() routine.

Job Cleanup / Retention

Completed jobs (succeeded, failed, cancelled) are retained for job_ttl seconds (default: 86400 = 24 hours). The cleanup_completed_jobs() routine delegates to InMemoryJobStore.remove_expired() which performs a single-pass, single-lock removal to avoid N+1 lock acquisition overhead.

Error Audit Trail

When a job fails with an exception, the error_message field on the AsyncJob model captures a string representation of the error ("ExceptionType: message"). This enables post-mortem analysis without requiring log file searches. The error_message field is:

• NULL for non-failed jobs
• Included in as_cli_dict() output when set
• Persisted in the async_jobs.error_message database column

Payload Serialization

Job payloads are JSON-serialized with a schema_version field for forward compatibility:

{
  "schema_version": 1,
  "plan_id": "01HXYZ...",
  "phase": "execute",
  "extra": {}
}

The schema_version allows future changes to the payload format without breaking existing workers.

Timestamps

All timestamps in the async subsystem use timezone-aware UTC datetimes (datetime.now(UTC)) for consistency with the rest of the codebase and to avoid comparison failures when mixing naive and aware datetimes across distributed systems.

Database Table

The async_jobs table stores all job metadata:

Column	Type	Description
job_id	String(26)	ULID primary key
plan_id	String(26)	Logical plan reference (no FK)
phase	String(20)	execute or apply
status	String(20)	queued/running/succeeded/etc
payload_json	Text	Serialized payload
created_at	String(30)	ISO-8601 timestamp
started_at	String(30)	Worker pickup time
finished_at	String(30)	Completion time
worker_id	String(255)	Assigned worker ID
last_heartbeat	String(30)	Last worker heartbeat
schema_version	Integer	Payload schema version
error_message	Text (NULL)	Error details for failed jobs

The plan_id column is a logical reference to v3_plans without a foreign key constraint, because jobs may outlive their plans (e.g., during cleanup windows).

Indexes

• ix_async_jobs_plan_id — plan lookups
• ix_async_jobs_status — queue polling
• ix_async_jobs_worker_id — worker-scoped queries
• ix_async_jobs_created_at — age-based ordering

CHECK Constraints

• ck_async_jobs_status — status IN ('queued','running','succeeded','failed','cancelled')
• ck_async_jobs_phase — phase IN ('execute','apply')

Health Report

Worker health is surfaced in agents diagnostics:

Async workers: enabled, max_workers=4, poll_interval=1.0s

The health report includes:

• Worker ID and status
• Jobs processed / failed / cancelled counts
• Last heartbeat timestamp (UTC)
• Queue depth (queued/running/total) via atomic snapshot

The queue-depth metrics are obtained via InMemoryJobStore.snapshot_counts() under a single lock acquisition to guarantee consistency.

Review Fixes Applied

The following issues were identified during code review and fixed:

ID	Category	Fix Summary
B1	Bug (High)	Race condition in cancel_job + pickup_and_execute — cancel_job now only signals the token for running jobs
B2	Bug (Med)	record_job_completed() was called on cancelled jobs — now uses separate record_job_cancelled() counter
B3	Bug (Med)	Serial execution despite max_workers — replaced with ThreadPoolExecutor for true concurrency
B4	Bug (Low)	datetime.now() without timezone — switched to datetime.now(UTC) everywhere
B5	Bug (Low)	Misleading FK comment on plan_id — corrected to "logical reference (no FK)"
T1	Test (High)	No DB-level tests — added SQLAlchemy round-trip and CHECK constraint Behave scenarios
T2	Test (Med)	No concurrent execution test — added multi-threaded Behave scenario
T3	Test (Med)	Timing-based flaky sleeps — replaced with deterministic sync (thread join, clock-advance loop)
T4	Test (Low)	Overly loose error assertion — tightened to InvalidJobTransitionError only
T5	Test (Low)	Robot tests lacked assertion granularity — added per-check output and traceback on failure
P1	Perf (Low)	get_health_report acquired 3 separate locks — replaced with atomic snapshot_counts()
P2	Perf (Low)	cleanup_completed_jobs used N+1 locking — replaced with single-pass remove_expired()
SEC1	Security	No error data persisted — added error_message field to AsyncJob and AsyncJobModel
S1	Spec	"No Plan Queuing" contradiction — documented reconciliation note in this file