# 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:

```json
{
  "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 |