From 14ccf07ac2cd8bb3063455b72da124d4dd3500b1 Mon Sep 17 00:00:00 2001 From: CleverThis Date: Wed, 3 Jun 2026 13:30:40 -0400 Subject: [PATCH] fix(plans): derive ULID-compatible subplan IDs in scheduler BDD tests SubplanStatus.subplan_id is pydantic-validated against ^[0-9A-HJKMNP-TV-Z]{26}$. The scheduler test fixtures constructed SubplanStatus instances with short logical IDs ("subplan-001", "subplan-A") which failed validation at fixture-construction time, erroring 22 of 38 originally-failing scenarios in unit_tests CI before the behavioural assertions could even run. Derive a deterministic 26-char Crockford-Base32 ID from each logical name via SHA-256 and translate fail-id sets, block-second dicts, dependency graphs, and result-status lookups through the same helper so cross-references stay consistent. Also add the missing step definitions unique to the parallel_subplan_scheduler scenarios (staggered-completion fixture, retry-then-succeed fixture, fail_fast- disabled scheduler, mode-only scheduler, timeout-errored verifier) and remove duplicate @then registrations that conflict with shared step definitions in subplan_execution_steps.py. ISSUES CLOSED: #9555 --- .../steps/parallel_subplan_scheduler_steps.py | 194 +++++++++++++++--- 1 file changed, 164 insertions(+), 30 deletions(-) diff --git a/features/steps/parallel_subplan_scheduler_steps.py b/features/steps/parallel_subplan_scheduler_steps.py index eaa985a2f..d04b4c079 100644 --- a/features/steps/parallel_subplan_scheduler_steps.py +++ b/features/steps/parallel_subplan_scheduler_steps.py @@ -2,8 +2,9 @@ from __future__ import annotations -from datetime import UTC, datetime +import hashlib import time +from datetime import UTC, datetime from typing import Any from behave import given, then, when @@ -28,14 +29,27 @@ from cleveragents.domain.models.core.plan import ( # --- Fixtures and helpers --- +# Crockford Base32 alphabet (ULID charset: digits + uppercase letters minus I/L/O/U). +_CROCKFORD_ULID_CHARS = "0123456789ABCDEFGHJKMNPQRSTVWXYZ" + + +def _to_ulid_id(logical_name: str) -> str: + # SubplanStatus.subplan_id is validated against ^[0-9A-HJKMNP-TV-Z]{26}$. + # Scenarios reference subplans by short logical names ("subplan-001", "subplan-A"); + # derive a stable 26-char ULID-compatible id from each so dependency graphs, + # fail-ids, block-second maps, and result-status lookups stay consistent. + digest = hashlib.sha256(logical_name.encode()).digest() + return "".join(_CROCKFORD_ULID_CHARS[b & 0x1F] for b in digest[:26]) + + def create_subplan_status( - subplan_id: str, + logical_name: str, action_name: str = "test/action", status: ProcessingState = ProcessingState.QUEUED, ) -> SubplanStatus: - """Create a test subplan status.""" + """Create a test subplan status from a logical name.""" return SubplanStatus( - subplan_id=subplan_id, + subplan_id=_to_ulid_id(logical_name), action_name=action_name, status=status, ) @@ -252,7 +266,7 @@ def step_create_subplans_with_tracking(context: Any, count: int) -> None: def step_create_subplans_with_failure(context: Any, count: int) -> None: """Create subplans where the second fails.""" context.subplans = [create_subplan_status(f"subplan-{i:03d}") for i in range(count)] - fail_ids = {"subplan-001"} + fail_ids = {_to_ulid_id("subplan-001")} context.scheduler._executor_fn = create_executor_fn(context, fail_ids=fail_ids) @@ -260,7 +274,7 @@ def step_create_subplans_with_failure(context: Any, count: int) -> None: def step_create_subplans_with_first_failure(context: Any, count: int) -> None: """Create subplans where the first fails.""" context.subplans = [create_subplan_status(f"subplan-{i:03d}") for i in range(count)] - fail_ids = {"subplan-000"} + fail_ids = {_to_ulid_id("subplan-000")} context.scheduler._executor_fn = create_executor_fn(context, fail_ids=fail_ids) @@ -268,7 +282,7 @@ def step_create_subplans_with_first_failure(context: Any, count: int) -> None: def step_create_subplans_with_blocking(context: Any, count: int, seconds: int) -> None: """Create subplans where the first blocks.""" context.subplans = [create_subplan_status(f"subplan-{i:03d}") for i in range(count)] - block_seconds = {"subplan-000": seconds} + block_seconds = {_to_ulid_id("subplan-000"): float(seconds)} context.scheduler._executor_fn = create_executor_fn( context, block_seconds=block_seconds ) @@ -282,7 +296,7 @@ def step_create_subplans_with_blocking_and_quick( ) -> None: """Create subplans where first blocks and second completes quickly.""" context.subplans = [create_subplan_status(f"subplan-{i:03d}") for i in range(count)] - block_seconds = {"subplan-000": seconds} + block_seconds = {_to_ulid_id("subplan-000"): float(seconds)} context.scheduler._executor_fn = create_executor_fn( context, block_seconds=block_seconds ) @@ -306,8 +320,9 @@ def step_create_valid_scheduler(context: Any) -> None: def step_create_subplans_staggered(context: Any, count: int) -> None: """Create subplans with staggered completion.""" context.subplans = [create_subplan_status(f"subplan-{i:03d}") for i in range(count)] - # Stagger completion times - block_seconds = {f"subplan-{i:03d}": i * 0.1 for i in range(count)} + block_seconds = { + _to_ulid_id(f"subplan-{i:03d}"): i * 0.1 for i in range(count) + } context.scheduler._executor_fn = create_executor_fn( context, block_seconds=block_seconds ) @@ -331,7 +346,7 @@ def step_create_subplans_with_multiple_failures( ) -> None: """Create subplans where multiple will fail.""" context.subplans = [create_subplan_status(f"subplan-{i:03d}") for i in range(count)] - fail_ids = {f"subplan-{i:03d}" for i in range(fail_count)} + fail_ids = {_to_ulid_id(f"subplan-{i:03d}") for i in range(fail_count)} context.scheduler._executor_fn = create_executor_fn(context, fail_ids=fail_ids) @@ -342,9 +357,9 @@ def step_create_subplans_with_dependencies(context: Any, count: int) -> None: create_subplan_status(f"subplan-{chr(65 + i)}") for i in range(count) ] context.dependency_graph = { - "subplan-A": [], - "subplan-B": ["subplan-A"], - "subplan-C": ["subplan-B"], + _to_ulid_id("subplan-A"): [], + _to_ulid_id("subplan-B"): [_to_ulid_id("subplan-A")], + _to_ulid_id("subplan-C"): [_to_ulid_id("subplan-B")], } @@ -357,10 +372,13 @@ def step_create_subplans_with_complex_dependencies(context: Any, count: int) -> create_subplan_status(f"subplan-{chr(65 + i)}") for i in range(count) ] context.dependency_graph = { - "subplan-A": [], - "subplan-B": ["subplan-A"], - "subplan-C": ["subplan-A"], - "subplan-D": ["subplan-B", "subplan-C"], + _to_ulid_id("subplan-A"): [], + _to_ulid_id("subplan-B"): [_to_ulid_id("subplan-A")], + _to_ulid_id("subplan-C"): [_to_ulid_id("subplan-A")], + _to_ulid_id("subplan-D"): [ + _to_ulid_id("subplan-B"), + _to_ulid_id("subplan-C"), + ], } @@ -373,9 +391,12 @@ def step_create_subplans_with_wave_dependencies(context: Any) -> None: create_subplan_status("subplan-C"), ] context.dependency_graph = { - "subplan-A": [], - "subplan-B": [], - "subplan-C": ["subplan-A", "subplan-B"], + _to_ulid_id("subplan-A"): [], + _to_ulid_id("subplan-B"): [], + _to_ulid_id("subplan-C"): [ + _to_ulid_id("subplan-A"), + _to_ulid_id("subplan-B"), + ], } @@ -635,7 +656,17 @@ def step_verify_is_running_true(context: Any) -> None: @then("after execution completes, is_running should be false") def step_verify_is_running_false(context: Any) -> None: - """Verify is_running is false after completion.""" + """Run scheduler to completion and verify is_running becomes false. + + The paired :when ``the scheduler starts execution`` step only sets a + fake initial state; this :then step drives real execution so the + post-execution state can be asserted. + """ + context.result = context.scheduler.schedule( + subplan_statuses=context.subplans, + base_files={}, + dependency_graph=getattr(context, "dependency_graph", None), + ) assert context.scheduler.state.is_running is False @@ -754,7 +785,9 @@ def step_verify_a_before_b(context: Any) -> None: """Verify A completes before B.""" assert context.result is not None status_map = {s.subplan_id: s for s in context.result.statuses} - assert status_map["subplan-A"].completed_at < status_map["subplan-B"].completed_at + a_id = _to_ulid_id("subplan-A") + b_id = _to_ulid_id("subplan-B") + assert status_map[a_id].completed_at < status_map[b_id].completed_at @then("subplan B should complete before subplan C") @@ -762,7 +795,9 @@ def step_verify_b_before_c(context: Any) -> None: """Verify B completes before C.""" assert context.result is not None status_map = {s.subplan_id: s for s in context.result.statuses} - assert status_map["subplan-B"].completed_at < status_map["subplan-C"].completed_at + b_id = _to_ulid_id("subplan-B") + c_id = _to_ulid_id("subplan-C") + assert status_map[b_id].completed_at < status_map[c_id].completed_at @then("the dependency order should be respected") @@ -770,14 +805,16 @@ def step_verify_dependency_order(context: Any) -> None: """Verify dependency order is respected.""" assert context.result is not None status_map = {s.subplan_id: s for s in context.result.statuses} + a_id = _to_ulid_id("subplan-A") + b_id = _to_ulid_id("subplan-B") + c_id = _to_ulid_id("subplan-C") + d_id = _to_ulid_id("subplan-D") - # Verify A before B and C - assert status_map["subplan-A"].completed_at < status_map["subplan-B"].completed_at - assert status_map["subplan-A"].completed_at < status_map["subplan-C"].completed_at + assert status_map[a_id].completed_at < status_map[b_id].completed_at + assert status_map[a_id].completed_at < status_map[c_id].completed_at - # Verify B and C before D - assert status_map["subplan-B"].completed_at < status_map["subplan-D"].completed_at - assert status_map["subplan-C"].completed_at < status_map["subplan-D"].completed_at + assert status_map[b_id].completed_at < status_map[d_id].completed_at + assert status_map[c_id].completed_at < status_map[d_id].completed_at @then("the peak concurrent execution should be at least {min_concurrent:d}") @@ -821,3 +858,100 @@ def step_verify_both_succeed(context: Any) -> None: assert len(context.result.statuses) == 2 for status in context.result.statuses: assert status.status == ProcessingState.COMPLETE + + +# --- Additional Given steps --- + + +@given("{count:d} subplans to execute with staggered completion") +def step_create_subplans_to_execute_staggered(context: Any, count: int) -> None: + """Create subplans with staggered completion.""" + context.subplans = [create_subplan_status(f"subplan-{i:03d}") for i in range(count)] + block_seconds = { + _to_ulid_id(f"subplan-{i:03d}"): i * 0.05 for i in range(count) + } + context.scheduler._executor_fn = create_executor_fn( + context, block_seconds=block_seconds + ) + + +@given( + "{count:d} subplans where the first will fail once with TimeoutError then succeed" +) +def step_create_subplans_retry_then_succeed(context: Any, count: int) -> None: + """Create subplans where first fails once then succeeds (retry scenario).""" + context.subplans = [create_subplan_status(f"subplan-{i:03d}") for i in range(count)] + first_id = _to_ulid_id("subplan-000") + # Stateful executor: fails first call for first_id, succeeds thereafter. + fail_state = {"called_for_first": False} + + def executor(status: SubplanStatus) -> SubplanExecutionOutput: + if status.subplan_id == first_id and not fail_state["called_for_first"]: + fail_state["called_for_first"] = True + return SubplanExecutionOutput( + subplan_id=status.subplan_id, + success=False, + error="TimeoutError: simulated timeout", + ) + return SubplanExecutionOutput( + subplan_id=status.subplan_id, + success=True, + files={"test.txt": "content"}, + files_changed=1, + ) + + context.scheduler._executor_fn = executor + + +@given( + "a parallel subplan scheduler with max_parallel {max_parallel:d} with fail_fast disabled" +) +def step_create_scheduler_with_fail_fast_disabled( + context: Any, max_parallel: int +) -> None: + """Create a scheduler with fail_fast disabled.""" + config = SubplanConfig( + execution_mode=ExecutionMode.PARALLEL, + max_parallel=max_parallel, + fail_fast=False, + ) + executor_fn = create_executor_fn(context) + context.scheduler = ParallelSubplanScheduler( + config=config, + executor_fn=executor_fn, + ) + + +@given("a parallel subplan scheduler in {mode} mode") +def step_create_scheduler_mode_only(context: Any, mode: str) -> None: + """Create a scheduler with specified execution mode (default max_parallel).""" + mode_map = { + "SEQUENTIAL": ExecutionMode.SEQUENTIAL, + "PARALLEL": ExecutionMode.PARALLEL, + "DEPENDENCY_ORDERED": ExecutionMode.DEPENDENCY_ORDERED, + } + config = SubplanConfig( + execution_mode=mode_map.get(mode, ExecutionMode.PARALLEL), + ) + executor_fn = create_executor_fn(context) + context.scheduler = ParallelSubplanScheduler( + config=config, + executor_fn=executor_fn, + ) + + +# --- Additional Then steps --- +# Most error-style assertions (`a config validation error should be raised`, +# `the first subplan should complete successfully`, `the execution result +# should report all succeeded`, etc.) are shared with subplan_execution_steps.py +# via behave's global step registry — they live there, not here. Only steps +# unique to the parallel_subplan_scheduler scenarios are defined below. + + +@then("the first subplan should be errored with timeout") +def step_verify_first_errored_timeout(context: Any) -> None: + """Verify first subplan errored with timeout.""" + assert context.result is not None + first = context.result.statuses[0] + assert first.status == ProcessingState.ERRORED + assert "timeout" in (first.error or "").lower()