From 837ff4217b81e63e59b806457be4a264488feeed Mon Sep 17 00:00:00 2001 From: CoreRasurae Date: Wed, 4 Mar 2026 01:45:38 +0000 Subject: [PATCH] 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 --- CHANGELOG.md | 13 + alembic/versions/m6_003_async_jobs_table.py | 58 + benchmarks/async_execution_bench.py | 338 ++++ docs/reference/async_architecture.md | 255 +++ features/async_execution.feature | 555 ++++++ features/steps/async_execution_steps.py | 1591 +++++++++++++++++ features/steps/decision_di_wiring_steps.py | 1 + robot/async_execution.robot | 52 + robot/helper_async_execution.py | 198 ++ robot/helper_decision_di.py | 1 + .../application/services/async_worker.py | 654 +++++++ .../services/plan_lifecycle_service.py | 52 + src/cleveragents/cli/commands/system.py | 31 + src/cleveragents/config/settings.py | 31 + .../domain/models/core/__init__.py | 22 + .../domain/models/core/async_job.py | 379 ++++ .../infrastructure/database/models.py | 64 + vulture_whitelist.py | 48 + 18 files changed, 4343 insertions(+) create mode 100644 alembic/versions/m6_003_async_jobs_table.py create mode 100644 benchmarks/async_execution_bench.py create mode 100644 docs/reference/async_architecture.md create mode 100644 features/async_execution.feature create mode 100644 features/steps/async_execution_steps.py create mode 100644 robot/async_execution.robot create mode 100644 robot/helper_async_execution.py create mode 100644 src/cleveragents/application/services/async_worker.py create mode 100644 src/cleveragents/domain/models/core/async_job.py diff --git a/CHANGELOG.md b/CHANGELOG.md index db650c3f5..d789f65c4 100644 --- a/CHANGELOG.md +++ b/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 diff --git a/alembic/versions/m6_003_async_jobs_table.py b/alembic/versions/m6_003_async_jobs_table.py new file mode 100644 index 000000000..9a6443b9b --- /dev/null +++ b/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") diff --git a/benchmarks/async_execution_bench.py b/benchmarks/async_execution_bench.py new file mode 100644 index 000000000..36cee354d --- /dev/null +++ b/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) diff --git a/docs/reference/async_architecture.md b/docs/reference/async_architecture.md new file mode 100644 index 000000000..8dc410923 --- /dev/null +++ b/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 ` +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-`) +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 | diff --git a/features/async_execution.feature b/features/async_execution.feature new file mode 100644 index 000000000..e51f85d03 --- /dev/null +++ b/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***" diff --git a/features/steps/async_execution_steps.py b/features/steps/async_execution_steps.py new file mode 100644 index 000000000..efefa0158 --- /dev/null +++ b/features/steps/async_execution_steps.py @@ -0,0 +1,1591 @@ +"""Step definitions for async execution tests.""" + +from __future__ import annotations + +import json +import threading +import time +from datetime import UTC, datetime, timedelta +from unittest.mock import MagicMock + +from behave import given, then, when +from behave.runner import Context +from pydantic import ValidationError +from ulid import ULID + +from cleveragents.application.services.async_worker import ( + AsyncWorker, + AsyncWorkerConfig, + CancellationToken, + InMemoryJobStore, + WorkerHealthReport, +) +from cleveragents.application.services.plan_lifecycle_service import ( + PlanLifecycleService, +) +from cleveragents.domain.models.core.async_job import ( + AsyncJob, + AsyncJobStatus, + InvalidJobTransitionError, + can_transition_job, + deserialize_job_payload, + serialize_job_payload, +) + + +def _make_job_id() -> str: + return str(ULID()) + + +PLAN_ID = "01HXYZ01HXYZ01HXYZ01HXYZ01" + + +# --------------------------------------------------------------------------- +# Given steps +# --------------------------------------------------------------------------- + + +@given("an async job store is initialised") +def step_async_job_store_initialised(context: Context) -> None: + context.job_store = InMemoryJobStore() + context.async_job = None + context.error = None + + +@given('I create a new async job for plan "{plan_id}" phase "{phase}"') +def step_create_async_job(context: Context, plan_id: str, phase: str) -> None: + job = AsyncJob( + job_id=_make_job_id(), + plan_id=plan_id, + phase=phase, + ) + context.job_store.add(job) + context.async_job = job + + +@given("I create a running async job") +def step_create_running_async_job(context: Context) -> None: + job = AsyncJob( + job_id=_make_job_id(), + plan_id=PLAN_ID, + phase="execute", + ) + context.job_store.add(job) + job.mark_running("worker-test") + context.job_store.update(job) + context.async_job = job + + +@given("I create {count:d} queued async jobs") +def step_create_n_queued_jobs(context: Context, count: int) -> None: + context.created_jobs = [] + for _ in range(count): + job = AsyncJob( + job_id=_make_job_id(), + plan_id=PLAN_ID, + phase="execute", + ) + context.job_store.add(job) + context.created_jobs.append(job) + + +@given("an async worker is initialised with max_workers {max_workers:d}") +def step_async_worker_initialised(context: Context, max_workers: int) -> None: + config = AsyncWorkerConfig( + enabled=True, + max_workers=max_workers, + poll_interval=0.1, + job_timeout=3600, + job_ttl=86400, + ) + context.job_store = InMemoryJobStore() + context.worker = AsyncWorker(config, context.job_store) + context.async_job = None + context.error = None + + +@given("an async worker is initialised with a failing executor") +def step_async_worker_failing_executor(context: Context) -> None: + def failing_executor(job: AsyncJob, token: CancellationToken) -> None: + raise RuntimeError("Execution failed") + + config = AsyncWorkerConfig( + enabled=True, + max_workers=2, + poll_interval=0.1, + job_timeout=3600, + job_ttl=86400, + ) + context.job_store = InMemoryJobStore() + context.worker = AsyncWorker(config, context.job_store, failing_executor) + context.async_job = None + context.error = None + + +@given("an async worker is initialised with job_timeout {timeout:d}") +def step_async_worker_timeout(context: Context, timeout: int) -> None: + config = AsyncWorkerConfig( + enabled=True, + max_workers=2, + poll_interval=0.1, + job_timeout=timeout, + job_ttl=86400, + ) + context.job_store = InMemoryJobStore() + context.worker = AsyncWorker(config, context.job_store) + context.async_job = None + + +@given("an async worker is initialised with job_ttl {ttl:d}") +def step_async_worker_ttl(context: Context, ttl: int) -> None: + config = AsyncWorkerConfig( + enabled=True, + max_workers=2, + poll_interval=0.1, + job_timeout=3600, + job_ttl=ttl, + ) + context.job_store = InMemoryJobStore() + context.worker = AsyncWorker(config, context.job_store) + context.async_job = None + + +@given("a queued async job exists") +def step_queued_job_exists(context: Context) -> None: + job = AsyncJob( + job_id=_make_job_id(), + plan_id=PLAN_ID, + phase="execute", + ) + context.job_store.add(job) + context.async_job = job + + +@given("a running async job with expired heartbeat exists") +def step_running_job_expired_heartbeat(context: Context) -> None: + job = AsyncJob( + job_id=_make_job_id(), + plan_id=PLAN_ID, + phase="execute", + ) + context.job_store.add(job) + job.mark_running("worker-stuck") + # Backdate heartbeat + job.last_heartbeat = datetime.now(UTC) - timedelta(seconds=10) + job.started_at = datetime.now(UTC) - timedelta(seconds=10) + context.job_store.update(job) + context.async_job = job + + +@given("a completed async job older than TTL exists") +def step_completed_job_older_than_ttl(context: Context) -> None: + job = AsyncJob( + job_id=_make_job_id(), + plan_id=PLAN_ID, + phase="execute", + ) + context.job_store.add(job) + job.mark_running("worker-old") + job.mark_succeeded() + # Backdate finished_at + job.finished_at = datetime.now(UTC) - timedelta(seconds=10) + context.job_store.update(job) + context.async_job = job + + +@given("a cancellation token is created") +def step_cancellation_token_created(context: Context) -> None: + context.token = CancellationToken() + + +@given("I cancel the async token") +def step_cancel_token(context: Context) -> None: + context.token.cancel() + + +@given('a worker health report for "{worker_id}"') +def step_health_report_created(context: Context, worker_id: str) -> None: + context.health_report = WorkerHealthReport(worker_id) + + +@given("I mark the async job as succeeded") +def step_given_mark_succeeded(context: Context) -> None: + context.async_job.mark_succeeded() + context.job_store.update(context.async_job) + + +@given("I mark the async job as failed") +def step_given_mark_failed(context: Context) -> None: + context.async_job.mark_failed() + context.job_store.update(context.async_job) + + +# --------------------------------------------------------------------------- +# When steps +# --------------------------------------------------------------------------- + + +@when('I mark the async job as running with worker "{worker_id}"') +def step_mark_running(context: Context, worker_id: str) -> None: + context.async_job.mark_running(worker_id) + context.job_store.update(context.async_job) + + +@when("I mark the async job as succeeded") +def step_mark_succeeded(context: Context) -> None: + context.async_job.mark_succeeded() + context.job_store.update(context.async_job) + + +@when("I mark the async job as failed") +def step_mark_failed(context: Context) -> None: + context.async_job.mark_failed() + context.job_store.update(context.async_job) + + +@when("I mark the async job as cancelled") +def step_mark_cancelled(context: Context) -> None: + context.async_job.mark_cancelled() + context.job_store.update(context.async_job) + + +@when("I try to mark the async job as succeeded") +def step_try_mark_succeeded(context: Context) -> None: + try: + context.async_job.mark_succeeded() + context.error = None + except InvalidJobTransitionError as exc: + context.error = exc + + +@when('I try to mark the async job as running with worker "{worker_id}"') +def step_try_mark_running(context: Context, worker_id: str) -> None: + try: + context.async_job.mark_running(worker_id) + context.error = None + except (InvalidJobTransitionError, ValueError) as exc: + context.error = exc + + +@when("I try to mark the async job as running with an empty worker_id") +def step_try_mark_running_empty(context: Context) -> None: + try: + context.async_job.mark_running("") + context.error = None + except (InvalidJobTransitionError, ValueError) as exc: + context.error = exc + + +@when("the worker picks up and executes the job") +def step_worker_pickup_execute(context: Context) -> None: + context.worker.pickup_and_execute(context.async_job) + + +@when("I cancel the async job") +def step_cancel_job(context: Context) -> None: + context.worker.cancel_job(context.async_job.job_id) + + +@when("the worker detects stuck jobs") +def step_detect_stuck_jobs(context: Context) -> None: + context.stuck_jobs = context.worker.detect_stuck_jobs() + + +@when("the worker runs job cleanup") +def step_run_cleanup(context: Context) -> None: + context.cleaned_count = context.worker.cleanup_completed_jobs() + + +@when("the worker is started") +def step_worker_started(context: Context) -> None: + context.worker.start() + + +@when("a shutdown is requested") +def step_shutdown_requested(context: Context) -> None: + context.worker.request_shutdown() + # Wait for the poll thread to observe the shutdown event rather + # than relying on an arbitrary sleep that can be flaky on slow CI. + poll_thread = context.worker._poll_thread + if poll_thread is not None: + poll_thread.join(timeout=2.0) + context.worker.stop() + + +@when('I serialize a job payload for plan "{plan_id}" phase "{phase}"') +def step_serialize_payload(context: Context, plan_id: str, phase: str) -> None: + context.payload_json = serialize_job_payload(plan_id, phase) + context.original_plan_id = plan_id + context.original_phase = phase + + +@when("I serialize a job payload with extra data") +def step_serialize_payload_extra(context: Context) -> None: + context.payload_json = serialize_job_payload( + PLAN_ID, "execute", extra={"key": "value"} + ) + + +@when("I try to deserialize invalid JSON") +def step_try_deserialize_invalid(context: Context) -> None: + try: + deserialize_job_payload("not valid json {{{") + context.error = None + except ValueError as exc: + context.error = exc + + +@when("I request the worker health report") +def step_request_health_report(context: Context) -> None: + context.health_data = context.worker.get_health_report() + + +@when("I try to create AsyncWorkerConfig with max_workers {max_workers:d}") +def step_try_create_config_max_workers(context: Context, max_workers: int) -> None: + try: + AsyncWorkerConfig(max_workers=max_workers) + context.config_error = None + except ValueError as exc: + context.config_error = exc + + +@when("I try to create AsyncWorkerConfig with poll_interval {interval:d}") +def step_try_create_config_poll_interval(context: Context, interval: int) -> None: + try: + AsyncWorkerConfig(poll_interval=float(interval)) + context.config_error = None + except ValueError as exc: + context.config_error = exc + + +@when("I try to add the same job again") +def step_try_add_duplicate(context: Context) -> None: + try: + context.job_store.add(context.async_job) + context.error = None + except ValueError as exc: + context.error = exc + + +@when('I list jobs by status "{status}"') +def step_list_by_status(context: Context, status: str) -> None: + context.job_list = context.job_store.list_by_status(AsyncJobStatus(status)) + + +@when("I record a heartbeat on the async job") +def step_record_heartbeat(context: Context) -> None: + before = context.async_job.last_heartbeat + # Busy-wait until the clock advances past the previous heartbeat + # to avoid flaky failures on fast systems where 10ms may not be + # enough to produce a distinct datetime.now(UTC) value. + while datetime.now(UTC) <= before: + time.sleep(0.001) + context.async_job.record_heartbeat() + context.heartbeat_before = before + + +@when("I try to record a heartbeat on the async job") +def step_try_record_heartbeat(context: Context) -> None: + try: + context.async_job.record_heartbeat() + context.error = None + except ValueError as exc: + context.error = exc + + +@when("I get the async job CLI dict") +def step_get_cli_dict(context: Context) -> None: + context.cli_dict = context.async_job.as_cli_dict() + + +@when('I try to create an async job with phase "{phase}"') +def step_try_create_invalid_phase(context: Context, phase: str) -> None: + try: + AsyncJob( + job_id=_make_job_id(), + plan_id=PLAN_ID, + phase=phase, + ) + context.error = None + except ValidationError as exc: + context.error = exc + + +@when("I try to create an async job with invalid payload JSON") +def step_try_create_invalid_payload(context: Context) -> None: + try: + AsyncJob( + job_id=_make_job_id(), + plan_id=PLAN_ID, + phase="execute", + payload_json="not json!!!", + ) + context.error = None + except ValidationError as exc: + context.error = exc + + +@when("I cancel the async token") +def step_when_cancel_token(context: Context) -> None: + context.token.cancel() + + +@when("I reset the token") +def step_reset_token(context: Context) -> None: + context.token.reset() + + +@when("I record {completed:d} completed jobs and {failed:d} failed job") +def step_record_jobs(context: Context, completed: int, failed: int) -> None: + for _ in range(completed): + context.health_report.record_job_completed() + for _ in range(failed): + context.health_report.record_job_failed() + + +@when("I try to serialize a payload with empty plan_id") +def step_try_serialize_empty_plan_id(context: Context) -> None: + try: + serialize_job_payload("", "execute") + context.error = None + except ValueError as exc: + context.error = exc + + +@when("I try to deserialize an empty payload") +def step_try_deserialize_empty(context: Context) -> None: + try: + deserialize_job_payload("") + context.error = None + except ValueError as exc: + context.error = exc + + +@when("I call can_transition_job with invalid types") +def step_can_transition_invalid(context: Context) -> None: + try: + can_transition_job("queued", "running") # type: ignore[arg-type] + context.error = None + except TypeError as exc: + context.error = exc + + +# --------------------------------------------------------------------------- +# Then steps +# --------------------------------------------------------------------------- + + +@then('the async job status should be "{status}"') +def step_check_status(context: Context, status: str) -> None: + assert context.async_job.status.value == status, ( + f"Expected {status}, got {context.async_job.status.value}" + ) + + +@then('the async job phase should be "{phase}"') +def step_check_phase(context: Context, phase: str) -> None: + assert context.async_job.phase == phase + + +@then("the async job schema_version should be {version:d}") +def step_check_schema_version(context: Context, version: int) -> None: + assert context.async_job.schema_version == version + + +@then('the async job worker_id should be "{worker_id}"') +def step_check_worker_id(context: Context, worker_id: str) -> None: + assert context.async_job.worker_id == worker_id + + +@then("the async job started_at should be set") +def step_check_started_at(context: Context) -> None: + assert context.async_job.started_at is not None + + +@then("the async job finished_at should be set") +def step_check_finished_at(context: Context) -> None: + assert context.async_job.finished_at is not None + + +@then("an InvalidJobTransitionError should be raised") +def step_check_transition_error(context: Context) -> None: + assert isinstance(context.error, InvalidJobTransitionError), ( + f"Expected InvalidJobTransitionError, got {type(context.error)}: {context.error}" + ) + + +@then("the worker health should show {count:d} job processed") +def step_check_health_processed(context: Context, count: int) -> None: + assert context.worker.health.jobs_processed == count + + +@then("the worker health should show {count:d} job failed") +def step_check_health_failed(context: Context, count: int) -> None: + assert context.worker.health.jobs_failed == count + + +@then('the stuck job should be marked as "{status}"') +def step_check_stuck_status(context: Context, status: str) -> None: + job = context.job_store.get(context.async_job.job_id) + assert job is not None + assert job.status.value == status + + +@then("the completed job should be removed from the store") +def step_check_job_removed(context: Context) -> None: + job = context.job_store.get(context.async_job.job_id) + assert job is None + + +@then("the worker should stop running") +def step_check_worker_stopped(context: Context) -> None: + assert not context.worker.is_running + + +@then("the serialized payload should contain schema_version {version:d}") +def step_check_payload_schema(context: Context, version: int) -> None: + data = json.loads(context.payload_json) + assert data["schema_version"] == version + + +@then("the deserialized payload should match the original") +def step_check_deserialized(context: Context) -> None: + data = deserialize_job_payload(context.payload_json) + assert data["plan_id"] == context.original_plan_id + assert data["phase"] == context.original_phase + + +@then("the serialized payload should contain the extra data") +def step_check_extra_data(context: Context) -> None: + data = json.loads(context.payload_json) + assert data["extra"]["key"] == "value" + + +@then("a ValueError should be raised for async deserialize") +def step_check_value_error(context: Context) -> None: + assert isinstance(context.error, ValueError), ( + f"Expected ValueError, got {type(context.error)}" + ) + + +@then("the health report should contain worker_id") +def step_check_health_worker_id(context: Context) -> None: + assert "worker_id" in context.health_data + + +@then("the health report should contain config details") +def step_check_health_config(context: Context) -> None: + assert "config" in context.health_data + assert context.health_data["config"]["max_workers"] == 4 + + +@then("a ValueError should be raised for config") +def step_check_config_error(context: Context) -> None: + assert isinstance(context.config_error, ValueError) + + +@then("a ValueError should be raised for duplicate job") +def step_check_duplicate_error(context: Context) -> None: + assert isinstance(context.error, ValueError) + + +@then("the job list should have {count:d} entries") +def step_check_job_list_count(context: Context, count: int) -> None: + assert len(context.job_list) == count + + +@then("the async job last_heartbeat should be updated") +def step_check_heartbeat_updated(context: Context) -> None: + assert context.async_job.last_heartbeat is not None + if context.heartbeat_before is not None: + assert context.async_job.last_heartbeat >= context.heartbeat_before + + +@then("a ValueError should be raised for heartbeat") +def step_check_heartbeat_error(context: Context) -> None: + assert isinstance(context.error, ValueError) + + +@then("the async job should be terminal") +def step_check_is_terminal(context: Context) -> None: + assert context.async_job.is_terminal + + +@then("the async job should be running") +def step_check_is_running(context: Context) -> None: + assert context.async_job.is_running + + +@then('the async CLI dict should contain key "{key}"') +def step_check_cli_dict_key(context: Context, key: str) -> None: + assert key in context.cli_dict + + +@then("a validation error should be raised for phase") +def step_check_phase_error(context: Context) -> None: + assert context.error is not None + + +@then("a validation error should be raised for payload") +def step_check_payload_error(context: Context) -> None: + assert context.error is not None + + +@then("a ValueError should be raised for worker_id") +def step_check_worker_id_error(context: Context) -> None: + assert isinstance(context.error, ValueError) + + +@then("the token should be cancelled") +def step_check_token_cancelled(context: Context) -> None: + assert context.token.is_cancelled + + +@then("the token wait should return True") +def step_check_token_wait(context: Context) -> None: + assert context.token.wait(timeout=0.01) + + +@then("the token should not be cancelled") +def step_check_token_not_cancelled(context: Context) -> None: + assert not context.token.is_cancelled + + +@then("the health report should show {processed:d} processed and {failed:d} failed") +def step_check_health_counts(context: Context, processed: int, failed: int) -> None: + assert context.health_report.jobs_processed == processed + assert context.health_report.jobs_failed == failed + + +@then("a ValueError should be raised for serialize") +def step_check_serialize_error(context: Context) -> None: + assert isinstance(context.error, ValueError) + + +@then("a ValueError should be raised for deserialize") +def step_check_deserialize_error(context: Context) -> None: + assert isinstance(context.error, ValueError) + + +@then("a TypeError should be raised for async transition") +def step_check_type_error(context: Context) -> None: + assert isinstance(context.error, TypeError) + + +# ---- Additional Coverage Steps ---- + + +@when("I try to create a WorkerHealthReport with empty worker_id") +def step_try_empty_health_report(context: Context) -> None: + try: + WorkerHealthReport("") + context.error = None + except ValueError as exc: + context.error = exc + + +@then("a ValueError should be raised for empty worker_id") +def step_check_empty_worker_id_error(context: Context) -> None: + assert isinstance(context.error, ValueError) + + +@when("I try to create AsyncWorkerConfig with non-bool enabled") +def step_try_config_nonbool(context: Context) -> None: + try: + AsyncWorkerConfig(enabled="yes") # type: ignore[arg-type] + context.error = None + except TypeError as exc: + context.error = exc + + +@then("a TypeError should be raised for async config enabled") +def step_check_config_enabled_error(context: Context) -> None: + assert isinstance(context.error, TypeError) + + +@when("I try to create AsyncWorkerConfig with job_timeout {timeout:d}") +def step_try_config_job_timeout(context: Context, timeout: int) -> None: + try: + AsyncWorkerConfig(job_timeout=timeout) + context.error = None + except ValueError as exc: + context.error = exc + + +@when("I try to create AsyncWorkerConfig with job_ttl {ttl:d}") +def step_try_config_job_ttl(context: Context, ttl: int) -> None: + try: + AsyncWorkerConfig(job_ttl=ttl) + context.error = None + except ValueError as exc: + context.error = exc + + +@then("a ValueError should be raised for async config validation") +def step_check_config_validation_error(context: Context) -> None: + assert isinstance(context.error, ValueError) + + +@when("I try to add a non-AsyncJob to the store") +def step_try_add_non_job(context: Context) -> None: + try: + context.job_store.add("not a job") # type: ignore[arg-type] + context.error = None + except TypeError as exc: + context.error = exc + + +@then("a TypeError should be raised for async store type") +def step_check_store_type_error(context: Context) -> None: + assert isinstance(context.error, TypeError) + + +@when("I try to get a job with empty job_id") +def step_try_get_empty_id(context: Context) -> None: + try: + context.job_store.get("") + context.error = None + except ValueError as exc: + context.error = exc + + +@then("a ValueError should be raised for async store get") +def step_check_store_get_error(context: Context) -> None: + assert isinstance(context.error, ValueError) + + +@when("I try to list jobs by invalid status type") +def step_try_list_invalid_status(context: Context) -> None: + try: + context.job_store.list_by_status("invalid") # type: ignore[arg-type] + context.error = None + except TypeError as exc: + context.error = exc + + +@when("I try to update with a non-AsyncJob") +def step_try_update_non_job(context: Context) -> None: + try: + context.job_store.update("not a job") # type: ignore[arg-type] + context.error = None + except TypeError as exc: + context.error = exc + + +@when("I try to update a non-existent job") +def step_try_update_nonexistent(context: Context) -> None: + try: + job = AsyncJob( + job_id=str(ULID()), + plan_id="01HXYZ01HXYZ01HXYZ01HXYZ01", + phase="execute", + ) + context.job_store.update(job) + context.error = None + except ValueError as exc: + context.error = exc + + +@then("a ValueError should be raised for async store update") +def step_check_store_update_error(context: Context) -> None: + assert isinstance(context.error, ValueError) + + +@when("I try to remove a job with empty job_id") +def step_try_remove_empty_id(context: Context) -> None: + try: + context.job_store.remove("") + context.error = None + except ValueError as exc: + context.error = exc + + +@then("a ValueError should be raised for async store remove") +def step_check_store_remove_error(context: Context) -> None: + assert isinstance(context.error, ValueError) + + +@when("I try to create AsyncWorker with invalid config") +def step_try_worker_invalid_config(context: Context) -> None: + try: + AsyncWorker(config="bad", job_store=InMemoryJobStore()) # type: ignore[arg-type] + context.error = None + except TypeError as exc: + context.error = exc + + +@then("a TypeError should be raised for async worker config") +def step_check_worker_config_error(context: Context) -> None: + assert isinstance(context.error, TypeError) + + +@when("I try to create AsyncWorker with invalid job_store") +def step_try_worker_invalid_store(context: Context) -> None: + try: + AsyncWorker(config=AsyncWorkerConfig(), job_store="bad") # type: ignore[arg-type] + context.error = None + except TypeError as exc: + context.error = exc + + +@then("a TypeError should be raised for async worker store") +def step_check_worker_store_error(context: Context) -> None: + assert isinstance(context.error, TypeError) + + +@when("I try to pickup a non-AsyncJob") +def step_try_pickup_non_job(context: Context) -> None: + try: + context.worker.pickup_and_execute("bad") # type: ignore[arg-type] + context.error = None + except TypeError as exc: + context.error = exc + + +@then("a TypeError should be raised for async worker pickup") +def step_check_worker_pickup_error(context: Context) -> None: + assert isinstance(context.error, TypeError) + + +@when("I try to cancel a job with empty job_id") +def step_try_cancel_empty_id(context: Context) -> None: + try: + context.worker.cancel_job("") + context.error = None + except ValueError as exc: + context.error = exc + + +@then("a ValueError should be raised for async worker cancel") +def step_check_worker_cancel_error(context: Context) -> None: + assert isinstance(context.error, ValueError) + + +@then('the worker should have a worker_id starting with "worker-"') +def step_check_worker_id_prefix(context: Context) -> None: + assert context.worker.worker_id.startswith("worker-") + + +@then("the worker config max_workers should be {max_workers:d}") +def step_check_worker_config_value(context: Context, max_workers: int) -> None: + assert context.worker.config.max_workers == max_workers + + +@when("I start the worker") +def step_start_worker(context: Context) -> None: + context.worker.start() + + +@when("I start the worker again") +def step_start_worker_again(context: Context) -> None: + context.worker.start() + + +@then("the worker should still be running") +def step_worker_still_running(context: Context) -> None: + assert context.worker.is_running + + +@when("I stop the worker") +def step_stop_worker(context: Context) -> None: + context.worker.stop() + + +@then("the worker should not be running") +def step_worker_not_running(context: Context) -> None: + assert not context.worker.is_running + + +@then("cancelling a non-existent job should return false") +def step_cancel_nonexistent_job(context: Context) -> None: + result = context.worker.cancel_job("01ZZZZZZZZZZZZZZZZZZZZZZZZ") + assert result is False + + +@then("the async job payload property should return a dict") +def step_check_payload_property(context: Context) -> None: + payload = context.async_job.payload + assert isinstance(payload, dict) + + +@when('I try to create an async job with payload_json "not-json{{"') +def step_try_invalid_payload_json(context: Context) -> None: + try: + AsyncJob( + job_id=str(ULID()), + plan_id="01HXYZ01HXYZ01HXYZ01HXYZ01", + phase="execute", + payload_json="not-json{", + ) + context.error = None + except Exception as exc: + context.error = exc + + +# --------------------------------------------------------------------------- +# Review-fix steps: concurrent execution (T2) +# --------------------------------------------------------------------------- + + +@given("{count:d} queued async jobs exist") +def step_n_queued_jobs_exist(context: Context, count: int) -> None: + context.created_jobs = [] + for _ in range(count): + job = AsyncJob( + job_id=_make_job_id(), + plan_id=PLAN_ID, + phase="execute", + ) + context.job_store.add(job) + context.created_jobs.append(job) + + +@when("the worker executes all queued jobs concurrently") +def step_worker_executes_concurrently(context: Context) -> None: + """Use threads to verify concurrent execution.""" + threads: list[threading.Thread] = [] + for job in context.created_jobs: + t = threading.Thread(target=context.worker.pickup_and_execute, args=(job,)) + threads.append(t) + t.start() + for t in threads: + t.join(timeout=5.0) + + +@then("all {count:d} jobs should be in a terminal state") +def step_all_jobs_terminal(context: Context, count: int) -> None: + terminal = [j for j in context.created_jobs if j.is_terminal] + assert len(terminal) == count, f"Expected {count} terminal, got {len(terminal)}" + + +@then("the worker health should show {count:d} jobs processed total") +def step_check_total_processed(context: Context, count: int) -> None: + assert context.worker.health.jobs_processed == count, ( + f"Expected {count} processed, got {context.worker.health.jobs_processed}" + ) + + +@then("the worker health should show {count:d} jobs processed") +def step_check_health_processed_exact(context: Context, count: int) -> None: + assert context.worker.health.jobs_processed == count + + +@then("the worker health should show {count:d} jobs failed") +def step_check_health_failed_exact(context: Context, count: int) -> None: + assert context.worker.health.jobs_failed == count + + +# --------------------------------------------------------------------------- +# Review-fix steps: error_message audit trail (SEC1) +# --------------------------------------------------------------------------- + + +@then("the async job should have an error_message set") +def step_check_error_message_set(context: Context) -> None: + assert context.async_job.error_message is not None, ( + "Expected error_message to be set, but it was None" + ) + assert len(context.async_job.error_message) > 0 + + +# --------------------------------------------------------------------------- +# Review-fix steps: AsyncJobModel DB round-trip (T1) +# --------------------------------------------------------------------------- + + +@when("I create and persist an AsyncJobModel to the database") +def step_create_persist_db_model(context: Context) -> None: + from sqlalchemy import create_engine + from sqlalchemy.orm import sessionmaker + + from cleveragents.infrastructure.database.models import AsyncJobModel, Base + + engine = create_engine("sqlite:///:memory:") + Base.metadata.create_all(engine) + context.db_engine = engine + context.db_session_factory = sessionmaker(bind=engine) + + job_id = str(ULID()) + context.db_job_id = job_id + with context.db_session_factory() as session: + model = AsyncJobModel( + job_id=job_id, + plan_id=PLAN_ID, + phase="execute", + status="queued", + payload_json='{"schema_version": 1}', + created_at=datetime.now(UTC).isoformat(), + schema_version=1, + ) + session.add(model) + session.commit() + + +@then("the AsyncJobModel should be retrievable with correct fields") +def step_check_db_model_retrieve(context: Context) -> None: + from cleveragents.infrastructure.database.models import AsyncJobModel + + with context.db_session_factory() as session: + row = session.query(AsyncJobModel).filter_by(job_id=context.db_job_id).first() + assert row is not None, "AsyncJobModel not found in DB" + assert row.plan_id == PLAN_ID + assert row.phase == "execute" + assert row.status == "queued" + assert row.schema_version == 1 + + +@then("the database CHECK constraint should reject invalid status") +def step_check_db_constraint(context: Context) -> None: + from sqlalchemy.exc import IntegrityError + + from cleveragents.infrastructure.database.models import AsyncJobModel + + with context.db_session_factory() as session: + bad = AsyncJobModel( + job_id=str(ULID()), + plan_id=PLAN_ID, + phase="execute", + status="invalid_status", + payload_json="{}", + created_at=datetime.now(UTC).isoformat(), + schema_version=1, + ) + session.add(bad) + try: + session.commit() + # SQLite enforces CHECK constraints; this should fail + context.db_constraint_enforced = False + except IntegrityError: + session.rollback() + context.db_constraint_enforced = True + + assert context.db_constraint_enforced, ( + "Expected CHECK constraint to reject invalid status" + ) + + +# --------------------------------------------------------------------------- +# Coverage steps: Safe initialisation and cleanup +# --------------------------------------------------------------------------- + + +@given("an async worker is initialised with a blocking executor") +def step_worker_with_blocking_executor(context: Context) -> None: + """Create a worker whose executor blocks until a threading.Event is set.""" + entered_event = threading.Event() + release_event = threading.Event() + context.blocking_entered = entered_event + context.blocking_release = release_event + + def blocking_executor(job: AsyncJob, token: CancellationToken) -> None: + entered_event.set() + # Block until released or cancelled + while not release_event.is_set() and not token.is_cancelled: + release_event.wait(timeout=0.05) + + config = AsyncWorkerConfig( + enabled=True, + max_workers=2, + poll_interval=0.05, + job_timeout=3600, + job_ttl=86400, + ) + context.job_store = InMemoryJobStore() + context.worker = AsyncWorker(config, context.job_store, blocking_executor) + context.async_job = None + context.error = None + + +@given("an async worker is initialised with a terminal-forcing executor") +def step_worker_with_terminal_executor(context: Context) -> None: + """Create a worker whose executor forcefully marks the job succeeded.""" + + def terminal_executor(job: AsyncJob, token: CancellationToken) -> None: + # Force the job to succeeded during execution — pickup_and_execute + # should detect this and skip its own transition. + job.mark_succeeded() + + config = AsyncWorkerConfig( + enabled=True, + max_workers=2, + poll_interval=0.1, + job_timeout=3600, + job_ttl=86400, + ) + context.job_store = InMemoryJobStore() + context.worker = AsyncWorker(config, context.job_store, terminal_executor) + context.async_job = None + context.error = None + + +@when("the worker processes all queued jobs within {seconds:d} seconds") +def step_worker_processes_queued(context: Context, seconds: int) -> None: + """Wait until context.async_job reaches a terminal state.""" + deadline = time.monotonic() + seconds + while time.monotonic() < deadline: + if context.async_job.is_terminal: + return + time.sleep(0.05) + assert context.async_job.is_terminal, ( + f"Job did not reach terminal state within {seconds}s " + f"(status={context.async_job.status.value})" + ) + + +@when("all created jobs reach a terminal state within {seconds:d} seconds") +def step_all_created_jobs_terminal(context: Context, seconds: int) -> None: + deadline = time.monotonic() + seconds + while time.monotonic() < deadline: + if all(j.is_terminal for j in context.created_jobs): + return + time.sleep(0.05) + pending = [j for j in context.created_jobs if not j.is_terminal] + assert not pending, f"{len(pending)} job(s) still not terminal after {seconds}s" + + +@when("the blocking executor has been entered") +def step_blocking_executor_entered(context: Context) -> None: + entered = context.blocking_entered.wait(timeout=5.0) + assert entered, "Blocking executor was not entered within 5s" + + +@when("I dispatch the job in a background thread") +def step_dispatch_job_background(context: Context) -> None: + t = threading.Thread( + target=context.worker.pickup_and_execute, + args=(context.async_job,), + ) + context.background_job_thread = t + t.start() + + +@when("I cancel the running async job via the worker") +def step_cancel_running_job(context: Context) -> None: + result = context.worker.cancel_job(context.async_job.job_id) + context.cancel_result = result + # Release the blocking executor so the thread can finish + if hasattr(context, "blocking_release"): + context.blocking_release.set() + + +@when("the background job thread completes") +def step_background_thread_completes(context: Context) -> None: + context.background_job_thread.join(timeout=5.0) + assert not context.background_job_thread.is_alive(), ( + "Background job thread did not complete within 5s" + ) + + +@given("the async job is already cancelled") +def step_job_already_cancelled(context: Context) -> None: + context.async_job.mark_cancelled() + context.job_store.update(context.async_job) + + +@then("the cancellation token for the job should be signalled") +def step_check_cancel_token_signalled(context: Context) -> None: + # After stop(), all tokens should have been cancelled + # We verify the job reached a terminal state or the token was signalled + # The _cancel_all_running_jobs path fires during stop() + # Since the job was in-flight, it should be terminal + job = context.job_store.get(context.async_job.job_id) + assert job is not None + # The job should be terminal (cancelled or succeeded) after stop + # releases executor and cancels tokens + assert job.is_terminal or context.async_job.is_terminal, ( + f"Expected terminal job after stop, got {job.status.value}" + ) + + +@then("the worker health should show {count:d} jobs cancelled") +def step_check_health_cancelled(context: Context, count: int) -> None: + assert context.worker.health.jobs_cancelled == count, ( + f"Expected {count} cancelled, got {context.worker.health.jobs_cancelled}" + ) + + +@then("the async job should be in a terminal state") +def step_check_job_terminal(context: Context) -> None: + assert context.async_job.is_terminal, ( + f"Expected terminal, got {context.async_job.status.value}" + ) + + +# --------------------------------------------------------------------------- +# Coverage steps: Job store operations +# --------------------------------------------------------------------------- + + +@given("I add {count:d} jobs in mixed statuses to the store") +def step_add_mixed_jobs(context: Context, count: int) -> None: + context.store_jobs = [] + statuses_to_set = ["queued", "running", "succeeded"] + for i in range(count): + job = AsyncJob(job_id=_make_job_id(), plan_id=PLAN_ID, phase="execute") + target = statuses_to_set[i % len(statuses_to_set)] + if target == "running": + job.mark_running(f"worker-{i}") + elif target == "succeeded": + job.mark_running(f"worker-{i}") + job.mark_succeeded() + context.job_store.add(job) + context.store_jobs.append(job) + + +@when("I call list_all on the job store") +def step_call_list_all(context: Context) -> None: + context.all_jobs = context.job_store.list_all() + + +@then("the store should return {count:d} jobs") +def step_store_returns_count(context: Context, count: int) -> None: + assert len(context.all_jobs) == count, ( + f"Expected {count}, got {len(context.all_jobs)}" + ) + + +@then("the store count should be {count:d}") +def step_store_count(context: Context, count: int) -> None: + assert context.job_store.count() == count, ( + f"Expected count {count}, got {context.job_store.count()}" + ) + + +@given("a queued async job exists in the store") +def step_queued_job_in_store(context: Context) -> None: + job = AsyncJob(job_id=_make_job_id(), plan_id=PLAN_ID, phase="execute") + context.job_store.add(job) + context.async_job = job + + +@when("I remove the job from the store") +def step_remove_job_from_store(context: Context) -> None: + context.job_store.remove(context.async_job.job_id) + + +# --------------------------------------------------------------------------- +# Coverage steps: Domain model validators +# --------------------------------------------------------------------------- + + +@given("a queued async job exists standalone") +def step_queued_job_standalone(context: Context) -> None: + context.job_store = InMemoryJobStore() + context.async_job = AsyncJob( + job_id=_make_job_id(), plan_id=PLAN_ID, phase="execute" + ) + context.job_store.add(context.async_job) + context.error = None + + +@when("I try to transition_to with a string instead of enum") +def step_transition_with_string(context: Context) -> None: + try: + context.async_job.transition_to("running") # type: ignore[arg-type] + context.error = None + except TypeError as exc: + context.error = exc + + +@when("I call can_transition_job with a string to_status") +def step_can_transition_with_string(context: Context) -> None: + try: + can_transition_job(AsyncJobStatus.QUEUED, "running") # type: ignore[arg-type] + context.error = None + except TypeError as exc: + context.error = exc + + +@then("an async TypeError should be raised") +def step_check_async_type_error(context: Context) -> None: + assert isinstance(context.error, TypeError), ( + f"Expected TypeError, got {type(context.error)}: {context.error}" + ) + + +@when("I try to serialize a payload with an empty phase") +def step_serialize_empty_phase(context: Context) -> None: + try: + serialize_job_payload(PLAN_ID, "") + context.error = None + except ValueError as exc: + context.error = exc + + +@when("I try to serialize a payload with schema_version {version:d}") +def step_serialize_bad_version(context: Context, version: int) -> None: + try: + serialize_job_payload(PLAN_ID, "execute", schema_version=version) + context.error = None + except ValueError as exc: + context.error = exc + + +@then('an async ValueError should be raised matching "{fragment}"') +def step_check_async_value_error_message(context: Context, fragment: str) -> None: + assert isinstance(context.error, ValueError), ( + f"Expected ValueError, got {type(context.error)}: {context.error}" + ) + assert fragment in str(context.error), f"Expected '{fragment}' in '{context.error}'" + assert fragment in str(context.error), f"Expected '{fragment}' in '{context.error}'" + + +@when('I try to deserialize the JSON string "{json_str}"') +def step_deserialize_non_object(context: Context, json_str: str) -> None: + try: + deserialize_job_payload(json_str) + context.error = None + except ValueError as exc: + context.error = exc + + +@when('I mark the async job as failed with error "{error_msg}"') +def step_mark_failed_with_error(context: Context, error_msg: str) -> None: + context.async_job.mark_failed(error=error_msg) + + +@then('the async CLI dict error_message should be "{expected}"') +def step_check_cli_dict_error_message(context: Context, expected: str) -> None: + assert context.cli_dict["error_message"] == expected, ( + f"Expected '{expected}', got '{context.cli_dict.get('error_message')}'" + ) + + +# --------------------------------------------------------------------------- +# Coverage steps: Transition-conflict edge cases +# --------------------------------------------------------------------------- + + +@given('the async job is already running with worker "{worker_id}"') +def step_job_already_running(context: Context, worker_id: str) -> None: + """Manually transition the job to running so pickup_and_execute + encounters an InvalidJobTransitionError on mark_running.""" + context.async_job.mark_running(worker_id) + context.job_store.update(context.async_job) + + +@given("an async worker is initialised with a crash-after-terminal executor") +def step_worker_crash_after_terminal(context: Context) -> None: + """Executor that marks the job succeeded then raises, testing the + inner InvalidJobTransitionError path in the exception handler.""" + + def crash_after_terminal(job: AsyncJob, token: CancellationToken) -> None: + job.mark_succeeded() + raise RuntimeError("Post-terminal crash") + + config = AsyncWorkerConfig( + enabled=True, + max_workers=2, + poll_interval=0.1, + job_timeout=3600, + job_ttl=86400, + ) + context.job_store = InMemoryJobStore() + context.worker = AsyncWorker(config, context.job_store, crash_after_terminal) + context.async_job = None + context.error = None + + +@given("the running async job has already been cancelled externally") +def step_running_job_cancelled_externally(context: Context) -> None: + """Simulate a race where the job becomes cancelled between the + stuck-detection check and the mark_failed call.""" + context.async_job.mark_cancelled() + context.job_store.update(context.async_job) + + +@then("no additional stuck jobs should be reported") +def step_no_stuck_jobs(context: Context) -> None: + assert len(context.stuck_jobs) == 0, ( + f"Expected 0 stuck jobs, got {len(context.stuck_jobs)}" + ) + + +# --------------------------------------------------------------------------- +# Review-fix steps: Plan lifecycle async wiring (P1) +# --------------------------------------------------------------------------- + + +def _make_mock_settings(async_enabled: bool = False) -> MagicMock: + """Create a mock Settings object with async config.""" + settings = MagicMock() + settings.async_enabled = async_enabled + return settings + + +@given("a plan lifecycle service with async enabled and a job store") +def step_lifecycle_async_enabled(context: Context) -> None: + context.lifecycle_job_store = InMemoryJobStore() + context.lifecycle_settings = _make_mock_settings(async_enabled=True) + context.lifecycle_service = PlanLifecycleService( + settings=context.lifecycle_settings, + job_store=context.lifecycle_job_store, + ) + context.lifecycle_plan = None + + +@given("a plan lifecycle service with async disabled") +def step_lifecycle_async_disabled(context: Context) -> None: + context.lifecycle_job_store = InMemoryJobStore() + context.lifecycle_settings = _make_mock_settings(async_enabled=False) + context.lifecycle_service = PlanLifecycleService( + settings=context.lifecycle_settings, + job_store=context.lifecycle_job_store, + ) + context.lifecycle_plan = None + + +@given("a plan lifecycle service with async enabled but no job store") +def step_lifecycle_async_no_store(context: Context) -> None: + context.lifecycle_job_store = None + context.lifecycle_settings = _make_mock_settings(async_enabled=True) + context.lifecycle_service = PlanLifecycleService( + settings=context.lifecycle_settings, + job_store=None, + ) + context.lifecycle_plan = None + + +@given("a plan in Strategize/COMPLETE state exists") +def step_plan_strategize_complete(context: Context) -> None: + context.lifecycle_service.create_action( + name="local/test-action", + description="Test action", + definition_of_done="Done", + strategy_actor="local/strategy", + execution_actor="local/executor", + ) + plan = context.lifecycle_service.use_action("local/test-action") + context.lifecycle_service.start_strategize(plan.identity.plan_id) + context.lifecycle_service.complete_strategize(plan.identity.plan_id) + context.lifecycle_plan = context.lifecycle_service.get_plan(plan.identity.plan_id) + + +@given("a plan in Execute/COMPLETE state exists") +def step_plan_execute_complete(context: Context) -> None: + context.lifecycle_service.create_action( + name="local/test-action-apply", + description="Test action for apply", + definition_of_done="Done", + strategy_actor="local/strategy", + execution_actor="local/executor", + ) + plan = context.lifecycle_service.use_action("local/test-action-apply") + context.lifecycle_service.start_strategize(plan.identity.plan_id) + plan = context.lifecycle_service.complete_strategize(plan.identity.plan_id) + # complete_strategize may auto-progress; if not, manually advance + plan = context.lifecycle_service.get_plan(plan.identity.plan_id) + if plan.phase.value == "strategize": + plan = context.lifecycle_service.execute_plan(plan.identity.plan_id) + # Clear any jobs enqueued by execute_plan before testing apply + if context.lifecycle_job_store is not None: + for j in context.lifecycle_job_store.list_all(): + context.lifecycle_job_store.remove(j.job_id) + context.lifecycle_service.start_execute(plan.identity.plan_id) + context.lifecycle_service.complete_execute(plan.identity.plan_id) + context.lifecycle_plan = context.lifecycle_service.get_plan(plan.identity.plan_id) + + +@when("I execute the plan via the lifecycle service") +def step_execute_plan_lifecycle(context: Context) -> None: + plan = context.lifecycle_plan + context.lifecycle_plan = context.lifecycle_service.execute_plan( + plan.identity.plan_id + ) + + +@when("I apply the plan via the lifecycle service") +def step_apply_plan_lifecycle(context: Context) -> None: + plan = context.lifecycle_plan + context.lifecycle_plan = context.lifecycle_service.apply_plan(plan.identity.plan_id) + + +@then("the plan should be in Execute/QUEUED state") +def step_check_plan_execute_queued(context: Context) -> None: + assert context.lifecycle_plan.phase.value == "execute", ( + f"Expected execute, got {context.lifecycle_plan.phase.value}" + ) + assert context.lifecycle_plan.processing_state.value == "queued", ( + f"Expected queued, got {context.lifecycle_plan.processing_state.value}" + ) + + +@then("the plan should be in Apply/QUEUED state") +def step_check_plan_apply_queued(context: Context) -> None: + assert context.lifecycle_plan.phase.value == "apply", ( + f"Expected apply, got {context.lifecycle_plan.phase.value}" + ) + assert context.lifecycle_plan.processing_state.value == "queued", ( + f"Expected queued, got {context.lifecycle_plan.processing_state.value}" + ) + + +@then('the job store should contain {count:d} async job for phase "{phase}"') +def step_check_job_store_phase(context: Context, count: int, phase: str) -> None: + jobs = context.lifecycle_job_store.list_by_status(AsyncJobStatus.QUEUED) + matching = [j for j in jobs if j.phase == phase] + assert len(matching) == count, ( + f"Expected {count} job(s) for phase {phase}, got {len(matching)}" + ) + + +@then("the enqueued job plan_id should match the plan") +def step_check_enqueued_plan_id(context: Context) -> None: + jobs = context.lifecycle_job_store.list_all() + assert len(jobs) > 0, "No jobs in store" + plan_id = context.lifecycle_plan.identity.plan_id + assert any(j.plan_id == plan_id for j in jobs), ( + f"No job with plan_id {plan_id} found" + ) + + +@then("no async jobs should be in the job store") +def step_check_no_async_jobs(context: Context) -> None: + assert context.lifecycle_job_store.count() == 0, ( + f"Expected 0 jobs, got {context.lifecycle_job_store.count()}" + ) + + +# --------------------------------------------------------------------------- +# Review-fix steps: Error message redaction (P2) +# --------------------------------------------------------------------------- + + +@given("an async worker is initialised with a secret-leaking executor") +def step_worker_secret_leaking_executor(context: Context) -> None: + """Create a worker whose executor raises an error containing a secret.""" + + def leaking_executor(job: AsyncJob, token: CancellationToken) -> None: + raise RuntimeError( + "Connection failed: api_key=sk-proj-ABCDEFGHIJ1234567890abcdef" + ) + + config = AsyncWorkerConfig( + enabled=True, + max_workers=2, + poll_interval=0.1, + job_timeout=3600, + job_ttl=86400, + ) + context.job_store = InMemoryJobStore() + context.worker = AsyncWorker(config, context.job_store, leaking_executor) + context.async_job = None + context.error = None + + +@then("the async job error_message should not contain the raw secret") +def step_check_no_raw_secret(context: Context) -> None: + error_msg = context.async_job.error_message + assert error_msg is not None, "error_message is None" + assert "sk-proj-ABCDEFGHIJ1234567890abcdef" not in error_msg, ( + f"Raw secret found in error_message: {error_msg}" + ) + + +@then('the async job error_message should contain "***REDACTED***"') +def step_check_redacted_marker(context: Context) -> None: + error_msg = context.async_job.error_message + assert error_msg is not None, "error_message is None" + assert "***REDACTED***" in error_msg, ( + f"Expected ***REDACTED*** in error_message, got: {error_msg}" + ) diff --git a/features/steps/decision_di_wiring_steps.py b/features/steps/decision_di_wiring_steps.py index 7cb48c8cf..74ae88357 100644 --- a/features/steps/decision_di_wiring_steps.py +++ b/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( diff --git a/robot/async_execution.robot b/robot/async_execution.robot new file mode 100644 index 000000000..b06136e26 --- /dev/null +++ b/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 diff --git a/robot/helper_async_execution.py b/robot/helper_async_execution.py new file mode 100644 index 000000000..43b074988 --- /dev/null +++ b/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 + +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]} ", 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() diff --git a/robot/helper_decision_di.py b/robot/helper_decision_di.py index b711c2610..823a4bf57 100644 --- a/robot/helper_decision_di.py +++ b/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( diff --git a/src/cleveragents/application/services/async_worker.py b/src/cleveragents/application/services/async_worker.py new file mode 100644 index 000000000..47d159dc7 --- /dev/null +++ b/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 diff --git a/src/cleveragents/application/services/plan_lifecycle_service.py b/src/cleveragents/application/services/plan_lifecycle_service.py index 235601051..e8bde8aae 100644 --- a/src/cleveragents/application/services/plan_lifecycle_service.py +++ b/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: diff --git a/src/cleveragents/cli/commands/system.py b/src/cleveragents/cli/commands/system.py index bdf4e29b1..053131a27 100644 --- a/src/cleveragents/cli/commands/system.py +++ b/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) diff --git a/src/cleveragents/config/settings.py b/src/cleveragents/config/settings.py index fe770f912..830f32c12 100644 --- a/src/cleveragents/config/settings.py +++ b/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, diff --git a/src/cleveragents/domain/models/core/__init__.py b/src/cleveragents/domain/models/core/__init__.py index dc674a275..9fefdd1a5 100644 --- a/src/cleveragents/domain/models/core/__init__.py +++ b/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", ] diff --git a/src/cleveragents/domain/models/core/async_job.py b/src/cleveragents/domain/models/core/async_job.py new file mode 100644 index 000000000..531c6a95a --- /dev/null +++ b/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 diff --git a/src/cleveragents/infrastructure/database/models.py b/src/cleveragents/infrastructure/database/models.py index e29650793..80d271680 100644 --- a/src/cleveragents/infrastructure/database/models.py +++ b/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"), + ) diff --git a/vulture_whitelist.py b/vulture_whitelist.py index 00418ec16..36303264d 100644 --- a/vulture_whitelist.py +++ b/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