CHANGELOG.md

@@ -37,6 +37,19 @@
   `summary_strategy`). `context show` displays ACMS pipeline configuration alongside context
   policy. Includes 28 Behave BDD scenarios for wiring coverage, updated Robot Framework
   integration tests, and reference documentation. (#499)
+- Added async command execution infrastructure allowing plan phases (Execute,
+  Apply) to run as background jobs processed by a thread pool of workers.
+  `AsyncJob` Pydantic v2 domain model with ULID primary key, status state machine
+  (`queued -> running -> succeeded/failed/cancelled`), and payload schema versioning.
+  `AsyncWorker` service with `ThreadPoolExecutor`-backed concurrent execution,
+  race-safe cancellation contract, stuck job detection, and job cleanup.
+  `InMemoryJobStore` with atomic `snapshot_counts()` and single-pass `remove_expired()`.
+  Plan lifecycle service wired to enqueue jobs when `async.enabled` is True.
+  Error messages redacted via `shared/redaction.py` before persisting to the audit
+  trail. Configurable via `async.enabled`, `async.max_workers`, `async.poll_interval`,
+  `async.job_timeout`, `async.job_ttl`. Includes `AsyncJobModel` SQLAlchemy model,
+  Alembic migration, Behave BDD scenarios, Robot Framework integration tests, ASV
+  benchmarks, and `docs/reference/async_architecture.md`. (#312)
 - Added hot/warm/cold context tiers with `ContextTier`, `ActorRole`, `TieredFragment`,
   `TierBudget`, `ActorContextView`, `TierMetrics`, and `ScopedBackendView` models.
   `ContextTierService` provides store/get, promotion/demotion with cold-tier summarisation

alembic/versions/m6_003_async_jobs_table.py

@@ -0,0 +1,58 @@
+"""Add async_jobs table for background job execution.
+
+Revision ID: m6_003_async_jobs_table
+Revises: m6_002_merge_safety_and_checkpoint
+Create Date: 2026-03-03 00:00:00
+
+"""
+
+from collections.abc import Sequence
+
+from alembic import op
+import sqlalchemy as sa
+
+# revision identifiers, used by Alembic.
+revision: str = "m6_003_async_jobs_table"
+down_revision: str | Sequence[str] | None = "m6_002_merge_safety_and_checkpoint"
+branch_labels: str | Sequence[str] | None = None
+depends_on: str | Sequence[str] | None = None
+
+
+def upgrade() -> None:
+    """Create the async_jobs table."""
+    op.create_table(
+        "async_jobs",
+        sa.Column("job_id", sa.String(26), primary_key=True),
+        sa.Column("plan_id", sa.String(26), nullable=False),
+        sa.Column("phase", sa.String(20), nullable=False),
+        sa.Column("status", sa.String(20), nullable=False, server_default="queued"),
+        sa.Column("payload_json", sa.Text(), nullable=False, server_default="{}"),
+        sa.Column("created_at", sa.String(30), nullable=False),
+        sa.Column("started_at", sa.String(30), nullable=True),
+        sa.Column("finished_at", sa.String(30), nullable=True),
+        sa.Column("worker_id", sa.String(255), nullable=True),
+        sa.Column("last_heartbeat", sa.String(30), nullable=True),
+        sa.Column("schema_version", sa.Integer(), nullable=False, server_default="1"),
+        sa.Column("error_message", sa.Text(), nullable=True),
+        sa.CheckConstraint(
+            "status IN ('queued', 'running', 'succeeded', 'failed', 'cancelled')",
+            name="ck_async_jobs_status",
+        ),
+        sa.CheckConstraint(
+            "phase IN ('execute', 'apply')",
+            name="ck_async_jobs_phase",
+        ),
+    )
+    op.create_index("ix_async_jobs_plan_id", "async_jobs", ["plan_id"])
+    op.create_index("ix_async_jobs_status", "async_jobs", ["status"])
+    op.create_index("ix_async_jobs_worker_id", "async_jobs", ["worker_id"])
+    op.create_index("ix_async_jobs_created_at", "async_jobs", ["created_at"])
+
+
+def downgrade() -> None:
+    """Drop the async_jobs table."""
+    op.drop_index("ix_async_jobs_created_at", table_name="async_jobs")
+    op.drop_index("ix_async_jobs_worker_id", table_name="async_jobs")
+    op.drop_index("ix_async_jobs_status", table_name="async_jobs")
+    op.drop_index("ix_async_jobs_plan_id", table_name="async_jobs")
+    op.drop_table("async_jobs")

benchmarks/async_execution_bench.py

@@ -0,0 +1,338 @@
+"""ASV benchmarks for async command execution and workers.
+
+Measures the performance of:
+- AsyncJob model creation and validation overhead
+- AsyncJob state machine transitions
+- InMemoryJobStore CRUD operations
+- AsyncWorkerConfig instantiation
+- WorkerHealthReport lifecycle
+- CancellationToken operations
+- Payload serialization/deserialization
+- AsyncWorker job pickup and execution
+"""
+
+from __future__ import annotations
+
+import importlib
+import sys
+from pathlib import Path
+
+# Ensure the local *source* tree is importable even when ASV has an
+# older build of the package installed.
+_SRC = str(Path(__file__).resolve().parents[1] / "src")
+if _SRC not in sys.path:
+    sys.path.insert(0, _SRC)
+
+# Force-reload so ASV picks up the source tree version.
+import cleveragents  # noqa: E402
+
+importlib.reload(cleveragents)
+
+from cleveragents.application.services.async_worker import (  # noqa: E402
+    AsyncWorker,
+    AsyncWorkerConfig,
+    CancellationToken,
+    InMemoryJobStore,
+    WorkerHealthReport,
+)
+from cleveragents.domain.models.core.async_job import (  # noqa: E402
+    AsyncJob,
+    AsyncJobStatus,
+    InvalidJobTransitionError,
+    can_transition_job,
+    deserialize_job_payload,
+    serialize_job_payload,
+)
+
+# A valid ULID for benchmarks
+_ULID_A = "01ARZ3NDEKTSV4RRFFQ69G5FAV"
+_ULID_B = "01ARZ3NDEKTSV4RRFFQ69G5FAW"
+
+
+# ---------------------------------------------------------------------------
+# AsyncJob model creation benchmarks
+# ---------------------------------------------------------------------------
+
+
+class AsyncJobCreationSuite:
+    """Benchmark AsyncJob model creation overhead."""
+
+    timeout = 60
+
+    def time_create_minimal_job(self) -> None:
+        AsyncJob(job_id=_ULID_A, plan_id=_ULID_B, phase="execute")
+
+    def time_create_job_with_payload(self) -> None:
+        AsyncJob(
+            job_id=_ULID_A,
+            plan_id=_ULID_B,
+            phase="apply",
+            payload_json='{"key": "value", "count": 42}',
+        )
+
+    def time_create_job_all_fields(self) -> None:
+        AsyncJob(
+            job_id=_ULID_A,
+            plan_id=_ULID_B,
+            phase="execute",
+            status=AsyncJobStatus.QUEUED,
+            payload_json='{"schema_version": 1}',
+            schema_version=1,
+        )
+
+
+# ---------------------------------------------------------------------------
+# AsyncJob state machine benchmarks
+# ---------------------------------------------------------------------------
+
+
+class AsyncJobTransitionSuite:
+    """Benchmark AsyncJob state machine transition overhead."""
+
+    timeout = 60
+
+    def setup(self) -> None:
+        self._job_id = _ULID_A
+        self._plan_id = _ULID_B
+
+    def time_transition_queued_to_running(self) -> None:
+        job = AsyncJob(job_id=self._job_id, plan_id=self._plan_id, phase="execute")
+        job.mark_running("worker-bench")
+
+    def time_transition_running_to_succeeded(self) -> None:
+        job = AsyncJob(job_id=self._job_id, plan_id=self._plan_id, phase="execute")
+        job.mark_running("worker-bench")
+        job.mark_succeeded()
+
+    def time_transition_running_to_failed(self) -> None:
+        job = AsyncJob(job_id=self._job_id, plan_id=self._plan_id, phase="execute")
+        job.mark_running("worker-bench")
+        job.mark_failed()
+
+    def time_transition_queued_to_cancelled(self) -> None:
+        job = AsyncJob(job_id=self._job_id, plan_id=self._plan_id, phase="execute")
+        job.mark_cancelled()
+
+    def time_can_transition_check(self) -> None:
+        can_transition_job(AsyncJobStatus.QUEUED, AsyncJobStatus.RUNNING)
+
+    def time_full_lifecycle(self) -> None:
+        job = AsyncJob(job_id=self._job_id, plan_id=self._plan_id, phase="execute")
+        job.mark_running("worker-bench")
+        job.record_heartbeat()
+        job.mark_succeeded()
+
+    def time_as_cli_dict(self) -> None:
+        job = AsyncJob(job_id=self._job_id, plan_id=self._plan_id, phase="execute")
+        job.as_cli_dict()
+
+    def time_is_terminal_check(self) -> None:
+        job = AsyncJob(
+            job_id=self._job_id,
+            plan_id=self._plan_id,
+            phase="execute",
+            status=AsyncJobStatus.QUEUED,
+        )
+        job.is_terminal
+
+
+# ---------------------------------------------------------------------------
+# Payload serialization benchmarks
+# ---------------------------------------------------------------------------
+
+
+class PayloadSerializationSuite:
+    """Benchmark payload serialization/deserialization overhead."""
+
+    timeout = 60
+
+    def time_serialize_minimal(self) -> None:
+        serialize_job_payload(_ULID_A, "execute")
+
+    def time_serialize_with_extra(self) -> None:
+        serialize_job_payload(
+            _ULID_A, "apply", extra={"steps": ["s1", "s2"], "retries": 3}
+        )
+
+    def time_deserialize_minimal(self) -> None:
+        deserialize_job_payload('{"plan_id": "test", "phase": "execute"}')
+
+    def time_deserialize_with_extra(self) -> None:
+        deserialize_job_payload(
+            '{"plan_id": "test", "phase": "apply", "extra": {"steps": ["s1"]}}'
+        )
+
+    def time_roundtrip(self) -> None:
+        payload = serialize_job_payload(_ULID_A, "execute", extra={"key": "val"})
+        deserialize_job_payload(payload)
+
+
+# ---------------------------------------------------------------------------
+# InMemoryJobStore benchmarks
+# ---------------------------------------------------------------------------
+
+
+class InMemoryJobStoreSuite:
+    """Benchmark InMemoryJobStore CRUD operations."""
+
+    timeout = 60
+
+    def setup(self) -> None:
+        self.store = InMemoryJobStore()
+        # Pre-populate with 100 jobs
+        for i in range(100):
+            ulid = f"01ARZ3NDEKTSV4RRFFQ69G{i:04d}"[:26]
+            job = AsyncJob(job_id=ulid, plan_id=_ULID_B, phase="execute")
+            self.store.add(job)
+
+    def time_add_job(self) -> None:
+        store = InMemoryJobStore()
+        job = AsyncJob(job_id=_ULID_A, plan_id=_ULID_B, phase="execute")
+        store.add(job)
+
+    def time_get_job(self) -> None:
+        self.store.get(_ULID_A[:26])
+
+    def time_list_by_status(self) -> None:
+        self.store.list_by_status(AsyncJobStatus.QUEUED)
+
+    def time_list_all(self) -> None:
+        self.store.list_all()
+
+    def time_count(self) -> None:
+        self.store.count()
+
+
+# ---------------------------------------------------------------------------
+# AsyncWorkerConfig benchmarks
+# ---------------------------------------------------------------------------
+
+
+class AsyncWorkerConfigSuite:
+    """Benchmark AsyncWorkerConfig instantiation overhead."""
+
+    timeout = 60
+
+    def time_default_config(self) -> None:
+        AsyncWorkerConfig()
+
+    def time_custom_config(self) -> None:
+        AsyncWorkerConfig(
+            enabled=True,
+            max_workers=8,
+            poll_interval=0.5,
+            job_timeout=7200,
+            job_ttl=172800,
+        )
+
+
+# ---------------------------------------------------------------------------
+# WorkerHealthReport benchmarks
+# ---------------------------------------------------------------------------
+
+
+class WorkerHealthReportSuite:
+    """Benchmark WorkerHealthReport lifecycle overhead."""
+
+    timeout = 60
+
+    def setup(self) -> None:
+        self.report = WorkerHealthReport("worker-bench")
+
+    def time_create_report(self) -> None:
+        WorkerHealthReport("worker-bench")
+
+    def time_record_heartbeat(self) -> None:
+        self.report.record_heartbeat()
+
+    def time_record_job_completed(self) -> None:
+        self.report.record_job_completed()
+
+    def time_record_job_failed(self) -> None:
+        self.report.record_job_failed()
+
+    def time_as_dict(self) -> None:
+        self.report.as_dict()
+
+
+# ---------------------------------------------------------------------------
+# CancellationToken benchmarks
+# ---------------------------------------------------------------------------
+
+
+class CancellationTokenSuite:
+    """Benchmark CancellationToken operations."""
+
+    timeout = 60
+
+    def time_create_token(self) -> None:
+        CancellationToken()
+
+    def time_check_not_cancelled(self) -> None:
+        token = CancellationToken()
+        token.is_cancelled
+
+    def time_cancel_and_check(self) -> None:
+        token = CancellationToken()
+        token.cancel()
+        token.is_cancelled
+
+    def time_reset_token(self) -> None:
+        token = CancellationToken()
+        token.cancel()
+        token.reset()
+
+    def time_wait_with_timeout(self) -> None:
+        token = CancellationToken()
+        token.wait(timeout=0.0)
+
+
+# ---------------------------------------------------------------------------
+# AsyncWorker benchmarks
+# ---------------------------------------------------------------------------
+
+
+class AsyncWorkerSuite:
+    """Benchmark AsyncWorker job execution overhead."""
+
+    timeout = 60
+
+    def setup(self) -> None:
+        self.config = AsyncWorkerConfig(enabled=True)
+        self.store = InMemoryJobStore()
+
+    def time_create_worker(self) -> None:
+        AsyncWorker(config=self.config, job_store=self.store)
+
+    def time_pickup_and_execute(self) -> None:
+        store = InMemoryJobStore()
+        job = AsyncJob(job_id=_ULID_A, plan_id=_ULID_B, phase="execute")
+        store.add(job)
+        worker = AsyncWorker(config=self.config, job_store=store)
+        worker.pickup_and_execute(job)
+
+    def time_get_health_report(self) -> None:
+        worker = AsyncWorker(config=self.config, job_store=self.store)
+        worker.get_health_report()
+
+    def time_detect_stuck_jobs_empty(self) -> None:
+        worker = AsyncWorker(config=self.config, job_store=self.store)
+        worker.detect_stuck_jobs()
+
+    def time_cleanup_completed_jobs_empty(self) -> None:
+        worker = AsyncWorker(config=self.config, job_store=self.store)
+        worker.cleanup_completed_jobs()
+
+
+# ---------------------------------------------------------------------------
+# InvalidJobTransitionError benchmarks
+# ---------------------------------------------------------------------------
+
+
+class ErrorConstructionSuite:
+    """Benchmark error construction overhead."""
+
+    timeout = 60
+
+    def time_invalid_transition_error(self) -> None:
+        InvalidJobTransitionError(AsyncJobStatus.SUCCEEDED, AsyncJobStatus.RUNNING)

docs/reference/async_architecture.md

@@ -0,0 +1,255 @@
+# 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 |

features/async_execution.feature

@@ -0,0 +1,555 @@
+Feature: Async Job Execution and Workers
+  As a developer
+  I want plan phases to execute as background jobs
+  So that long-running operations don't block the CLI
+
+  # ---- AsyncJob Domain Model ----
+
+  Scenario: Create an async job with default values
+    Given an async job store is initialised
+    And I create a new async job for plan "01HXYZ01HXYZ01HXYZ01HXYZ01" phase "execute"
+    Then the async job status should be "queued"
+    And the async job phase should be "execute"
+    And the async job schema_version should be 1
+
+  Scenario: AsyncJob status transitions follow the state machine
+    Given an async job store is initialised
+    And I create a new async job for plan "01HXYZ01HXYZ01HXYZ01HXYZ01" phase "execute"
+    When I mark the async job as running with worker "worker-001"
+    Then the async job status should be "running"
+    And the async job worker_id should be "worker-001"
+    And the async job started_at should be set
+
+  Scenario: AsyncJob transition from running to succeeded
+    Given an async job store is initialised
+    And I create a running async job
+    When I mark the async job as succeeded
+    Then the async job status should be "succeeded"
+    And the async job finished_at should be set
+
+  Scenario: AsyncJob transition from running to failed
+    Given an async job store is initialised
+    And I create a running async job
+    When I mark the async job as failed
+    Then the async job status should be "failed"
+    And the async job finished_at should be set
+
+  Scenario: AsyncJob transition from queued to cancelled
+    Given an async job store is initialised
+    And I create a new async job for plan "01HXYZ01HXYZ01HXYZ01HXYZ01" phase "execute"
+    When I mark the async job as cancelled
+    Then the async job status should be "cancelled"
+    And the async job finished_at should be set
+
+  Scenario: AsyncJob transition from running to cancelled
+    Given an async job store is initialised
+    And I create a running async job
+    When I mark the async job as cancelled
+    Then the async job status should be "cancelled"
+
+  Scenario: Invalid transition from queued to succeeded raises error
+    Given an async job store is initialised
+    And I create a new async job for plan "01HXYZ01HXYZ01HXYZ01HXYZ01" phase "execute"
+    When I try to mark the async job as succeeded
+    Then an InvalidJobTransitionError should be raised
+
+  Scenario: Invalid transition from succeeded to running raises error
+    Given an async job store is initialised
+    And I create a running async job
+    And I mark the async job as succeeded
+    When I try to mark the async job as running with worker "worker-002"
+    Then an InvalidJobTransitionError should be raised
+
+  Scenario: Invalid transition from failed raises error
+    Given an async job store is initialised
+    And I create a running async job
+    And I mark the async job as failed
+    When I try to mark the async job as succeeded
+    Then an InvalidJobTransitionError should be raised
+
+  # ---- Worker Pickup and Execution ----
+
+  Scenario: Worker picks up and executes a queued job
+    Given an async worker is initialised with max_workers 2
+    And a queued async job exists
+    When the worker picks up and executes the job
+    Then the async job status should be "succeeded"
+    And the worker health should show 1 job processed
+
+  Scenario: Worker marks job as failed on execution error
+    Given an async worker is initialised with a failing executor
+    And a queued async job exists
+    When the worker picks up and executes the job
+    Then the async job status should be "failed"
+    And the worker health should show 1 job failed
+
+  # ---- Job Cancellation ----
+
+  Scenario: Cancel a running job via cancellation token
+    Given an async worker is initialised with max_workers 2
+    And a queued async job exists
+    When I cancel the async job
+    Then the async job status should be "cancelled"
+
+  # ---- Stuck Job Detection ----
+
+  Scenario: Detect and fail stuck jobs with expired heartbeat
+    Given an async worker is initialised with job_timeout 1
+    And a running async job with expired heartbeat exists
+    When the worker detects stuck jobs
+    Then the stuck job should be marked as "failed"
+
+  # ---- Job TTL and Cleanup ----
+
+  Scenario: Cleanup completed jobs older than retention TTL
+    Given an async worker is initialised with job_ttl 1
+    And a completed async job older than TTL exists
+    When the worker runs job cleanup
+    Then the completed job should be removed from the store
+
+  # ---- Graceful Shutdown ----
+
+  Scenario: Worker stops gracefully on shutdown request
+    Given an async worker is initialised with max_workers 2
+    When the worker is started
+    And a shutdown is requested
+    Then the worker should stop running
+
+  # ---- Payload Serialization ----
+
+  Scenario: Serialize and deserialize job payload with schema version
+    When I serialize a job payload for plan "01HXYZ01HXYZ01HXYZ01HXYZ01" phase "execute"
+    Then the serialized payload should contain schema_version 1
+    And the deserialized payload should match the original
+
+  Scenario: Serialize payload with extra data
+    When I serialize a job payload with extra data
+    Then the serialized payload should contain the extra data
+
+  Scenario: Deserialize invalid JSON raises error
+    When I try to deserialize invalid JSON
+    Then a ValueError should be raised for async deserialize
+
+  # ---- Worker Health Report ----
+
+  Scenario: Worker health report contains expected fields
+    Given an async worker is initialised with max_workers 4
+    When I request the worker health report
+    Then the health report should contain worker_id
+    And the health report should contain config details
+
+  # ---- AsyncWorkerConfig Validation ----
+
+  Scenario: AsyncWorkerConfig rejects invalid max_workers
+    When I try to create AsyncWorkerConfig with max_workers 0
+    Then a ValueError should be raised for config
+
+  Scenario: AsyncWorkerConfig rejects invalid poll_interval
+    When I try to create AsyncWorkerConfig with poll_interval 0
+    Then a ValueError should be raised for config
+
+  # ---- InMemoryJobStore ----
+
+  Scenario: Job store rejects duplicate job IDs
+    Given an async job store is initialised
+    And I create a new async job for plan "01HXYZ01HXYZ01HXYZ01HXYZ01" phase "execute"
+    When I try to add the same job again
+    Then a ValueError should be raised for duplicate job
+
+  Scenario: Job store lists jobs by status
+    Given an async job store is initialised
+    And I create 3 queued async jobs
+    When I list jobs by status "queued"
+    Then the job list should have 3 entries
+
+  Scenario: AsyncJob heartbeat recording
+    Given an async job store is initialised
+    And I create a running async job
+    When I record a heartbeat on the async job
+    Then the async job last_heartbeat should be updated
+
+  Scenario: Heartbeat on non-running job raises error
+    Given an async job store is initialised
+    And I create a new async job for plan "01HXYZ01HXYZ01HXYZ01HXYZ01" phase "execute"
+    When I try to record a heartbeat on the async job
+    Then a ValueError should be raised for heartbeat
+
+  Scenario: AsyncJob is_terminal property
+    Given an async job store is initialised
+    And I create a running async job
+    When I mark the async job as succeeded
+    Then the async job should be terminal
+
+  Scenario: AsyncJob is_running property
+    Given an async job store is initialised
+    And I create a running async job
+    Then the async job should be running
+
+  Scenario: AsyncJob as_cli_dict returns expected keys
+    Given an async job store is initialised
+    And I create a running async job
+    When I get the async job CLI dict
+    Then the async CLI dict should contain key "job_id"
+    And the async CLI dict should contain key "status"
+    And the async CLI dict should contain key "worker_id"
+
+  Scenario: Invalid phase raises validation error
+    When I try to create an async job with phase "invalid"
+    Then a validation error should be raised for phase
+
+  Scenario: Invalid payload JSON raises validation error
+    When I try to create an async job with invalid payload JSON
+    Then a validation error should be raised for payload
+
+  Scenario: Mark running with empty worker_id raises error
+    Given an async job store is initialised
+    And I create a new async job for plan "01HXYZ01HXYZ01HXYZ01HXYZ01" phase "execute"
+    When I try to mark the async job as running with an empty worker_id
+    Then a ValueError should be raised for worker_id
+
+  Scenario: CancellationToken cooperative cancellation
+    Given a cancellation token is created
+    When I cancel the async token
+    Then the token should be cancelled
+    And the token wait should return True
+
+  Scenario: CancellationToken reset
+    Given a cancellation token is created
+    And I cancel the async token
+    When I reset the token
+    Then the token should not be cancelled
+
+  Scenario: WorkerHealthReport tracks jobs
+    Given a worker health report for "test-worker"
+    When I record 3 completed jobs and 1 failed job
+    Then the health report should show 3 processed and 1 failed
+
+  Scenario: Serialize payload with empty plan_id raises error
+    When I try to serialize a payload with empty plan_id
+    Then a ValueError should be raised for serialize
+
+  Scenario: Deserialize empty payload raises error
+    When I try to deserialize an empty payload
+    Then a ValueError should be raised for deserialize
+
+  Scenario: can_transition_job with invalid types raises TypeError
+    When I call can_transition_job with invalid types
+    Then a TypeError should be raised for async transition
+
+  # ---- Additional Coverage: Validation Guards ----
+
+  Scenario: WorkerHealthReport with empty worker_id raises error
+    When I try to create a WorkerHealthReport with empty worker_id
+    Then a ValueError should be raised for empty worker_id
+
+  Scenario: AsyncWorkerConfig with invalid enabled type raises error
+    When I try to create AsyncWorkerConfig with non-bool enabled
+    Then a TypeError should be raised for async config enabled
+
+  Scenario: AsyncWorkerConfig with invalid job_timeout raises error
+    When I try to create AsyncWorkerConfig with job_timeout 0
+    Then a ValueError should be raised for async config validation
+
+  Scenario: AsyncWorkerConfig with invalid job_ttl raises error
+    When I try to create AsyncWorkerConfig with job_ttl 0
+    Then a ValueError should be raised for async config validation
+
+  Scenario: InMemoryJobStore add with non-AsyncJob raises error
+    Given an async job store is initialised
+    When I try to add a non-AsyncJob to the store
+    Then a TypeError should be raised for async store type
+
+  Scenario: InMemoryJobStore get with empty job_id raises error
+    Given an async job store is initialised
+    When I try to get a job with empty job_id
+    Then a ValueError should be raised for async store get
+
+  Scenario: InMemoryJobStore list_by_status with invalid type raises error
+    Given an async job store is initialised
+    When I try to list jobs by invalid status type
+    Then a TypeError should be raised for async store type
+
+  Scenario: InMemoryJobStore update with non-AsyncJob raises error
+    Given an async job store is initialised
+    When I try to update with a non-AsyncJob
+    Then a TypeError should be raised for async store type
+
+  Scenario: InMemoryJobStore update with non-existent job raises error
+    Given an async job store is initialised
+    When I try to update a non-existent job
+    Then a ValueError should be raised for async store update
+
+  Scenario: InMemoryJobStore remove with empty job_id raises error
+    Given an async job store is initialised
+    When I try to remove a job with empty job_id
+    Then a ValueError should be raised for async store remove
+
+  Scenario: AsyncWorker with invalid config type raises error
+    When I try to create AsyncWorker with invalid config
+    Then a TypeError should be raised for async worker config
+
+  Scenario: AsyncWorker with invalid job_store type raises error
+    When I try to create AsyncWorker with invalid job_store
+    Then a TypeError should be raised for async worker store
+
+  Scenario: AsyncWorker pickup with non-AsyncJob raises error
+    Given an async worker is initialised with max_workers 2
+    When I try to pickup a non-AsyncJob
+    Then a TypeError should be raised for async worker pickup
+
+  Scenario: AsyncWorker cancel_job with empty job_id raises error
+    Given an async worker is initialised with max_workers 2
+    When I try to cancel a job with empty job_id
+    Then a ValueError should be raised for async worker cancel
+
+  Scenario: AsyncWorker worker_id property returns value
+    Given an async worker is initialised with max_workers 2
+    Then the worker should have a worker_id starting with "worker-"
+
+  Scenario: AsyncWorker config property returns config
+    Given an async worker is initialised with max_workers 2
+    Then the worker config max_workers should be 2
+
+  Scenario: AsyncWorker start when already running does nothing
+    Given an async worker is initialised with max_workers 2
+    When I start the worker
+    And I start the worker again
+    Then the worker should still be running
+
+  Scenario: AsyncWorker stop after start stops cleanly
+    Given an async worker is initialised with max_workers 2
+    When I start the worker
+    And I stop the worker
+    Then the worker should not be running
+
+  Scenario: AsyncWorker stop when not running does nothing
+    Given an async worker is initialised with max_workers 2
+    When I stop the worker
+    Then the worker should not be running
+
+  Scenario: AsyncWorker cancel job that does not exist returns false
+    Given an async worker is initialised with max_workers 2
+    Then cancelling a non-existent job should return false
+
+  Scenario: AsyncJob payload property deserializes JSON
+    Given an async job store is initialised
+    And I create a new async job for plan "01HXYZ01HXYZ01HXYZ01HXYZ01" phase "execute"
+    Then the async job payload property should return a dict
+
+  Scenario: AsyncJob validate_payload_json rejects invalid JSON
+    When I try to create an async job with payload_json "not-json{"
+    Then a validation error should be raised for payload
+
+  Scenario: AsyncJob as_cli_dict includes optional fields when set
+    Given an async job store is initialised
+    And I create a running async job
+    And I mark the async job as succeeded
+    When I get the async job CLI dict
+    Then the async CLI dict should contain key "started_at"
+    And the async CLI dict should contain key "finished_at"
+
+  # ---- Review Fixes: Concurrent Execution (T2) ----
+
+  Scenario: Worker executes multiple jobs concurrently
+    Given an async worker is initialised with max_workers 4
+    And 3 queued async jobs exist
+    When the worker executes all queued jobs concurrently
+    Then all 3 jobs should be in a terminal state
+    And the worker health should show 3 jobs processed total
+
+  # ---- Review Fixes: Cancel records cancelled counter (B2) ----
+
+  Scenario: Cancelled job increments cancelled counter not processed counter
+    Given an async worker is initialised with max_workers 2
+    And a queued async job exists
+    When I cancel the async job
+    Then the async job status should be "cancelled"
+    And the worker health should show 0 jobs processed
+    And the worker health should show 0 jobs failed
+
+  # ---- Review Fixes: Error message audit trail (SEC1) ----
+
+  Scenario: Failed job records error message for audit trail
+    Given an async worker is initialised with a failing executor
+    And a queued async job exists
+    When the worker picks up and executes the job
+    Then the async job status should be "failed"
+    And the async job should have an error_message set
+
+  # ---- Review Fixes: DB model round-trip (T1) ----
+
+  Scenario: AsyncJobModel round-trip through SQLAlchemy
+    When I create and persist an AsyncJobModel to the database
+    Then the AsyncJobModel should be retrievable with correct fields
+    And the database CHECK constraint should reject invalid status
+
+  # ---- Coverage: Safe Initialisation and Cleanup ----
+
+  Scenario: Worker start creates thread pool and stop destroys it cleanly
+    Given an async worker is initialised with max_workers 2
+    And a queued async job exists
+    When I start the worker
+    And the worker processes all queued jobs within 3 seconds
+    And I stop the worker
+    Then the worker should not be running
+    And the async job status should be "succeeded"
+
+  Scenario: Worker poll loop dispatches jobs via thread pool
+    Given an async worker is initialised with max_workers 4
+    And 2 queued async jobs exist
+    When I start the worker
+    And all created jobs reach a terminal state within 3 seconds
+    And I stop the worker
+    Then all 2 jobs should be in a terminal state
+    And the worker health should show 2 jobs processed total
+
+  Scenario: Worker stop cancels in-flight jobs via tokens
+    Given an async worker is initialised with a blocking executor
+    And a queued async job exists
+    When I start the worker
+    And the blocking executor has been entered
+    And I stop the worker
+    Then the cancellation token for the job should be signalled
+
+  # ---- Coverage: Cancellation Paths ----
+
+  Scenario: Running job cancelled via token records cancelled count
+    Given an async worker is initialised with a blocking executor
+    And a queued async job exists
+    When I dispatch the job in a background thread
+    And the blocking executor has been entered
+    And I cancel the running async job via the worker
+    And the background job thread completes
+    Then the async job status should be "cancelled"
+    And the worker health should show 1 jobs cancelled
+    And the worker health should show 0 jobs processed
+
+  Scenario: Pickup and execute skips already-terminal job safely
+    Given an async worker is initialised with max_workers 2
+    And a queued async job exists
+    And the async job is already cancelled
+    When the worker picks up and executes the job
+    Then the async job status should be "cancelled"
+    And the worker health should show 0 jobs processed
+
+  Scenario: Executor that moves job to terminal during execution is handled
+    Given an async worker is initialised with a terminal-forcing executor
+    And a queued async job exists
+    When the worker picks up and executes the job
+    Then the async job should be in a terminal state
+    And the worker health should show 0 jobs processed
+
+  # ---- Coverage: Job Store Operations ----
+
+  Scenario: Job store list_all returns all jobs regardless of status
+    Given an async job store is initialised
+    And I add 3 jobs in mixed statuses to the store
+    When I call list_all on the job store
+    Then the store should return 3 jobs
+    And the store count should be 3
+
+  Scenario: Job store remove deletes a job and count reflects it
+    Given an async job store is initialised
+    And a queued async job exists in the store
+    When I remove the job from the store
+    Then the store count should be 0
+
+  # ---- Coverage: Domain Model Validators ----
+
+  Scenario: transition_to rejects non-AsyncJobStatus target
+    Given a queued async job exists standalone
+    When I try to transition_to with a string instead of enum
+    Then an async TypeError should be raised
+
+  Scenario: can_transition_job rejects non-enum to_status argument
+    When I call can_transition_job with a string to_status
+    Then an async TypeError should be raised
+
+  Scenario: serialize_job_payload rejects empty phase
+    When I try to serialize a payload with an empty phase
+    Then an async ValueError should be raised matching "phase"
+
+  Scenario: serialize_job_payload rejects zero schema version
+    When I try to serialize a payload with schema_version 0
+    Then an async ValueError should be raised matching "schema_version"
+
+  Scenario: deserialize_job_payload rejects non-object JSON
+    When I try to deserialize the JSON string "[1,2,3]"
+    Then an async ValueError should be raised matching "JSON object"
+
+  Scenario: Pickup handles already-running job gracefully
+    Given an async worker is initialised with max_workers 2
+    And a queued async job exists
+    And the async job is already running with worker "other-worker"
+    When the worker picks up and executes the job
+    Then the async job status should be "running"
+    And the worker health should show 0 jobs processed
+
+  Scenario: Executor exception on already-terminal job skips mark_failed
+    Given an async worker is initialised with a crash-after-terminal executor
+    And a queued async job exists
+    When the worker picks up and executes the job
+    Then the async job should be in a terminal state
+
+  Scenario: detect_stuck_jobs tolerates already-terminal jobs
+    Given an async worker is initialised with job_timeout 1
+    And a running async job with expired heartbeat exists
+    And the running async job has already been cancelled externally
+    When the worker detects stuck jobs
+    Then no additional stuck jobs should be reported
+
+  Scenario: CLI dict includes error_message for failed job
+    Given a queued async job exists standalone
+    When I mark the async job as running with worker "w-err"
+    And I mark the async job as failed with error "SomeError: detail"
+    And I get the async job CLI dict
+    Then the async CLI dict should contain key "error_message"
+    And the async CLI dict error_message should be "SomeError: detail"
+
+  # ---- Review Fix: Plan lifecycle async wiring (P1) ----
+
+  Scenario: execute_plan enqueues async job when async is enabled
+    Given a plan lifecycle service with async enabled and a job store
+    And a plan in Strategize/COMPLETE state exists
+    When I execute the plan via the lifecycle service
+    Then the plan should be in Execute/QUEUED state
+    And the job store should contain 1 async job for phase "execute"
+    And the enqueued job plan_id should match the plan
+
+  Scenario: apply_plan enqueues async job when async is enabled
+    Given a plan lifecycle service with async enabled and a job store
+    And a plan in Execute/COMPLETE state exists
+    When I apply the plan via the lifecycle service
+    Then the plan should be in Apply/QUEUED state
+    And the job store should contain 1 async job for phase "apply"
+    And the enqueued job plan_id should match the plan
+
+  Scenario: execute_plan does not enqueue when async is disabled
+    Given a plan lifecycle service with async disabled
+    And a plan in Strategize/COMPLETE state exists
+    When I execute the plan via the lifecycle service
+    Then the plan should be in Execute/QUEUED state
+    And no async jobs should be in the job store
+
+  Scenario: apply_plan does not enqueue when async is disabled
+    Given a plan lifecycle service with async disabled
+    And a plan in Execute/COMPLETE state exists
+    When I apply the plan via the lifecycle service
+    Then the plan should be in Apply/QUEUED state
+    And no async jobs should be in the job store
+
+  Scenario: execute_plan does not enqueue when no job store is configured
+    Given a plan lifecycle service with async enabled but no job store
+    And a plan in Strategize/COMPLETE state exists
+    When I execute the plan via the lifecycle service
+    Then the plan should be in Execute/QUEUED state
+
+  # ---- Review Fix: Error message redaction (P2) ----
+
+  Scenario: Failed job error message redacts secrets
+    Given an async worker is initialised with a secret-leaking executor
+    And a queued async job exists
+    When the worker picks up and executes the job
+    Then the async job status should be "failed"
+    And the async job error_message should not contain the raw secret
+    And the async job error_message should contain "***REDACTED***"

features/steps/decision_di_wiring_steps.py

@@ -108,6 +108,7 @@ def step_create_lifecycle_with_decision(context: Context) -> None:
 
     mock_settings = create_autospec(Settings, instance=True)
     mock_settings.database_url = "sqlite:///:memory:"
+    mock_settings.async_enabled = False
 
     decision_svc = DecisionService(settings=mock_settings, unit_of_work=uow)
     lifecycle_svc = PlanLifecycleService(

robot/async_execution.robot

@@ -0,0 +1,52 @@
+*** Settings ***
+Documentation     Smoke tests for async job execution: domain model, worker,
+...               payload serialization, and config validation.
+Resource          ${CURDIR}/common.resource
+Suite Setup       Setup Test Environment
+Suite Teardown    Cleanup Test Environment
+
+*** Variables ***
+${HELPER_SCRIPT}    robot/helper_async_execution.py
+
+*** Test Cases ***
+Async Job Status Transitions
+    [Documentation]    Verify job status transitions follow the state machine
+    [Tags]    async    domain    transitions
+    ${result}=    Run Process    ${PYTHON}    ${HELPER_SCRIPT}    status-transitions    cwd=${WORKSPACE}
+    Should Be Equal As Integers    ${result.rc}    0
+    Should Contain    ${result.stdout}    status-transitions-ok
+
+Async Worker Pickup And Execute
+    [Documentation]    Verify worker picks up and executes a queued job
+    [Tags]    async    worker    execution
+    ${result}=    Run Process    ${PYTHON}    ${HELPER_SCRIPT}    worker-execute    cwd=${WORKSPACE}
+    Should Be Equal As Integers    ${result.rc}    0
+    Should Contain    ${result.stdout}    worker-execute-ok
+
+Async Job Payload Serialization
+    [Documentation]    Verify payload serialization round-trip
+    [Tags]    async    serialization
+    ${result}=    Run Process    ${PYTHON}    ${HELPER_SCRIPT}    payload-roundtrip    cwd=${WORKSPACE}
+    Should Be Equal As Integers    ${result.rc}    0
+    Should Contain    ${result.stdout}    payload-roundtrip-ok
+
+Async Worker Health Report
+    [Documentation]    Verify worker health report contains expected fields
+    [Tags]    async    worker    health
+    ${result}=    Run Process    ${PYTHON}    ${HELPER_SCRIPT}    health-report    cwd=${WORKSPACE}
+    Should Be Equal As Integers    ${result.rc}    0
+    Should Contain    ${result.stdout}    health-report-ok
+
+Async Stuck Job Detection
+    [Documentation]    Verify stuck jobs are detected and marked failed
+    [Tags]    async    worker    stuck
+    ${result}=    Run Process    ${PYTHON}    ${HELPER_SCRIPT}    stuck-detection    cwd=${WORKSPACE}
+    Should Be Equal As Integers    ${result.rc}    0
+    Should Contain    ${result.stdout}    stuck-detection-ok
+
+Async Job Cleanup
+    [Documentation]    Verify completed jobs are cleaned up after TTL
+    [Tags]    async    worker    cleanup
+    ${result}=    Run Process    ${PYTHON}    ${HELPER_SCRIPT}    job-cleanup    cwd=${WORKSPACE}
+    Should Be Equal As Integers    ${result.rc}    0
+    Should Contain    ${result.stdout}    job-cleanup-ok

robot/helper_async_execution.py

@@ -0,0 +1,198 @@
+"""Helper script for Robot Framework async execution smoke tests.
+
+Usage:
+    python robot/helper_async_execution.py <test-name>
+
+Supported test names:
+    status-transitions   Test job status state machine
+    worker-execute       Test worker pickup and execution
+    payload-roundtrip    Test payload serialization round-trip
+    health-report        Test worker health report
+    stuck-detection      Test stuck job detection
+    job-cleanup          Test completed job cleanup
+"""
+
+from __future__ import annotations
+
+import sys
+import traceback
+from datetime import UTC, datetime, timedelta
+from pathlib import Path
+
+# Ensure src is on the path
+sys.path.insert(0, str(Path(__file__).resolve().parents[1] / "src"))
+
+from ulid import ULID
+
+from cleveragents.application.services.async_worker import (
+    AsyncWorker,
+    AsyncWorkerConfig,
+    InMemoryJobStore,
+)
+from cleveragents.domain.models.core.async_job import (
+    AsyncJob,
+    AsyncJobStatus,
+    InvalidJobTransitionError,
+    deserialize_job_payload,
+    serialize_job_payload,
+)
+
+PLAN_ID = "01HXYZ01HXYZ01HXYZ01HXYZ01"
+
+
+def _job_id() -> str:
+    return str(ULID())
+
+
+def test_status_transitions() -> None:
+    """Test job status transitions follow the state machine."""
+    # queued -> running -> succeeded
+    job = AsyncJob(job_id=_job_id(), plan_id=PLAN_ID, phase="execute")
+    assert job.status == AsyncJobStatus.QUEUED, "FAIL: initial status should be queued"
+    job.mark_running("w1")
+    assert job.status == AsyncJobStatus.RUNNING, "FAIL: status should be running"
+    job.mark_succeeded()
+    assert job.status == AsyncJobStatus.SUCCEEDED, "FAIL: status should be succeeded"
+    assert job.is_terminal, "FAIL: succeeded job should be terminal"
+    print("  CHECK queued->running->succeeded: pass")
+
+    # queued -> running -> failed (with error message)
+    job2 = AsyncJob(job_id=_job_id(), plan_id=PLAN_ID, phase="execute")
+    job2.mark_running("w2")
+    job2.mark_failed(error="test error")
+    assert job2.status == AsyncJobStatus.FAILED, "FAIL: status should be failed"
+    assert job2.error_message == "test error", "FAIL: error_message should be set"
+    print("  CHECK queued->running->failed (with error): pass")
+
+    # queued -> cancelled
+    job3 = AsyncJob(job_id=_job_id(), plan_id=PLAN_ID, phase="apply")
+    job3.mark_cancelled()
+    assert job3.status == AsyncJobStatus.CANCELLED, "FAIL: status should be cancelled"
+    print("  CHECK queued->cancelled: pass")
+
+    # Invalid: queued -> succeeded
+    job4 = AsyncJob(job_id=_job_id(), plan_id=PLAN_ID, phase="execute")
+    try:
+        job4.mark_succeeded()
+        raise AssertionError("Should have raised InvalidJobTransitionError")
+    except InvalidJobTransitionError:
+        pass
+    print("  CHECK queued->succeeded raises error: pass")
+
+    print("status-transitions-ok")
+
+
+def test_worker_execute() -> None:
+    """Test worker pickup and execution."""
+    store = InMemoryJobStore()
+    config = AsyncWorkerConfig(enabled=True, max_workers=2, poll_interval=0.1)
+    worker = AsyncWorker(config, store)
+
+    job = AsyncJob(job_id=_job_id(), plan_id=PLAN_ID, phase="execute")
+    store.add(job)
+
+    worker.pickup_and_execute(job)
+    assert job.status == AsyncJobStatus.SUCCEEDED
+    assert worker.health.jobs_processed == 1
+    print("worker-execute-ok")
+
+
+def test_payload_roundtrip() -> None:
+    """Test payload serialization round-trip."""
+    payload = serialize_job_payload(PLAN_ID, "execute", extra={"key": "val"})
+    data = deserialize_job_payload(payload)
+    assert data["plan_id"] == PLAN_ID
+    assert data["phase"] == "execute"
+    assert data["schema_version"] == 1
+    assert data["extra"]["key"] == "val"
+    print("payload-roundtrip-ok")
+
+
+def test_health_report() -> None:
+    """Test worker health report."""
+    store = InMemoryJobStore()
+    config = AsyncWorkerConfig(enabled=True, max_workers=4, poll_interval=0.5)
+    worker = AsyncWorker(config, store)
+
+    report = worker.get_health_report()
+    assert "worker_id" in report
+    assert "config" in report
+    assert report["config"]["max_workers"] == 4
+    print("health-report-ok")
+
+
+def test_stuck_detection() -> None:
+    """Test stuck job detection."""
+    store = InMemoryJobStore()
+    config = AsyncWorkerConfig(
+        enabled=True, max_workers=2, poll_interval=0.1, job_timeout=1
+    )
+    worker = AsyncWorker(config, store)
+
+    job = AsyncJob(job_id=_job_id(), plan_id=PLAN_ID, phase="execute")
+    store.add(job)
+    job.mark_running("worker-stuck")
+    job.last_heartbeat = datetime.now(UTC) - timedelta(seconds=10)
+    job.started_at = datetime.now(UTC) - timedelta(seconds=10)
+    store.update(job)
+
+    stuck = worker.detect_stuck_jobs()
+    assert len(stuck) == 1
+    assert stuck[0].status == AsyncJobStatus.FAILED
+    print("stuck-detection-ok")
+
+
+def test_job_cleanup() -> None:
+    """Test completed job cleanup."""
+    store = InMemoryJobStore()
+    config = AsyncWorkerConfig(
+        enabled=True, max_workers=2, poll_interval=0.1, job_ttl=1
+    )
+    worker = AsyncWorker(config, store)
+
+    job = AsyncJob(job_id=_job_id(), plan_id=PLAN_ID, phase="execute")
+    store.add(job)
+    job.mark_running("worker-old")
+    job.mark_succeeded()
+    job.finished_at = datetime.now(UTC) - timedelta(seconds=10)
+    store.update(job)
+
+    cleaned = worker.cleanup_completed_jobs()
+    assert cleaned == 1
+    assert store.get(job.job_id) is None
+    print("job-cleanup-ok")
+
+
+TESTS = {
+    "status-transitions": test_status_transitions,
+    "worker-execute": test_worker_execute,
+    "payload-roundtrip": test_payload_roundtrip,
+    "health-report": test_health_report,
+    "stuck-detection": test_stuck_detection,
+    "job-cleanup": test_job_cleanup,
+}
+
+
+def main() -> None:
+    if len(sys.argv) < 2:
+        print(f"Usage: {sys.argv[0]} <test-name>", file=sys.stderr)
+        print(f"Available: {', '.join(TESTS)}", file=sys.stderr)
+        sys.exit(1)
+
+    test_name = sys.argv[1]
+    test_fn = TESTS.get(test_name)
+    if test_fn is None:
+        print(f"Unknown test: {test_name}", file=sys.stderr)
+        sys.exit(1)
+
+    try:
+        test_fn()
+    except Exception as exc:
+        # Print detailed assertion context for Robot test diagnostics
+        print(f"FAIL: {test_name}: {exc}", file=sys.stderr)
+        traceback.print_exc(file=sys.stderr)
+        sys.exit(1)
+
+
+if __name__ == "__main__":
+    main()

robot/helper_decision_di.py

@@ -72,6 +72,7 @@ def _record_integration() -> None:
 
     mock_settings = create_autospec(Settings, instance=True)
     mock_settings.database_url = "sqlite:///:memory:"
+    mock_settings.async_enabled = False
 
     decision_svc = DecisionService(settings=mock_settings, unit_of_work=uow)
     lifecycle_svc = PlanLifecycleService(

src/cleveragents/application/services/async_worker.py

@@ -0,0 +1,654 @@
+"""Async worker service for background job execution.
+
+The ``AsyncWorker`` service polls for queued jobs and executes them
+concurrently via a ``ThreadPoolExecutor``.  It supports configurable
+concurrency, graceful shutdown via signal handling, stuck job detection,
+and job cleanup.
+
+## Features
+
+- Polling loop with configurable interval (``async.poll_interval``)
+- True concurrent execution via ThreadPoolExecutor (``async.max_workers``)
+- Graceful shutdown via SIGINT/SIGTERM signal handling
+- Stuck job detection: mark jobs failed after TTL if no heartbeat
+- Worker health report (last_heartbeat, jobs processed / failed / cancelled)
+- Cancellation token support with propagation to tool execution
+- Job cleanup routine to prune completed jobs older than retention
+- Race-safe cancellation: ``cancel_job`` only signals the token;
+  ``pickup_and_execute`` owns all state transitions
+
+## Configuration Keys
+
+| Key                    | Type  | Default | Description                        |
+|------------------------|-------|---------|------------------------------------|
+| ``async.enabled``      | bool  | False   | Enable async job execution         |
+| ``async.max_workers``  | int   | 4       | Max concurrent worker threads      |
+| ``async.poll_interval``| float | 1.0     | Poll interval in seconds           |
+| ``async.job_timeout``  | int   | 3600    | Job timeout in seconds             |
+| ``async.job_ttl``      | int   | 86400   | Completed job retention (sec)      |
+
+Based on issue #312 (feat(async): add async command execution and workers).
+"""
+
+from __future__ import annotations
+
+import logging
+import signal
+import threading
+from concurrent.futures import Future, ThreadPoolExecutor
+from datetime import UTC, datetime, timedelta
+from typing import Any
+from uuid import uuid4
+
+from cleveragents.domain.models.core.async_job import (
+    AsyncJob,
+    AsyncJobStatus,
+    InvalidJobTransitionError,
+)
+from cleveragents.shared.redaction import redact_value
+
+logger = logging.getLogger(__name__)
+
+
+class CancellationToken:
+    """Thread-safe cancellation token for cooperative cancellation.
+
+    Workers check this token periodically and stop execution when
+    cancellation is requested.
+    """
+
+    def __init__(self) -> None:
+        self._cancelled = threading.Event()
+
+    @property
+    def is_cancelled(self) -> bool:
+        """Check if cancellation has been requested."""
+        return self._cancelled.is_set()
+
+    def cancel(self) -> None:
+        """Request cancellation."""
+        self._cancelled.set()
+
+    def reset(self) -> None:
+        """Reset the cancellation token."""
+        self._cancelled.clear()
+
+    def wait(self, timeout: float | None = None) -> bool:
+        """Wait for cancellation or timeout.
+
+        Args:
+            timeout: Maximum seconds to wait (None = wait forever).
+
+        Returns:
+            True if cancelled, False if timed out.
+        """
+        return self._cancelled.wait(timeout=timeout)
+
+
+class WorkerHealthReport:
+    """Health report for an async worker.
+
+    Tracks the worker's activity and status for diagnostics.
+    """
+
+    def __init__(self, worker_id: str) -> None:
+        if not worker_id or not worker_id.strip():
+            raise ValueError("worker_id must be a non-empty string")
+        self.worker_id: str = worker_id
+        self.last_heartbeat: datetime = datetime.now(UTC)
+        self.jobs_processed: int = 0
+        self.jobs_failed: int = 0
+        self.jobs_cancelled: int = 0
+        self.is_running: bool = False
+        self.started_at: datetime = datetime.now(UTC)
+
+    def record_heartbeat(self) -> None:
+        """Record a heartbeat."""
+        self.last_heartbeat = datetime.now(UTC)
+
+    def record_job_completed(self) -> None:
+        """Record a successfully completed job."""
+        self.jobs_processed += 1
+        self.record_heartbeat()
+
+    def record_job_failed(self) -> None:
+        """Record a failed job."""
+        self.jobs_failed += 1
+        self.record_heartbeat()
+
+    def record_job_cancelled(self) -> None:
+        """Record a cancelled job."""
+        self.jobs_cancelled += 1
+        self.record_heartbeat()
+
+    def as_dict(self) -> dict[str, Any]:
+        """Return a dictionary representation for diagnostics."""
+        return {
+            "worker_id": self.worker_id,
+            "last_heartbeat": self.last_heartbeat.isoformat(),
+            "jobs_processed": self.jobs_processed,
+            "jobs_failed": self.jobs_failed,
+            "jobs_cancelled": self.jobs_cancelled,
+            "is_running": self.is_running,
+            "started_at": self.started_at.isoformat(),
+        }
+
+
+class AsyncWorkerConfig:
+    """Configuration for the async worker service."""
+
+    def __init__(
+        self,
+        enabled: bool = False,
+        max_workers: int = 4,
+        poll_interval: float = 1.0,
+        job_timeout: int = 3600,
+        job_ttl: int = 86400,
+    ) -> None:
+        if not isinstance(enabled, bool):
+            raise TypeError(f"enabled must be bool, got {type(enabled)}")
+        if not isinstance(max_workers, int) or max_workers < 1:
+            raise ValueError("max_workers must be a positive integer")
+        if not isinstance(poll_interval, (int, float)) or poll_interval <= 0:
+            raise ValueError("poll_interval must be a positive number")
+        if not isinstance(job_timeout, int) or job_timeout < 1:
+            raise ValueError("job_timeout must be a positive integer")
+        if not isinstance(job_ttl, int) or job_ttl < 1:
+            raise ValueError("job_ttl must be a positive integer")
+        self.enabled: bool = enabled
+        self.max_workers: int = max_workers
+        self.poll_interval: float = float(poll_interval)
+        self.job_timeout: int = job_timeout
+        self.job_ttl: int = job_ttl
+
+
+class InMemoryJobStore:
+    """Thread-safe in-memory job store for async jobs.
+
+    Provides CRUD operations for jobs, used by the worker and tests.
+    Production implementations would back this with the database.
+    """
+
+    def __init__(self) -> None:
+        self._jobs: dict[str, AsyncJob] = {}
+        self._lock = threading.Lock()
+
+    def add(self, job: AsyncJob) -> None:
+        """Add a job to the store.
+
+        Args:
+            job: The job to add.
+
+        Raises:
+            ValueError: If a job with the same ID already exists.
+        """
+        if not isinstance(job, AsyncJob):
+            raise TypeError(f"job must be AsyncJob, got {type(job)}")
+        with self._lock:
+            if job.job_id in self._jobs:
+                raise ValueError(f"Job {job.job_id} already exists")
+            self._jobs[job.job_id] = job
+
+    def get(self, job_id: str) -> AsyncJob | None:
+        """Get a job by ID.
+
+        Args:
+            job_id: The job ULID.
+
+        Returns:
+            The job, or None if not found.
+        """
+        if not job_id:
+            raise ValueError("job_id must be a non-empty string")
+        with self._lock:
+            return self._jobs.get(job_id)
+
+    def list_by_status(self, status: AsyncJobStatus) -> list[AsyncJob]:
+        """List jobs with a specific status.
+
+        Args:
+            status: The status to filter by.
+
+        Returns:
+            List of jobs with the given status.
+        """
+        if not isinstance(status, AsyncJobStatus):
+            raise TypeError(f"status must be AsyncJobStatus, got {type(status)}")
+        with self._lock:
+            return [j for j in self._jobs.values() if j.status == status]
+
+    def list_all(self) -> list[AsyncJob]:
+        """List all jobs."""
+        with self._lock:
+            return list(self._jobs.values())
+
+    def update(self, job: AsyncJob) -> None:
+        """Update a job in the store.
+
+        Args:
+            job: The job with updated fields.
+
+        Raises:
+            ValueError: If the job does not exist.
+        """
+        if not isinstance(job, AsyncJob):
+            raise TypeError(f"job must be AsyncJob, got {type(job)}")
+        with self._lock:
+            if job.job_id not in self._jobs:
+                raise ValueError(f"Job {job.job_id} not found")
+            self._jobs[job.job_id] = job
+
+    def remove(self, job_id: str) -> None:
+        """Remove a job from the store.
+
+        Args:
+            job_id: The job ULID.
+        """
+        if not job_id:
+            raise ValueError("job_id must be a non-empty string")
+        with self._lock:
+            self._jobs.pop(job_id, None)
+
+    def count(self) -> int:
+        """Return the total number of jobs."""
+        with self._lock:
+            return len(self._jobs)
+
+    def snapshot_counts(self) -> dict[str, int]:
+        """Return all status counts atomically in a single lock acquisition.
+
+        Returns:
+            Dictionary with ``queued``, ``running``, and ``total`` keys.
+        """
+        with self._lock:
+            queued = sum(
+                1 for j in self._jobs.values() if j.status == AsyncJobStatus.QUEUED
+            )
+            running = sum(
+                1 for j in self._jobs.values() if j.status == AsyncJobStatus.RUNNING
+            )
+            return {"queued": queued, "running": running, "total": len(self._jobs)}
+
+    def remove_expired(self, ttl_seconds: int) -> int:
+        """Remove terminal jobs whose ``finished_at`` exceeds TTL.
+
+        Performs a single-pass removal under one lock acquisition to
+        avoid N+1 locking overhead.
+
+        Args:
+            ttl_seconds: Retention period in seconds.
+
+        Returns:
+            Number of jobs removed.
+        """
+        now = datetime.now(UTC)
+        ttl_delta = timedelta(seconds=ttl_seconds)
+        with self._lock:
+            expired_ids = [
+                jid
+                for jid, job in self._jobs.items()
+                if (
+                    job.is_terminal
+                    and job.finished_at is not None
+                    and now - job.finished_at > ttl_delta
+                )
+            ]
+            for jid in expired_ids:
+                del self._jobs[jid]
+            return len(expired_ids)
+
+
+class AsyncWorker:
+    """Async worker service for background job execution.
+
+    Uses a ``ThreadPoolExecutor`` to run jobs concurrently up to
+    ``max_workers``.  The poll loop dispatches queued jobs to the pool
+    and respects the concurrency limit.
+
+    **Cancellation contract:** ``cancel_job`` only signals the
+    cancellation token.  The ``pickup_and_execute`` method that owns
+    the job is responsible for all state transitions, preventing the
+    race condition where two threads attempt conflicting transitions
+    on the same job object.
+    """
+
+    def __init__(
+        self,
+        config: AsyncWorkerConfig,
+        job_store: InMemoryJobStore,
+        job_executor: Any | None = None,
+    ) -> None:
+        if not isinstance(config, AsyncWorkerConfig):
+            raise TypeError(f"config must be AsyncWorkerConfig, got {type(config)}")
+        if not isinstance(job_store, InMemoryJobStore):
+            raise TypeError(
+                f"job_store must be InMemoryJobStore, got {type(job_store)}"
+            )
+        self._config = config
+        self._job_store = job_store
+        self._job_executor = job_executor
+        self._worker_id = f"worker-{uuid4().hex[:8]}"
+        self._shutdown_event = threading.Event()
+        self._cancellation_tokens: dict[str, CancellationToken] = {}
+        self._tokens_lock = threading.Lock()
+        self._health = WorkerHealthReport(self._worker_id)
+        self._poll_thread: threading.Thread | None = None
+        self._thread_pool: ThreadPoolExecutor | None = None
+        self._futures: dict[str, Future[None]] = {}
+        self._futures_lock = threading.Lock()
+        self._original_sigint: Any = None
+        self._original_sigterm: Any = None
+
+    @property
+    def config(self) -> AsyncWorkerConfig:
+        """Return the worker configuration."""
+        return self._config
+
+    @property
+    def worker_id(self) -> str:
+        """Return the worker ID."""
+        return self._worker_id
+
+    @property
+    def health(self) -> WorkerHealthReport:
+        """Return the worker health report."""
+        return self._health
+
+    @property
+    def is_running(self) -> bool:
+        """Check if the worker is running."""
+        return self._health.is_running
+
+    def start(self) -> None:
+        """Start the worker polling loop.
+
+        Creates a ``ThreadPoolExecutor`` for concurrent job execution,
+        installs signal handlers for graceful shutdown, and starts the
+        background polling thread.
+        """
+        if self._health.is_running:
+            return
+        self._health.is_running = True
+        self._shutdown_event.clear()
+        self._thread_pool = ThreadPoolExecutor(
+            max_workers=self._config.max_workers,
+            thread_name_prefix=f"job-{self._worker_id}",
+        )
+        self._install_signal_handlers()
+        self._poll_thread = threading.Thread(
+            target=self._poll_loop,
+            name=f"async-worker-{self._worker_id}",
+            daemon=True,
+        )
+        self._poll_thread.start()
+        logger.info("AsyncWorker %s started", self._worker_id)
+
+    def stop(self) -> None:
+        """Gracefully stop the worker.
+
+        Signals the polling loop to stop, cancels all running jobs,
+        shuts down the thread pool, and waits for the poll thread to
+        finish.
+        """
+        if not self._health.is_running:
+            return
+        logger.info("AsyncWorker %s stopping", self._worker_id)
+        self._shutdown_event.set()
+        self._cancel_all_running_jobs()
+        if self._thread_pool is not None:
+            self._thread_pool.shutdown(wait=True, cancel_futures=True)
+            self._thread_pool = None
+        if self._poll_thread is not None:
+            self._poll_thread.join(timeout=self._config.poll_interval * 3)
+        self._health.is_running = False
+        self._restore_signal_handlers()
+        logger.info("AsyncWorker %s stopped", self._worker_id)
+
+    def request_shutdown(self) -> None:
+        """Request a graceful shutdown (signal-handler safe)."""
+        self._shutdown_event.set()
+
+    def pickup_and_execute(self, job: AsyncJob) -> None:
+        """Pick up a job and execute it.
+
+        This method owns **all** state transitions for the job.
+        External cancellation (via ``cancel_job``) only signals the
+        cancellation token; this method checks the token and performs
+        the appropriate transition.
+
+        Args:
+            job: The job to execute.
+
+        Raises:
+            TypeError: If job is not an AsyncJob.
+        """
+        if not isinstance(job, AsyncJob):
+            raise TypeError(f"job must be AsyncJob, got {type(job)}")
+
+        token = CancellationToken()
+        with self._tokens_lock:
+            self._cancellation_tokens[job.job_id] = token
+
+        try:
+            # If the job was already cancelled before we got here, skip.
+            if job.is_terminal:
+                return
+
+            job.mark_running(self._worker_id)
+            self._job_store.update(job)
+            logger.info("Job %s picked up by %s", job.job_id, self._worker_id)
+
+            # Execute the job
+            if self._job_executor is not None:
+                self._job_executor(job, token)
+
+            # Determine final state — guard against concurrent cancellation
+            # that may have already transitioned this job via another path.
+            if job.is_terminal:
+                # Another thread already moved the job to a terminal state
+                # (should not happen with the new contract, but defensive).
+                logger.info("Job %s already terminal: %s", job.job_id, job.status.value)
+            elif token.is_cancelled:
+                job.mark_cancelled()
+                self._job_store.update(job)
+                self._health.record_job_cancelled()
+                logger.info("Job %s cancelled", job.job_id)
+            else:
+                job.mark_succeeded()
+                self._job_store.update(job)
+                self._health.record_job_completed()
+                logger.info("Job %s succeeded", job.job_id)
+        except InvalidJobTransitionError:
+            logger.warning("Job %s transition conflict (already terminal)", job.job_id)
+        except Exception as exc:
+            error_msg = redact_value(f"{type(exc).__name__}: {exc}")
+            logger.exception("Job %s failed with error", job.job_id)
+            try:
+                job.mark_failed(error=error_msg)
+                self._job_store.update(job)
+            except InvalidJobTransitionError:
+                pass
+            self._health.record_job_failed()
+        finally:
+            with self._tokens_lock:
+                self._cancellation_tokens.pop(job.job_id, None)
+            with self._futures_lock:
+                self._futures.pop(job.job_id, None)
+
+    def cancel_job(self, job_id: str) -> bool:
+        """Cancel a specific job.
+
+        Only signals the cancellation token.  The ``pickup_and_execute``
+        method that owns the job performs the actual state transition,
+        preventing a race where two threads attempt conflicting
+        transitions on the same job object.
+
+        For jobs that are still ``queued`` (not yet picked up), this
+        method directly transitions them to ``cancelled`` since no
+        worker thread owns them yet.
+
+        Args:
+            job_id: The job ULID to cancel.
+
+        Returns:
+            True if the job was cancelled or signalled, False otherwise.
+        """
+        if not job_id:
+            raise ValueError("job_id must be a non-empty string")
+
+        # Signal the cancellation token for running jobs
+        with self._tokens_lock:
+            token = self._cancellation_tokens.get(job_id)
+            if token is not None:
+                token.cancel()
+                # Job is owned by a worker thread — let it handle the
+                # state transition via pickup_and_execute.
+                return True
+
+        # For queued jobs not yet picked up, cancel directly.
+        job = self._job_store.get(job_id)
+        if job is not None and not job.is_terminal:
+            try:
+                job.mark_cancelled()
+                self._job_store.update(job)
+                return True
+            except InvalidJobTransitionError:
+                pass
+        return False
+
+    def detect_stuck_jobs(self) -> list[AsyncJob]:
+        """Detect and fail jobs that have exceeded the heartbeat TTL.
+
+        A job is considered stuck if it is in RUNNING state and its
+        last heartbeat is older than ``job_timeout`` seconds.
+
+        Returns:
+            List of jobs that were marked as failed.
+        """
+        running_jobs = self._job_store.list_by_status(AsyncJobStatus.RUNNING)
+        stuck: list[AsyncJob] = []
+        now = datetime.now(UTC)
+        timeout_delta = timedelta(seconds=self._config.job_timeout)
+
+        for job in running_jobs:
+            heartbeat = job.last_heartbeat or job.started_at or job.created_at
+            if now - heartbeat > timeout_delta:
+                try:
+                    job.mark_failed(error="Heartbeat timeout — job stuck")
+                    self._job_store.update(job)
+                    stuck.append(job)
+                    logger.warning(
+                        "Job %s marked failed: heartbeat timeout", job.job_id
+                    )
+                except InvalidJobTransitionError:
+                    pass
+        return stuck
+
+    def cleanup_completed_jobs(self) -> int:
+        """Remove completed jobs older than the retention TTL.
+
+        Delegates to ``InMemoryJobStore.remove_expired`` for a
+        single-pass, single-lock removal.
+
+        Returns:
+            Number of jobs cleaned up.
+        """
+        cleaned = self._job_store.remove_expired(self._config.job_ttl)
+        if cleaned > 0:
+            logger.debug("Cleaned up %d expired job(s)", cleaned)
+        return cleaned
+
+    def get_health_report(self) -> dict[str, Any]:
+        """Return a health report for diagnostics.
+
+        Uses ``snapshot_counts`` on the job store to obtain all queue
+        depth metrics under a single lock acquisition for consistency.
+
+        Returns:
+            Dictionary with worker health information.
+        """
+        self._health.record_heartbeat()
+        report = self._health.as_dict()
+        report["config"] = {
+            "enabled": self._config.enabled,
+            "max_workers": self._config.max_workers,
+            "poll_interval": self._config.poll_interval,
+            "job_timeout": self._config.job_timeout,
+            "job_ttl": self._config.job_ttl,
+        }
+        counts = self._job_store.snapshot_counts()
+        report["queued_jobs"] = counts["queued"]
+        report["running_jobs"] = counts["running"]
+        report["total_jobs"] = counts["total"]
+        return report
+
+    # -- Private methods ----------------------------------------------------
+
+    def _poll_loop(self) -> None:
+        """Main polling loop that dispatches queued jobs to the thread pool.
+
+        Each job is submitted to the ``ThreadPoolExecutor`` for truly
+        concurrent execution up to ``max_workers``.
+        """
+        while not self._shutdown_event.is_set():
+            try:
+                self._health.record_heartbeat()
+                # Determine available capacity in the pool
+                with self._futures_lock:
+                    active = sum(1 for f in self._futures.values() if not f.done())
+                capacity = self._config.max_workers - active
+                if capacity > 0:
+                    queued = self._job_store.list_by_status(AsyncJobStatus.QUEUED)
+                    for job in queued[:capacity]:
+                        if self._shutdown_event.is_set():
+                            break
+                        self._dispatch_job(job)
+            except Exception:
+                logger.exception("Error in poll loop")
+            self._shutdown_event.wait(timeout=self._config.poll_interval)
+
+    def _dispatch_job(self, job: AsyncJob) -> None:
+        """Submit a job to the thread pool for concurrent execution."""
+        if self._thread_pool is None:
+            # Fallback for direct pickup_and_execute calls (e.g., tests)
+            self.pickup_and_execute(job)
+            return
+        with self._futures_lock:
+            if job.job_id in self._futures:
+                return  # Already dispatched
+            future: Future[None] = self._thread_pool.submit(
+                self.pickup_and_execute, job
+            )
+            self._futures[job.job_id] = future
+
+    def _cancel_all_running_jobs(self) -> None:
+        """Cancel all currently running jobs."""
+        with self._tokens_lock:
+            for token in self._cancellation_tokens.values():
+                token.cancel()
+
+    def _signal_handler(self, signum: int, _frame: Any) -> None:
+        """Handle shutdown signals."""
+        logger.info("Received signal %d, requesting shutdown", signum)
+        self.request_shutdown()
+
+    def _install_signal_handlers(self) -> None:
+        """Install signal handlers for graceful shutdown."""
+        try:
+            self._original_sigint = signal.getsignal(signal.SIGINT)
+            self._original_sigterm = signal.getsignal(signal.SIGTERM)
+            signal.signal(signal.SIGINT, self._signal_handler)
+            signal.signal(signal.SIGTERM, self._signal_handler)
+        except (OSError, ValueError):
+            # Cannot set signal handlers from non-main thread
+            pass
+
+    def _restore_signal_handlers(self) -> None:
+        """Restore original signal handlers."""
+        try:
+            if self._original_sigint is not None:
+                signal.signal(signal.SIGINT, self._original_sigint)
+            if self._original_sigterm is not None:
+                signal.signal(signal.SIGTERM, self._original_sigterm)
+        except (OSError, ValueError):
+            pass

src/cleveragents/application/services/plan_lifecycle_service.py

@@ -64,6 +64,7 @@ from cleveragents.core.exceptions import (
     ValidationError,
 )
 from cleveragents.domain.models.core.action import Action, ActionArgument, ActionState
+from cleveragents.domain.models.core.async_job import AsyncJob, serialize_job_payload
 from cleveragents.domain.models.core.automation_profile import (
     BUILTIN_PROFILES,
     AutomationProfile,
@@ -82,6 +83,7 @@ from cleveragents.domain.models.core.plan import (
 )
 
 if TYPE_CHECKING:
+    from cleveragents.application.services.async_worker import InMemoryJobStore
     from cleveragents.application.services.decision_service import DecisionService
     from cleveragents.config.settings import Settings
     from cleveragents.infrastructure.database.unit_of_work import UnitOfWork
@@ -149,6 +151,7 @@ class PlanLifecycleService:
         settings: Settings,
         unit_of_work: UnitOfWork | None = None,
         decision_service: DecisionService | None = None,
+        job_store: InMemoryJobStore | None = None,
     ):
         """Initialize the plan lifecycle service.
 
@@ -163,10 +166,15 @@ class PlanLifecycleService:
                 automatically recording decisions during phase
                 transitions.  When ``None``, decision recording is
                 silently skipped.
+            job_store: Optional :class:`InMemoryJobStore` for enqueuing
+                async jobs when ``settings.async_enabled`` is True.
+                When ``None``, async job creation is silently skipped
+                even if async is enabled.
         """
         self.settings = settings
         self.unit_of_work = unit_of_work
         self.decision_service = decision_service
+        self._job_store = job_store
         self._logger = logger.bind(service="plan_lifecycle")
 
         # In-memory fallback storage (used only when no UoW is provided)
@@ -256,6 +264,44 @@ class PlanLifecycleService:
             with self.unit_of_work.transaction() as ctx:
                 self._persist_plan_update(plan, ctx)
 
+    def _maybe_enqueue_async_job(self, plan_id: str, phase: str) -> AsyncJob | None:
+        """Create and enqueue an async job if async execution is enabled.
+
+        Checks ``settings.async_enabled`` and the presence of a job
+        store.  When both conditions are met, creates an ``AsyncJob``
+        for the given plan phase and adds it to the store so the
+        ``AsyncWorker`` poll loop can pick it up.
+
+        Args:
+            plan_id: The plan ULID.
+            phase: The plan phase (``"execute"`` or ``"apply"``).
+
+        Returns:
+            The created ``AsyncJob`` if enqueued, or ``None`` if async
+            is disabled or no job store is configured.
+        """
+        if not self.settings.async_enabled:
+            return None
+        if self._job_store is None:
+            return None
+
+        job_id = self._generate_ulid()
+        payload = serialize_job_payload(plan_id, phase)
+        job = AsyncJob(
+            job_id=job_id,
+            plan_id=plan_id,
+            phase=phase,
+            payload_json=payload,
+        )
+        self._job_store.add(job)
+        self._logger.info(
+            "Async job enqueued",
+            job_id=job_id,
+            plan_id=plan_id,
+            phase=phase,
+        )
+        return job
+
     def update_error_details(
         self,
         plan_id: str,
@@ -831,6 +877,9 @@ class PlanLifecycleService:
             phase=plan.phase.value,
         )
 
+        # Enqueue async job when async execution is enabled
+        self._maybe_enqueue_async_job(plan_id, "execute")
+
         return plan
 
     def start_execute(self, plan_id: str) -> Plan:
@@ -938,6 +987,9 @@ class PlanLifecycleService:
             phase=plan.phase.value,
         )
 
+        # Enqueue async job when async execution is enabled
+        self._maybe_enqueue_async_job(plan_id, "apply")
+
         return plan
 
     def start_apply(self, plan_id: str) -> Plan:

src/cleveragents/cli/commands/system.py

@@ -389,6 +389,35 @@ def _check_stale_locks() -> dict[str, Any]:
         }
 
 
+def _check_async_worker_health() -> dict[str, Any]:
+    """Check async worker configuration and health."""
+    try:
+        from cleveragents.config.settings import get_settings
+
+        settings = get_settings()
+        enabled = settings.async_enabled
+        if not enabled:
+            return {
+                "name": "Async workers",
+                "status": CheckStatus.OK,
+                "details": "disabled (async.enabled=false)",
+            }
+        return {
+            "name": "Async workers",
+            "status": CheckStatus.OK,
+            "details": (
+                f"enabled, max_workers={settings.async_max_workers}, "
+                f"poll_interval={settings.async_poll_interval}s"
+            ),
+        }
+    except Exception:
+        return {
+            "name": "Async workers",
+            "status": CheckStatus.WARN,
+            "details": "unable to check",
+        }
+
+
 def build_diagnostics_data() -> dict[str, Any]:
     """Run all diagnostic checks and return structured results."""
     start = time.monotonic()
@@ -403,6 +432,8 @@ def build_diagnostics_data() -> dict[str, Any]:
     checks.append(_check_git())
     checks.append(_check_stale_locks())
 
+    checks.append(_check_async_worker_health())
+
     elapsed = time.monotonic() - start
 
     total = len(checks)

src/cleveragents/config/settings.py

@@ -404,6 +404,37 @@ class Settings(BaseSettings):
         description="Minimum files below which decomposition is skipped.",
     )
 
+    # Async job execution (M6 - #312)
+    async_enabled: bool = Field(
+        default=False,
+        validation_alias=AliasChoices("CLEVERAGENTS_ASYNC_ENABLED"),
+        description="Enable asynchronous job execution for plan phases.",
+    )
+    async_max_workers: int = Field(
+        default=4,
+        ge=1,
+        validation_alias=AliasChoices("CLEVERAGENTS_ASYNC_MAX_WORKERS"),
+        description="Maximum concurrent async workers.",
+    )
+    async_poll_interval: float = Field(
+        default=1.0,
+        gt=0,
+        validation_alias=AliasChoices("CLEVERAGENTS_ASYNC_POLL_INTERVAL"),
+        description="Polling interval in seconds for async workers.",
+    )
+    async_job_timeout: int = Field(
+        default=3600,
+        ge=1,
+        validation_alias=AliasChoices("CLEVERAGENTS_ASYNC_JOB_TIMEOUT"),
+        description="Job timeout in seconds (mark failed after no heartbeat).",
+    )
+    async_job_ttl: int = Field(
+        default=86400,
+        ge=1,
+        validation_alias=AliasChoices("CLEVERAGENTS_ASYNC_JOB_TTL"),
+        description="Completed job retention in seconds before cleanup.",
+    )
+
     # Mock providers flag (M4 - provider fixes)
     mock_providers: bool = Field(
         default=False,

src/cleveragents/domain/models/core/__init__.py

@@ -1,6 +1,20 @@
 # Action model (ActionState lives here now)
 from cleveragents.domain.models.core.action import ActionState
 from cleveragents.domain.models.core.actor import Actor
+from cleveragents.domain.models.core.async_job import (
+    TERMINAL_STATUSES as ASYNC_TERMINAL_STATUSES,
+)
+
+# Async job domain model (M6 — #312)
+from cleveragents.domain.models.core.async_job import (
+    VALID_JOB_TRANSITIONS,
+    AsyncJob,
+    AsyncJobStatus,
+    InvalidJobTransitionError,
+    can_transition_job,
+    deserialize_job_payload,
+    serialize_job_payload,
+)
 from cleveragents.domain.models.core.automation_profile import (
     BUILTIN_PROFILES,
     AutomationGuard,
@@ -273,13 +287,17 @@ from cleveragents.domain.models.core.uko import (
 )
 
 __all__ = [
+    "ASYNC_TERMINAL_STATUSES",
     "BUILTIN_PROFILES",
     "DEFAULT_LOCAL_ROLE_MAPPING",
     "DEFAULT_SAFETY_PROFILE",
     "ROLE_PERMISSIONS",
+    "VALID_JOB_TRANSITIONS",
     "ActionState",
     "Actor",
     "ActorLimits",
+    "AsyncJob",
+    "AsyncJobStatus",
     "AutomationGuard",
     "AutomationProfile",
     "AutonomyGuardrails",
@@ -338,6 +356,7 @@ __all__ = [
     "HistoricalOutcome",
     "InMemoryChangeSetStore",
     "InMemoryInvocationTracker",
+    "InvalidJobTransitionError",
     "Invariant",
     "InvariantEnforcementRecord",
     "InvariantScope",
@@ -442,7 +461,9 @@ __all__ = [
     "Validation",
     "ValidationMode",
     "can_transition",
+    "can_transition_job",
     "classify_error",
+    "deserialize_job_payload",
     "get_builtin_profile",
     "get_recovery_hints",
     "merge_invariants",
@@ -451,4 +472,5 @@ __all__ = [
     "parse_namespaced_name",
     "render_dod_template",
     "resolve_safety_profile",
+    "serialize_job_payload",
 ]

src/cleveragents/domain/models/core/async_job.py

@@ -0,0 +1,379 @@
+"""AsyncJob domain model for CleverAgents.
+
+Represents an asynchronous job that executes plan phases in background
+workers.  Jobs follow a strict state machine:
+
+```
+queued -> running -> succeeded
+                  -> failed
+                  -> cancelled
+```
+
+Only the transitions listed above are valid.  Attempting an invalid
+transition raises ``InvalidJobTransitionError``.
+
+## Fields
+
+| Field            | Type              | Description                           |
+|------------------|-------------------|---------------------------------------|
+| ``job_id``       | ``str`` (ULID)    | Unique job identifier                 |
+| ``plan_id``      | ``str`` (ULID)    | Plan this job belongs to              |
+| ``phase``        | ``str``           | Plan phase (execute / apply)          |
+| ``status``       | ``AsyncJobStatus``| Current job status                    |
+| ``payload_json`` | ``str``           | Serialized job payload                |
+| ``created_at``   | ``datetime``      | When the job was enqueued             |
+| ``started_at``   | ``datetime|None`` | When a worker picked it up            |
+| ``finished_at``  | ``datetime|None`` | When execution completed              |
+| ``worker_id``    | ``str|None``      | ID of the worker processing this job  |
+| ``last_heartbeat``| ``datetime|None``| Last heartbeat from the worker        |
+| ``schema_version``| ``int``          | Payload schema version                |
+
+Based on issue #312 (feat(async): add async command execution and workers).
+"""
+
+from __future__ import annotations
+
+import json
+import re
+from datetime import UTC, datetime
+from enum import StrEnum
+from typing import Any
+
+from pydantic import BaseModel, ConfigDict, Field, field_validator
+
+# ULID is 26 characters, Crockford's base32
+ULID_PATTERN = r"^[0-9A-HJKMNP-TV-Z]{26}$"
+_ULID_RE = re.compile(ULID_PATTERN)
+
+# Valid phases for async jobs
+VALID_PHASES = frozenset({"execute", "apply"})
+
+
+class AsyncJobStatus(StrEnum):
+    """Status of an asynchronous job.
+
+    Jobs follow a strict state machine:
+    ``queued`` -> ``running`` -> ``succeeded`` / ``failed`` / ``cancelled``
+    """
+
+    QUEUED = "queued"
+    RUNNING = "running"
+    SUCCEEDED = "succeeded"
+    FAILED = "failed"
+    CANCELLED = "cancelled"
+
+
+# Valid status transitions: from -> set of allowed targets
+VALID_JOB_TRANSITIONS: dict[AsyncJobStatus, frozenset[AsyncJobStatus]] = {
+    AsyncJobStatus.QUEUED: frozenset(
+        {AsyncJobStatus.RUNNING, AsyncJobStatus.CANCELLED}
+    ),
+    AsyncJobStatus.RUNNING: frozenset(
+        {AsyncJobStatus.SUCCEEDED, AsyncJobStatus.FAILED, AsyncJobStatus.CANCELLED}
+    ),
+    AsyncJobStatus.SUCCEEDED: frozenset(),
+    AsyncJobStatus.FAILED: frozenset(),
+    AsyncJobStatus.CANCELLED: frozenset(),
+}
+
+# Terminal states
+TERMINAL_STATUSES: frozenset[AsyncJobStatus] = frozenset(
+    {AsyncJobStatus.SUCCEEDED, AsyncJobStatus.FAILED, AsyncJobStatus.CANCELLED}
+)
+
+
+class InvalidJobTransitionError(Exception):
+    """Raised when an invalid job status transition is attempted."""
+
+    def __init__(self, from_status: AsyncJobStatus, to_status: AsyncJobStatus) -> None:
+        self.from_status = from_status
+        self.to_status = to_status
+        super().__init__(
+            f"Invalid job transition: {from_status.value} -> {to_status.value}"
+        )
+
+
+def can_transition_job(from_status: AsyncJobStatus, to_status: AsyncJobStatus) -> bool:
+    """Check if a job status transition is valid.
+
+    Args:
+        from_status: Current job status.
+        to_status: Target job status.
+
+    Returns:
+        True if the transition is valid.
+    """
+    if not isinstance(from_status, AsyncJobStatus):
+        raise TypeError(f"from_status must be AsyncJobStatus, got {type(from_status)}")
+    if not isinstance(to_status, AsyncJobStatus):
+        raise TypeError(f"to_status must be AsyncJobStatus, got {type(to_status)}")
+    return to_status in VALID_JOB_TRANSITIONS.get(from_status, frozenset())
+
+
+class AsyncJob(BaseModel):
+    """Domain model for an asynchronous job.
+
+    An async job represents a unit of work (a plan phase) that is executed
+    by a background worker.  Jobs are created when ``async.enabled`` is True
+    and a plan transitions to the execute or apply phase.
+
+    The ``payload_json`` field contains a JSON-serialized payload with the
+    data needed to execute the job.  The ``schema_version`` field tracks
+    the payload format version for forward compatibility.
+    """
+
+    job_id: str = Field(
+        ...,
+        description="Unique ULID identifier for this job",
+        pattern=ULID_PATTERN,
+    )
+    plan_id: str = Field(
+        ...,
+        description="ULID of the plan this job belongs to",
+        pattern=ULID_PATTERN,
+    )
+    phase: str = Field(
+        ...,
+        description="Plan phase this job executes (execute or apply)",
+    )
+    status: AsyncJobStatus = Field(
+        default=AsyncJobStatus.QUEUED,
+        description="Current job status",
+    )
+    payload_json: str = Field(
+        default="{}",
+        description="JSON-serialized job payload",
+    )
+    created_at: datetime = Field(
+        default_factory=lambda: datetime.now(UTC),
+        description="When the job was enqueued",
+    )
+    started_at: datetime | None = Field(
+        default=None,
+        description="When a worker picked up this job",
+    )
+    finished_at: datetime | None = Field(
+        default=None,
+        description="When execution completed",
+    )
+    worker_id: str | None = Field(
+        default=None,
+        description="ID of the worker processing this job",
+    )
+    last_heartbeat: datetime | None = Field(
+        default=None,
+        description="Last heartbeat timestamp from the worker",
+    )
+    schema_version: int = Field(
+        default=1,
+        ge=1,
+        description="Payload schema version for forward compatibility",
+    )
+    error_message: str | None = Field(
+        default=None,
+        description="Error details when the job fails (for audit trail)",
+    )
+
+    @field_validator("phase")
+    @classmethod
+    def validate_phase(cls: type[AsyncJob], v: str) -> str:
+        """Validate that phase is a valid async job phase."""
+        if v not in VALID_PHASES:
+            raise ValueError(f"phase must be one of {sorted(VALID_PHASES)}, got {v!r}")
+        return v
+
+    @field_validator("payload_json")
+    @classmethod
+    def validate_payload_json(cls: type[AsyncJob], v: str) -> str:
+        """Validate that payload_json is valid JSON."""
+        try:
+            json.loads(v)
+        except (json.JSONDecodeError, TypeError) as exc:
+            raise ValueError(f"payload_json must be valid JSON: {exc}") from exc
+        return v
+
+    # -- Properties ---------------------------------------------------------
+
+    @property
+    def is_terminal(self) -> bool:
+        """Check if the job is in a terminal state."""
+        return self.status in TERMINAL_STATUSES
+
+    @property
+    def is_running(self) -> bool:
+        """Check if the job is currently running."""
+        return self.status == AsyncJobStatus.RUNNING
+
+    @property
+    def payload(self) -> dict[str, Any]:
+        """Deserialize and return the job payload."""
+        return dict(json.loads(self.payload_json))
+
+    # -- Methods ------------------------------------------------------------
+
+    def transition_to(self, target: AsyncJobStatus) -> None:
+        """Transition the job to a new status.
+
+        Validates that the transition is allowed by the state machine
+        before applying the change.
+
+        Args:
+            target: The target status.
+
+        Raises:
+            TypeError: If target is not an AsyncJobStatus.
+            InvalidJobTransitionError: If the transition is not allowed.
+        """
+        if not isinstance(target, AsyncJobStatus):
+            raise TypeError(f"target must be AsyncJobStatus, got {type(target)}")
+        if not can_transition_job(self.status, target):
+            raise InvalidJobTransitionError(self.status, target)
+        self.status = target
+
+    def mark_running(self, worker_id: str) -> None:
+        """Mark the job as running by a specific worker.
+
+        Args:
+            worker_id: ID of the worker picking up this job.
+
+        Raises:
+            ValueError: If worker_id is empty.
+            InvalidJobTransitionError: If the job is not in QUEUED state.
+        """
+        if not worker_id or not worker_id.strip():
+            raise ValueError("worker_id must be a non-empty string")
+        self.transition_to(AsyncJobStatus.RUNNING)
+        self.worker_id = worker_id
+        self.started_at = datetime.now(UTC)
+        self.last_heartbeat = self.started_at
+
+    def mark_succeeded(self) -> None:
+        """Mark the job as successfully completed.
+
+        Raises:
+            InvalidJobTransitionError: If the job is not in RUNNING state.
+        """
+        self.transition_to(AsyncJobStatus.SUCCEEDED)
+        self.finished_at = datetime.now(UTC)
+
+    def mark_failed(self, error: str | None = None) -> None:
+        """Mark the job as failed.
+
+        Args:
+            error: Optional error message for audit trail.
+
+        Raises:
+            InvalidJobTransitionError: If the job is not in RUNNING state.
+        """
+        self.transition_to(AsyncJobStatus.FAILED)
+        self.finished_at = datetime.now(UTC)
+        if error:
+            self.error_message = error
+
+    def mark_cancelled(self) -> None:
+        """Mark the job as cancelled.
+
+        Raises:
+            InvalidJobTransitionError: If the job is already terminal.
+        """
+        self.transition_to(AsyncJobStatus.CANCELLED)
+        self.finished_at = datetime.now(UTC)
+
+    def record_heartbeat(self) -> None:
+        """Record a heartbeat from the worker.
+
+        Raises:
+            ValueError: If the job is not currently running.
+        """
+        if self.status != AsyncJobStatus.RUNNING:
+            raise ValueError(
+                f"Cannot record heartbeat for job in {self.status.value} state"
+            )
+        self.last_heartbeat = datetime.now(UTC)
+
+    def as_cli_dict(self) -> dict[str, Any]:
+        """Return a stable dictionary representation for CLI output."""
+        result: dict[str, Any] = {
+            "job_id": self.job_id,
+            "plan_id": self.plan_id,
+            "phase": self.phase,
+            "status": self.status.value,
+            "schema_version": self.schema_version,
+            "created_at": self.created_at.isoformat(),
+        }
+        if self.started_at:
+            result["started_at"] = self.started_at.isoformat()
+        if self.finished_at:
+            result["finished_at"] = self.finished_at.isoformat()
+        if self.worker_id:
+            result["worker_id"] = self.worker_id
+        if self.last_heartbeat:
+            result["last_heartbeat"] = self.last_heartbeat.isoformat()
+        if self.error_message:
+            result["error_message"] = self.error_message
+        return result
+
+    model_config = ConfigDict(
+        str_strip_whitespace=True,
+        validate_assignment=True,
+        use_enum_values=False,
+    )
+
+
+def serialize_job_payload(
+    plan_id: str,
+    phase: str,
+    extra: dict[str, Any] | None = None,
+    schema_version: int = 1,
+) -> str:
+    """Serialize a job payload to JSON with schema versioning.
+
+    Args:
+        plan_id: The plan ULID.
+        phase: The plan phase.
+        extra: Additional payload data.
+        schema_version: Schema version for forward compatibility.
+
+    Returns:
+        JSON string of the serialized payload.
+
+    Raises:
+        ValueError: If plan_id or phase is empty.
+    """
+    if not plan_id or not plan_id.strip():
+        raise ValueError("plan_id must be a non-empty string")
+    if not phase or not phase.strip():
+        raise ValueError("phase must be a non-empty string")
+    if schema_version < 1:
+        raise ValueError("schema_version must be >= 1")
+    payload: dict[str, Any] = {
+        "schema_version": schema_version,
+        "plan_id": plan_id,
+        "phase": phase,
+    }
+    if extra:
+        payload["extra"] = extra
+    return json.dumps(payload, default=str)
+
+
+def deserialize_job_payload(payload_json: str) -> dict[str, Any]:
+    """Deserialize a job payload from JSON.
+
+    Args:
+        payload_json: The JSON string to deserialize.
+
+    Returns:
+        The deserialized payload dictionary.
+
+    Raises:
+        ValueError: If the JSON is invalid or missing required fields.
+    """
+    if not payload_json or not payload_json.strip():
+        raise ValueError("payload_json must be a non-empty string")
+    try:
+        data: dict[str, Any] = json.loads(payload_json)
+    except (json.JSONDecodeError, TypeError) as exc:
+        raise ValueError(f"Invalid JSON payload: {exc}") from exc
+    if not isinstance(data, dict):
+        raise ValueError("Payload must be a JSON object")
+    return data

src/cleveragents/infrastructure/database/models.py

@@ -3028,3 +3028,67 @@ class LLMTraceModel(Base):  # type: ignore[misc]
         Index("ix_llm_traces_actor", "actor"),
         Index("ix_llm_traces_provider", "provider"),
     )
+
+
+# ---------------------------------------------------------------------------
+# Async Job Models (M6 - async command execution and workers, #312)
+# ---------------------------------------------------------------------------
+
+
+class AsyncJobModel(Base):  # type: ignore[misc]
+    """Database model for async jobs.
+
+    Stores background job metadata for asynchronous plan phase execution.
+    Jobs follow a strict state machine:
+    ``queued -> running -> succeeded / failed / cancelled``
+
+    Table: ``async_jobs``
+    """
+
+    __allow_unmapped__ = True
+    __tablename__ = "async_jobs"
+
+    # PK: ULID (26-char string)
+    job_id = Column(String(26), primary_key=True)
+
+    # Logical reference to v3_plans (no FK constraint — jobs may outlive plans)
+    plan_id = Column(String(26), nullable=False)
+
+    # Plan phase this job executes
+    phase = Column(String(20), nullable=False)
+
+    # Job status
+    status = Column(String(20), nullable=False, default="queued")
+
+    # Serialized job payload (JSON)
+    payload_json = Column(Text, nullable=False, default="{}")
+
+    # Timestamps (ISO-8601 strings)
+    created_at = Column(String(30), nullable=False)
+    started_at = Column(String(30), nullable=True)
+    finished_at = Column(String(30), nullable=True)
+
+    # Worker assignment
+    worker_id = Column(String(255), nullable=True)
+    last_heartbeat = Column(String(30), nullable=True)
+
+    # Payload schema version for forward compatibility
+    schema_version = Column(Integer, nullable=False, default=1)
+
+    # Error details for audit trail (NULL when not failed)
+    error_message = Column(Text, nullable=True)
+
+    __table_args__ = (
+        CheckConstraint(
+            "status IN ('queued', 'running', 'succeeded', 'failed', 'cancelled')",
+            name="ck_async_jobs_status",
+        ),
+        CheckConstraint(
+            "phase IN ('execute', 'apply')",
+            name="ck_async_jobs_phase",
+        ),
+        Index("ix_async_jobs_plan_id", "plan_id"),
+        Index("ix_async_jobs_status", "status"),
+        Index("ix_async_jobs_worker_id", "worker_id"),
+        Index("ix_async_jobs_created_at", "created_at"),
+    )

vulture_whitelist.py

@@ -736,3 +736,51 @@ ReconciliationResult  # noqa: B018, F821
 ConflictRecord  # noqa: B018, F821
 ScopeInvariants  # noqa: B018, F821
 reconcile_invariants  # noqa: B018, F821
+
+# Async job execution and workers — public API (issue #312)
+AsyncJobStatus  # noqa: B018, F821
+AsyncJob  # noqa: B018, F821
+InvalidJobTransitionError  # noqa: B018, F821
+VALID_JOB_TRANSITIONS  # noqa: B018, F821
+TERMINAL_STATUSES  # noqa: B018, F821
+VALID_PHASES  # noqa: B018, F821
+ULID_PATTERN  # noqa: B018, F821
+can_transition_job  # noqa: B018, F821
+serialize_job_payload  # noqa: B018, F821
+deserialize_job_payload  # noqa: B018, F821
+AsyncWorker  # noqa: B018, F821
+AsyncWorkerConfig  # noqa: B018, F821
+InMemoryJobStore  # noqa: B018, F821
+CancellationToken  # noqa: B018, F821
+WorkerHealthReport  # noqa: B018, F821
+is_terminal  # noqa: B018, F821
+is_running  # noqa: B018, F821
+payload  # noqa: B018, F821
+transition_to  # noqa: B018, F821
+mark_running  # noqa: B018, F821
+mark_succeeded  # noqa: B018, F821
+mark_failed  # noqa: B018, F821
+mark_cancelled  # noqa: B018, F821
+record_heartbeat  # noqa: B018, F821
+as_cli_dict  # noqa: B018, F821
+pickup_and_execute  # noqa: B018, F821
+cancel_job  # noqa: B018, F821
+detect_stuck_jobs  # noqa: B018, F821
+cleanup_completed_jobs  # noqa: B018, F821
+get_health_report  # noqa: B018, F821
+request_shutdown  # noqa: B018, F821
+list_by_status  # noqa: B018, F821
+snapshot_counts  # noqa: B018, F821
+remove_expired  # noqa: B018, F821
+record_job_cancelled  # noqa: B018, F821
+jobs_cancelled  # noqa: B018, F821
+error_message  # noqa: B018, F821
+_check_async_worker_health  # noqa: B018, F821
+async_enabled  # noqa: B018, F821
+async_max_workers  # noqa: B018, F821
+async_poll_interval  # noqa: B018, F821
+async_job_timeout  # noqa: B018, F821
+async_job_ttl  # noqa: B018, F821
+validate_phase  # noqa: B018, F821
+validate_payload_json  # noqa: B018, F821
+AsyncJobModel  # noqa: B018, F821