feat(async): add async command execution and workers

- Add AsyncJob domain model with status state machine and Pydantic validation
- Add AsyncWorker service with configurable concurrency and job store
- Add CancellationToken, WorkerHealthReport, InMemoryJobStore
- Add AsyncJobModel SQLAlchemy model and Alembic migration (m6_003)
- Add 5 async config keys to Settings (worker_id, concurrency, poll_interval, max_retries, timeout)
- Add _check_async_worker_health diagnostic check in system.py
- Add comprehensive Behave BDD tests (~60 scenarios) with full step definitions
- Add Robot Framework integration tests (6 smoke tests)
- Add ASV benchmark suite for async execution
- Add architecture documentation
- Update vulture_whitelist with new public API symbols
- All quality gates pass: lint, typecheck, unit_tests, integration_tests, coverage_report (97%)

1. Wire async job creation into PlanLifecycleService:
   - Add optional job_store parameter to __init__
   - Add _maybe_enqueue_async_job() helper that checks settings.async_enabled
     and job store presence before creating and enqueuing an AsyncJob
   - Call helper from execute_plan() (phase="execute") and apply_plan()
     (phase="apply") after phase transitions
   - When async is disabled or no job store is configured, behaviour is
     unchanged (silent no-op)

2. Redact secrets in failed job error messages:
   - Apply shared.redaction.redact_value() to the error string before
     persisting to AsyncJob.error_message, preventing accidental secret
     leakage (e.g. API keys in exception text) into the audit trail

Documentation:
- S1: Added specification reconciliation note (ADR-style) to
  async_architecture.md addressing tension between "No Plan Queuing"
  clause and the async subsystem authorised by issue #312

ISSUES CLOSED: #312

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)