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

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()