khyari hamza
229dbc8925
- add with_superseded_by() helper for frozen model mutation - document frozen model + superseded_by interaction in docstring - clarify sequence_number uniqueness is a persistence concern - add scenario for invalid corrects_decision_id ULID validation - add scenario for with_superseded_by copy behavior - type step helper dict as dict[str, Any]
557 lines
19 KiB
Python
557 lines
19 KiB
Python
"""Step definitions for decision_model.feature.
|
|
|
|
Tests the Decision, DecisionType, ContextSnapshot, ResourceRef,
|
|
and ArtifactRef domain models.
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
from typing import Any
|
|
|
|
from behave import given, then, when # type: ignore[import-untyped]
|
|
from behave.runner import Context
|
|
from pydantic import ValidationError
|
|
from ulid import ULID
|
|
|
|
from cleveragents.domain.models.core.decision import (
|
|
EXECUTE_TYPES,
|
|
STRATEGIZE_TYPES,
|
|
ArtifactRef,
|
|
ContextSnapshot,
|
|
Decision,
|
|
DecisionType,
|
|
ResourceRef,
|
|
)
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Helpers
|
|
# ---------------------------------------------------------------------------
|
|
|
|
_VALID_PLAN_ULID = str(ULID())
|
|
_VALID_PARENT_ULID = str(ULID())
|
|
_VALID_CORRECTED_ULID = str(ULID())
|
|
_VALID_SUPERSEDED_ULID = str(ULID())
|
|
|
|
|
|
def _make_decision(
|
|
plan_id: str = _VALID_PLAN_ULID,
|
|
parent_decision_id: str | None = None,
|
|
sequence_number: int = 0,
|
|
decision_type: DecisionType = DecisionType.PROMPT_DEFINITION,
|
|
question: str = "What approach should we take?",
|
|
chosen_option: str = "Build a REST API",
|
|
confidence_score: float | None = None,
|
|
context_snapshot: ContextSnapshot | None = None,
|
|
is_correction: bool = False,
|
|
corrects_decision_id: str | None = None,
|
|
correction_reason: str | None = None,
|
|
superseded_by: str | None = None,
|
|
artifacts_produced: list[ArtifactRef] | None = None,
|
|
) -> Decision:
|
|
kwargs: dict[str, Any] = {
|
|
"plan_id": plan_id,
|
|
"sequence_number": sequence_number,
|
|
"decision_type": decision_type,
|
|
"question": question,
|
|
"chosen_option": chosen_option,
|
|
"is_correction": is_correction,
|
|
}
|
|
if parent_decision_id is not None:
|
|
kwargs["parent_decision_id"] = parent_decision_id
|
|
if confidence_score is not None:
|
|
kwargs["confidence_score"] = confidence_score
|
|
if context_snapshot is not None:
|
|
kwargs["context_snapshot"] = context_snapshot
|
|
if corrects_decision_id is not None:
|
|
kwargs["corrects_decision_id"] = corrects_decision_id
|
|
if correction_reason is not None:
|
|
kwargs["correction_reason"] = correction_reason
|
|
if superseded_by is not None:
|
|
kwargs["superseded_by"] = superseded_by
|
|
if artifacts_produced is not None:
|
|
kwargs["artifacts_produced"] = artifacts_produced
|
|
return Decision(**kwargs)
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# DecisionType enum
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
@then("the DecisionType enum should have exactly {count:d} members")
|
|
def step_decision_type_count(context: Context, count: int) -> None:
|
|
assert len(DecisionType) == count, (
|
|
f"Expected {count} members, got {len(DecisionType)}"
|
|
)
|
|
|
|
|
|
@then('STRATEGIZE_TYPES should contain "{dtype}"')
|
|
def step_strategize_contains(context: Context, dtype: str) -> None:
|
|
assert DecisionType(dtype) in STRATEGIZE_TYPES, f"{dtype} not in STRATEGIZE_TYPES"
|
|
|
|
|
|
@then('STRATEGIZE_TYPES should not contain "{dtype}"')
|
|
def step_strategize_not_contains(context: Context, dtype: str) -> None:
|
|
assert DecisionType(dtype) not in STRATEGIZE_TYPES
|
|
|
|
|
|
@then("STRATEGIZE_TYPES should have exactly {count:d} members")
|
|
def step_strategize_count(context: Context, count: int) -> None:
|
|
assert len(STRATEGIZE_TYPES) == count
|
|
|
|
|
|
@then('EXECUTE_TYPES should contain "{dtype}"')
|
|
def step_execute_contains(context: Context, dtype: str) -> None:
|
|
assert DecisionType(dtype) in EXECUTE_TYPES, f"{dtype} not in EXECUTE_TYPES"
|
|
|
|
|
|
@then('EXECUTE_TYPES should not contain "{dtype}"')
|
|
def step_execute_not_contains(context: Context, dtype: str) -> None:
|
|
assert DecisionType(dtype) not in EXECUTE_TYPES
|
|
|
|
|
|
@then("EXECUTE_TYPES should have exactly {count:d} members")
|
|
def step_execute_count(context: Context, count: int) -> None:
|
|
assert len(EXECUTE_TYPES) == count
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Given steps
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
@given("a valid plan ULID")
|
|
def step_valid_plan_ulid(context: Context) -> None:
|
|
context.decision_plan_id = str(ULID())
|
|
|
|
|
|
@given("a valid parent decision ULID")
|
|
def step_valid_parent_ulid(context: Context) -> None:
|
|
context.decision_parent_id = str(ULID())
|
|
|
|
|
|
@given("a valid corrected decision ULID")
|
|
def step_valid_corrected_ulid(context: Context) -> None:
|
|
context.decision_corrected_id = str(ULID())
|
|
|
|
|
|
@given("a context snapshot with hash and resources")
|
|
def step_context_snapshot(context: Context) -> None:
|
|
context.decision_snapshot = ContextSnapshot(
|
|
hot_context_hash="sha256:abc123",
|
|
hot_context_ref="store://snapshots/abc123",
|
|
relevant_resources=[
|
|
ResourceRef(resource_id=str(ULID()), path="src/main.py"),
|
|
ResourceRef(resource_id=str(ULID())),
|
|
],
|
|
actor_state_ref="checkpoint://actor/001",
|
|
)
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# When steps — successful creation
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
@when("I create a prompt_definition decision with sequence {seq:d}")
|
|
def step_create_root(context: Context, seq: int) -> None:
|
|
context.decision_result = _make_decision(
|
|
plan_id=context.decision_plan_id,
|
|
sequence_number=seq,
|
|
decision_type=DecisionType.PROMPT_DEFINITION,
|
|
)
|
|
|
|
|
|
@when("I create a strategy_choice decision with sequence {seq:d} and a parent")
|
|
def step_create_strategy(context: Context, seq: int) -> None:
|
|
context.decision_result = _make_decision(
|
|
plan_id=context.decision_plan_id,
|
|
parent_decision_id=context.decision_parent_id,
|
|
sequence_number=seq,
|
|
decision_type=DecisionType.STRATEGY_CHOICE,
|
|
)
|
|
|
|
|
|
@when("I create an implementation_choice decision with sequence {seq:d} and a parent")
|
|
def step_create_impl(context: Context, seq: int) -> None:
|
|
context.decision_result = _make_decision(
|
|
plan_id=context.decision_plan_id,
|
|
parent_decision_id=context.decision_parent_id,
|
|
sequence_number=seq,
|
|
decision_type=DecisionType.IMPLEMENTATION_CHOICE,
|
|
)
|
|
|
|
|
|
@when("I create a user_intervention decision with sequence {seq:d}")
|
|
def step_create_user_intervention(context: Context, seq: int) -> None:
|
|
context.decision_result = _make_decision(
|
|
plan_id=context.decision_plan_id,
|
|
sequence_number=seq,
|
|
decision_type=DecisionType.USER_INTERVENTION,
|
|
)
|
|
|
|
|
|
@when("I create a decision with confidence score {score}")
|
|
def step_create_with_confidence(context: Context, score: str) -> None:
|
|
conf = None if score == "None" else float(score)
|
|
context.decision_result = _make_decision(
|
|
plan_id=context.decision_plan_id,
|
|
confidence_score=conf,
|
|
decision_type=DecisionType.STRATEGY_CHOICE,
|
|
)
|
|
|
|
|
|
@when("I create a correction decision")
|
|
def step_create_correction(context: Context) -> None:
|
|
context.decision_result = _make_decision(
|
|
plan_id=context.decision_plan_id,
|
|
decision_type=DecisionType.STRATEGY_CHOICE,
|
|
is_correction=True,
|
|
corrects_decision_id=context.decision_corrected_id,
|
|
correction_reason="Original approach was suboptimal",
|
|
)
|
|
|
|
|
|
@when("I create a decision that is superseded")
|
|
def step_create_superseded(context: Context) -> None:
|
|
context.decision_result = _make_decision(
|
|
plan_id=context.decision_plan_id,
|
|
decision_type=DecisionType.STRATEGY_CHOICE,
|
|
superseded_by=str(ULID()),
|
|
)
|
|
|
|
|
|
@when("I create a decision with the context snapshot")
|
|
def step_create_with_snapshot(context: Context) -> None:
|
|
context.decision_result = _make_decision(
|
|
plan_id=context.decision_plan_id,
|
|
decision_type=DecisionType.STRATEGY_CHOICE,
|
|
context_snapshot=context.decision_snapshot,
|
|
)
|
|
|
|
|
|
@when("I create a decision with artifacts")
|
|
def step_create_with_artifacts(context: Context) -> None:
|
|
context.decision_result = _make_decision(
|
|
plan_id=context.decision_plan_id,
|
|
decision_type=DecisionType.STRATEGY_CHOICE,
|
|
artifacts_produced=[
|
|
ArtifactRef(artifact_path="src/api.py", artifact_type="file"),
|
|
ArtifactRef(artifact_path="patches/fix.patch", artifact_type="patch"),
|
|
],
|
|
)
|
|
|
|
|
|
@when("I create a fully populated decision")
|
|
def step_create_fully_populated(context: Context) -> None:
|
|
context.decision_result = _make_decision(
|
|
plan_id=context.decision_plan_id,
|
|
parent_decision_id=str(ULID()),
|
|
sequence_number=5,
|
|
decision_type=DecisionType.STRATEGY_CHOICE,
|
|
question="Which framework?",
|
|
chosen_option="FastAPI",
|
|
confidence_score=0.92,
|
|
context_snapshot=ContextSnapshot(
|
|
hot_context_hash="sha256:def456",
|
|
hot_context_ref="store://snapshots/def456",
|
|
relevant_resources=[ResourceRef(resource_id=str(ULID()))],
|
|
actor_state_ref="checkpoint://actor/005",
|
|
),
|
|
artifacts_produced=[ArtifactRef(artifact_path="src/app.py")],
|
|
)
|
|
|
|
|
|
@when('I create a decision of type "{dtype}"')
|
|
def step_create_any_type(context: Context, dtype: str) -> None:
|
|
dt = DecisionType(dtype)
|
|
parent = None if dt == DecisionType.PROMPT_DEFINITION else str(ULID())
|
|
context.decision_result = _make_decision(
|
|
plan_id=context.decision_plan_id,
|
|
parent_decision_id=parent,
|
|
decision_type=dt,
|
|
)
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# When steps — expected failures
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
@when('I try to create a decision with plan_id "{plan_id}"')
|
|
def step_try_invalid_plan_id(context: Context, plan_id: str) -> None:
|
|
try:
|
|
_make_decision(plan_id=plan_id)
|
|
context.decision_error = None
|
|
except ValidationError as exc:
|
|
context.decision_error = exc
|
|
|
|
|
|
@when('I try to create a decision with parent_decision_id "{parent_id}"')
|
|
def step_try_invalid_parent(context: Context, parent_id: str) -> None:
|
|
try:
|
|
_make_decision(
|
|
plan_id=context.decision_plan_id,
|
|
parent_decision_id=parent_id,
|
|
decision_type=DecisionType.STRATEGY_CHOICE,
|
|
)
|
|
context.decision_error = None
|
|
except ValidationError as exc:
|
|
context.decision_error = exc
|
|
|
|
|
|
@when("I try to create a prompt_definition with a parent")
|
|
def step_try_root_with_parent(context: Context) -> None:
|
|
try:
|
|
_make_decision(
|
|
plan_id=context.decision_plan_id,
|
|
parent_decision_id=context.decision_parent_id,
|
|
decision_type=DecisionType.PROMPT_DEFINITION,
|
|
)
|
|
context.decision_error = None
|
|
except ValidationError as exc:
|
|
context.decision_error = exc
|
|
|
|
|
|
@when("I try to create a decision with confidence score {score}")
|
|
def step_try_bad_confidence(context: Context, score: str) -> None:
|
|
try:
|
|
_make_decision(
|
|
plan_id=context.decision_plan_id,
|
|
confidence_score=float(score),
|
|
decision_type=DecisionType.STRATEGY_CHOICE,
|
|
)
|
|
context.decision_error = None
|
|
except ValidationError as exc:
|
|
context.decision_error = exc
|
|
|
|
|
|
@when("I try to create a decision with is_correction true but no corrects_decision_id")
|
|
def step_try_correction_no_target(context: Context) -> None:
|
|
try:
|
|
_make_decision(
|
|
plan_id=context.decision_plan_id,
|
|
decision_type=DecisionType.STRATEGY_CHOICE,
|
|
is_correction=True,
|
|
)
|
|
context.decision_error = None
|
|
except ValidationError as exc:
|
|
context.decision_error = exc
|
|
|
|
|
|
@when("I try to create a decision with corrects_decision_id but is_correction false")
|
|
def step_try_correction_flag_mismatch(context: Context) -> None:
|
|
try:
|
|
_make_decision(
|
|
plan_id=context.decision_plan_id,
|
|
decision_type=DecisionType.STRATEGY_CHOICE,
|
|
is_correction=False,
|
|
corrects_decision_id=context.decision_corrected_id,
|
|
)
|
|
context.decision_error = None
|
|
except ValidationError as exc:
|
|
context.decision_error = exc
|
|
|
|
|
|
@when('I try to create a decision with superseded_by "{value}"')
|
|
def step_try_bad_superseded(context: Context, value: str) -> None:
|
|
try:
|
|
_make_decision(
|
|
plan_id=context.decision_plan_id,
|
|
decision_type=DecisionType.STRATEGY_CHOICE,
|
|
superseded_by=value,
|
|
)
|
|
context.decision_error = None
|
|
except ValidationError as exc:
|
|
context.decision_error = exc
|
|
|
|
|
|
@when('I try to create a decision with corrects_decision_id ULID "{value}"')
|
|
def step_try_bad_corrects_id(context: Context, value: str) -> None:
|
|
try:
|
|
_make_decision(
|
|
plan_id=context.decision_plan_id,
|
|
decision_type=DecisionType.STRATEGY_CHOICE,
|
|
is_correction=True,
|
|
corrects_decision_id=value,
|
|
)
|
|
context.decision_error = None
|
|
except ValidationError as exc:
|
|
context.decision_error = exc
|
|
|
|
|
|
@when("I call with_superseded_by on the decision")
|
|
def step_call_with_superseded_by(context: Context) -> None:
|
|
context.decision_superseded_copy = context.decision_result.with_superseded_by(
|
|
str(ULID())
|
|
)
|
|
|
|
|
|
@when("I try to create a ResourceRef with empty resource_id")
|
|
def step_try_empty_resource_ref(context: Context) -> None:
|
|
try:
|
|
ResourceRef(resource_id="")
|
|
context.decision_error = None
|
|
except ValidationError as exc:
|
|
context.decision_error = exc
|
|
|
|
|
|
@when("I try to create an ArtifactRef with empty artifact_path")
|
|
def step_try_empty_artifact_ref(context: Context) -> None:
|
|
try:
|
|
ArtifactRef(artifact_path="")
|
|
context.decision_error = None
|
|
except ValidationError as exc:
|
|
context.decision_error = exc
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Then steps
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
@then("the decision should be created successfully")
|
|
def step_decision_created(context: Context) -> None:
|
|
assert context.decision_result is not None
|
|
assert isinstance(context.decision_result, Decision)
|
|
|
|
|
|
@then("the decision should be a root decision")
|
|
def step_is_root(context: Context) -> None:
|
|
assert context.decision_result.is_root
|
|
|
|
|
|
@then("the decision should not be a root decision")
|
|
def step_not_root(context: Context) -> None:
|
|
assert not context.decision_result.is_root
|
|
|
|
|
|
@then("the decision should have a valid ULID as decision_id")
|
|
def step_valid_decision_id(context: Context) -> None:
|
|
import re
|
|
|
|
pattern = r"^[0-9A-HJKMNP-TV-Z]{26}$"
|
|
assert re.match(pattern, context.decision_result.decision_id), (
|
|
f"Invalid ULID: {context.decision_result.decision_id}"
|
|
)
|
|
|
|
|
|
@then('the decision type should be "{dtype}"')
|
|
def step_decision_type_check(context: Context, dtype: str) -> None:
|
|
assert context.decision_result.decision_type == DecisionType(dtype)
|
|
|
|
|
|
@then("the decision is_strategize_type should be true")
|
|
def step_is_strategize(context: Context) -> None:
|
|
assert context.decision_result.is_strategize_type
|
|
|
|
|
|
@then("the decision is_strategize_type should be false")
|
|
def step_not_strategize(context: Context) -> None:
|
|
assert not context.decision_result.is_strategize_type
|
|
|
|
|
|
@then("the decision is_execute_type should be true")
|
|
def step_is_execute(context: Context) -> None:
|
|
assert context.decision_result.is_execute_type
|
|
|
|
|
|
@then("the decision is_execute_type should be false")
|
|
def step_not_execute(context: Context) -> None:
|
|
assert not context.decision_result.is_execute_type
|
|
|
|
|
|
@then("the decision is_any_phase_type should be true")
|
|
def step_is_any_phase(context: Context) -> None:
|
|
assert context.decision_result.is_any_phase_type
|
|
|
|
|
|
@then("the decision confidence score should be {score}")
|
|
def step_confidence_value(context: Context, score: str) -> None:
|
|
if score == "None":
|
|
assert context.decision_result.confidence_score is None
|
|
else:
|
|
assert context.decision_result.confidence_score == float(score)
|
|
|
|
|
|
@then("the decision is_correction should be true")
|
|
def step_is_correction(context: Context) -> None:
|
|
assert context.decision_result.is_correction
|
|
|
|
|
|
@then("the decision is_superseded should be true")
|
|
def step_is_superseded(context: Context) -> None:
|
|
assert context.decision_result.is_superseded
|
|
|
|
|
|
@then("the original decision should not be superseded")
|
|
def step_original_not_superseded(context: Context) -> None:
|
|
assert not context.decision_result.is_superseded
|
|
|
|
|
|
@then("the superseded copy should be superseded")
|
|
def step_copy_is_superseded(context: Context) -> None:
|
|
assert context.decision_superseded_copy.is_superseded
|
|
assert context.decision_superseded_copy.decision_id == (
|
|
context.decision_result.decision_id
|
|
)
|
|
|
|
|
|
@then("a decision validation error should be raised")
|
|
def step_validation_error(context: Context) -> None:
|
|
assert context.decision_error is not None, (
|
|
"Expected ValidationError but none raised"
|
|
)
|
|
assert isinstance(context.decision_error, ValidationError)
|
|
|
|
|
|
@then('the decision error should mention "{text}"')
|
|
def step_error_mentions(context: Context, text: str) -> None:
|
|
assert text.lower() in str(context.decision_error).lower(), (
|
|
f"Expected '{text}' in error: {context.decision_error}"
|
|
)
|
|
|
|
|
|
@then("the decision context snapshot hash should not be empty")
|
|
def step_snapshot_hash_not_empty(context: Context) -> None:
|
|
assert context.decision_result.context_snapshot.hot_context_hash
|
|
|
|
|
|
@then("the decision context snapshot should have resources")
|
|
def step_snapshot_has_resources(context: Context) -> None:
|
|
assert len(context.decision_result.context_snapshot.relevant_resources) > 0
|
|
|
|
|
|
@then("the decision context snapshot hash should be empty")
|
|
def step_snapshot_hash_empty(context: Context) -> None:
|
|
assert context.decision_result.context_snapshot.hot_context_hash == ""
|
|
|
|
|
|
@then("the decision should have {count:d} artifacts produced")
|
|
def step_artifact_count(context: Context, count: int) -> None:
|
|
assert len(context.decision_result.artifacts_produced) == count
|
|
|
|
|
|
@then("the decision should round-trip through model_dump and model_validate")
|
|
def step_roundtrip(context: Context) -> None:
|
|
data = context.decision_result.model_dump()
|
|
restored = Decision.model_validate(data)
|
|
assert restored.decision_id == context.decision_result.decision_id
|
|
assert restored.decision_type == context.decision_result.decision_type
|
|
assert restored.question == context.decision_result.question
|
|
assert restored.chosen_option == context.decision_result.chosen_option
|
|
assert restored.confidence_score == context.decision_result.confidence_score
|
|
assert (
|
|
restored.context_snapshot.hot_context_hash
|
|
== context.decision_result.context_snapshot.hot_context_hash
|
|
)
|
|
assert len(restored.artifacts_produced) == len(
|
|
context.decision_result.artifacts_produced
|
|
)
|
|
|
|
|
|
@then('as_cli_dict should contain key "{key}"')
|
|
def step_cli_dict_key(context: Context, key: str) -> None:
|
|
cli = context.decision_result.as_cli_dict()
|
|
assert key in cli, f"Key '{key}' not in as_cli_dict: {list(cli.keys())}"
|