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
This commit is contained in:
2026-06-03 13:30:40 -04:00
committed by Forgejo
parent ebb543a9c3
commit 428ec07951
@@ -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()