diff --git a/backend/src/services/dashboard_testing/scenario/resolver.py b/backend/src/services/dashboard_testing/scenario/resolver.py new file mode 100644 index 000000000..978f4d2ee --- /dev/null +++ b/backend/src/services/dashboard_testing/scenario/resolver.py @@ -0,0 +1,148 @@ +# #region ScenarioGraph.Resolver.Resolve [C:4] [TYPE Function] [SEMANTICS scenario,resolve,parameter,revision] +# @ingroup ScenarioGraph +# @BRIEF Apply typed parameter/selector/manual resolutions and emit an immutable scenario revision. +# @PRE Base revision hash matches; changes target declared unresolved items. +# @POST Unrelated step ids/order remain unchanged; new parent/revision hashes link revisions. +# @SIDE_EFFECT None. +# @SIDE_EFFECT Logging (REASON/REFLECT markers required around revision emission). +# @RATIONALE Immutable revisions preserve auditability and byte-stable determinism for downstream pack compilation. +# @REJECTED In-place graph mutation — destroys revision history and breaks parent_revision_hash linkage. +# @DATA_CONTRACT ResolveScenarioRequest + BaseScenario -> DashboardTestScenario + +from __future__ import annotations + +from dataclasses import dataclass +from typing import Any + +from pydantic import ValidationError +from ss_tools.shared.cot_logger import log + +from src.core.logger import belief_scope +from src.services.dashboard_testing.scenario.models import DashboardTestScenario, ScenarioParameter +from src.services.dashboard_testing.scenario.serializer import serialize_json, sha256_hex + +_PARAM_TYPES = {"string": str, "integer": int, "decimal": float, "boolean": bool, "date": str, "datetime": str, "enum": str, "string_list": list} + + +@dataclass(frozen=True) +class ResolveChange: + kind: str # parameter | selector | manual_conversion | remove_step + target: str + value: Any + reason: str | None = None + + +# #region ScenarioGraph.Resolver.ValidateBaseRevision [C:2] [TYPE Function] [SEMANTICS scenario,resolve,revision] +# @ingroup ScenarioGraph +# @BRIEF Ensure the base revision hash matches the scenario under resolution. +# @POST Raises ValueError for stale base revisions. +def _check_base_revision(scenario: DashboardTestScenario, base_revision_hash: str | None) -> None: + if base_revision_hash is not None and base_revision_hash != scenario.revision_hash: + raise ValueError(f"stale base revision {base_revision_hash[:16]} != {scenario.revision_hash[:16]}") +# #endregion ScenarioGraph.Resolver.ValidateBaseRevision + + +# #region ScenarioGraph.Resolver.ApplyParameter [C:3] [TYPE Function] [SEMANTICS scenario,resolve,parameter] +# @ingroup ScenarioGraph +# @BRIEF Apply a typed parameter value; enforce type from the parameter spec. +# @POST Returns updated parameter list; raises ValueError/ValidationError on type mismatch. +def _apply_parameter(params: list[ScenarioParameter], target: str, value: Any) -> list[ScenarioParameter]: + updated = [p.model_copy(deep=True) for p in params] + found = False + for p in updated: + if p.name == target: + found = True + expected = _PARAM_TYPES.get(p.type) + if expected is not None and value is not None and not isinstance(value, expected): + raise ValidationError.from_exception_data("parameter", []) + p.value = value + p.status = "resolved" + if not found: + raise ValueError(f"unknown parameter {target!r}") + return updated +# #endregion ScenarioGraph.Resolver.ApplyParameter + + +# #region ScenarioGraph.Resolver.ApplySelector [C:3] [TYPE Function] [SEMANTICS scenario,resolve,selector] +# @ingroup ScenarioGraph +# @BRIEF Apply a selector hint to matching needs_selector steps. +# @POST Returns updated steps with needs_selector steps flipped to ready when a hint is supplied; +# the hint value is recorded in the step description for auditability. +def _apply_selector(steps: list[Any], target: str, value: str) -> list[Any]: + updated = [s.model_copy(deep=True) for s in steps] + found = False + for s in updated: + if s.id == target: + found = True + if s.automation_status == "needs_selector": + s.automation_status = "ready" + s.description = (s.description + " | selector_hint: " + value).strip() + if not found: + raise ValueError(f"unknown step {target!r}") + return updated +# #endregion ScenarioGraph.Resolver.ApplySelector + + +# #region ScenarioGraph.Resolver.ApplyManualConversion [C:2] [TYPE Function] [SEMANTICS scenario,resolve,manual] +# @ingroup ScenarioGraph +# @BRIEF Convert a step to a human checkpoint (manual conversion resolution). +# @POST Returns updated steps with the target step marked manual. +def _apply_manual_conversion(steps: list[Any], target: str) -> list[Any]: + updated = [s.model_copy(deep=True) for s in steps] + for s in updated: + if s.id == target: + s.automation_status = "manual" + s.risk = "human" + return updated + raise ValueError(f"unknown step {target!r}") +# #endregion ScenarioGraph.Resolver.ApplyManualConversion + + +# #region ScenarioGraph.Resolver.ApplyRemoveStep [C:2] [TYPE Function] [SEMANTICS scenario,resolve,remove] +# @ingroup ScenarioGraph +# @BRIEF Remove a step from the graph (explicit resolution operation). +# @POST Returns updated steps without the target step. +def _apply_remove_step(steps: list[Any], target: str) -> list[Any]: + updated = [s.model_copy(deep=True) for s in steps if s.id != target] + if len(updated) == len(steps): + raise ValueError(f"unknown step {target!r}") + return updated +# #endregion ScenarioGraph.Resolver.ApplyRemoveStep + + +def resolve_scenario( + scenario: DashboardTestScenario, + changes: list[ResolveChange], + *, + base_revision_hash: str | None = None, +) -> DashboardTestScenario: + """Apply typed resolutions and emit a new immutable revision linked via parent_revision_hash.""" + log("ScenarioGraph.Resolver.Resolve", "REASON", "Resolving scenario", + {"scenario_id": scenario.scenario_id, "changes": len(changes)}) + with belief_scope("ScenarioGraph.Resolver.Resolve", "Applying resolutions"): + _check_base_revision(scenario, base_revision_hash) + + params = [p.model_copy(deep=True) for p in scenario.parameters] + steps = [s.model_copy(deep=True) for s in scenario.steps] + + for change in changes: + if change.kind == "parameter": + params = _apply_parameter(params, change.target, change.value) + elif change.kind == "selector": + steps = _apply_selector(steps, change.target, str(change.value)) + elif change.kind == "manual_conversion": + steps = _apply_manual_conversion(steps, change.target) + elif change.kind == "remove_step": + steps = _apply_remove_step(steps, change.target) + else: + raise ValueError(f"invalid change kind {change.kind!r}") + + resolved = scenario.model_copy(deep=True) + resolved.parameters = params + resolved.steps = steps + resolved.parent_revision_hash = scenario.revision_hash + resolved.revision_hash = sha256_hex(serialize_json(resolved)) + log("ScenarioGraph.Resolver.Resolve", "REFLECT", "Scenario resolved", + {"new_revision": resolved.revision_hash[:16]}) + return resolved +# #endregion ScenarioGraph.Resolver.Resolve diff --git a/backend/tests/services/dashboard_testing/scenario/test_resolver.py b/backend/tests/services/dashboard_testing/scenario/test_resolver.py new file mode 100644 index 000000000..a1286bd1b --- /dev/null +++ b/backend/tests/services/dashboard_testing/scenario/test_resolver.py @@ -0,0 +1,66 @@ +# #region Test.Scenario.Resolver [C:3] [TYPE Module] [SEMANTICS testing,scenario,resolver,revision] +# @defgroup Test.Scenario Resolver immutable-revision tests. +# @LAYER Test +# @RELATION BINDS_TO -> [ScenarioGraph.Resolver.Resolve] +# @RATIONALE Resolution produces immutable revisions; unrelated structure must never change. +# @REJECTED Testing only happy-path resolution — would hide stale-base and unrelated-change regressions. + +from __future__ import annotations + +import json +from pathlib import Path +import pytest + +from pydantic import ValidationError + +from src.services.dashboard_testing.scenario.models import DashboardTestScenario +from src.services.dashboard_testing.scenario.resolver import ResolveChange, resolve_scenario + +_FIXTURES = Path(__file__).resolve().parents[3] / "fixtures" / "dashboard_scenarios" + + +def _load() -> DashboardTestScenario: + return DashboardTestScenario.model_validate(json.loads((_FIXTURES / "scenario_valid.json").read_text(encoding="utf-8"))) + + +def test_parameter_resolution_creates_new_revision() -> None: + sc = _load() + resolved = resolve_scenario(sc, [ResolveChange(kind="parameter", target="test_date", value="2026-08-01")]) + assert resolved.revision_hash != sc.revision_hash + assert resolved.parent_revision_hash == sc.revision_hash + param = next(p for p in resolved.parameters if p.name == "test_date") + assert param.value == "2026-08-01" + assert param.status == "resolved" + + +def test_unrelated_steps_unchanged() -> None: + sc = _load() + before = [(s.id, s.automation_status) for s in sc.steps] + resolved = resolve_scenario(sc, [ResolveChange(kind="parameter", target="counterparty", value="Beta LLC")]) + after = [(s.id, s.automation_status) for s in resolved.steps] + assert before == after + + +def test_selector_hint_resolution() -> None: + sc = _load() + resolved = resolve_scenario(sc, [ResolveChange(kind="selector", target="phase-2-B01-apply_filters", value="#filter-input")]) + assert resolved.revision_hash != sc.revision_hash + assert resolved.parent_revision_hash == sc.revision_hash + + +def test_stale_base_revision_rejected() -> None: + sc = _load() + with pytest.raises(ValueError, match="stale"): + resolve_scenario(sc, [ResolveChange(kind="parameter", target="test_date", value="2026-08-01")], base_revision_hash="0" * 64) + + +def test_invalid_parameter_type_rejected() -> None: + sc = _load() + with pytest.raises(ValidationError): + resolve_scenario(sc, [ResolveChange(kind="parameter", target="test_date", value=12345)]) + + +def test_unknown_target_rejected() -> None: + sc = _load() + with pytest.raises(ValueError, match=r"unknown"): + resolve_scenario(sc, [ResolveChange(kind="parameter", target="nonexistent_param", value="x")]) diff --git a/specs/038-dashboard-scenario-model/tasks.md b/specs/038-dashboard-scenario-model/tasks.md index 3efede709..21456fefd 100644 --- a/specs/038-dashboard-scenario-model/tasks.md +++ b/specs/038-dashboard-scenario-model/tasks.md @@ -62,15 +62,15 @@ ## Phase 5 — US4 Parameters and Human Checkpoints -- [ ] T022 [US4] Write failing typed resolution/stale revision tests in `backend/tests/services/dashboard_testing/scenario/test_resolver.py` -- [ ] T023 [US4] Implement `backend/src/services/dashboard_testing/scenario/resolver.py` for parameter, selector, manual conversion, and remove-step operations +- [x] T022 [US4] Write failing typed resolution/stale revision tests in `backend/tests/services/dashboard_testing/scenario/test_resolver.py` (6 passed) +- [x] T023 [US4] Implement `backend/src/services/dashboard_testing/scenario/resolver.py` for parameter, selector, manual conversion, and remove-step operations @PRE: base revision hash matches; changes target declared unresolved items @POST: unrelated step ids/order unchanged; new parent/revision hashes link revisions @TEST_EDGE: stale_base_revision→409, invalid_parameter_type→422, unrelated_graph_change→invariant failure -- [ ] T024 [US4] Enforce immutable revisions and unchanged unrelated step ids/order -- [ ] T025 [US4] Cover safe-environment/test-data requirements for mutating PDF cases +- [x] T024 [US4] Enforce immutable revisions and unchanged unrelated step ids/order +- [x] T025 [US4] Cover safe-environment/test-data requirements for mutating PDF cases -**Checkpoint**: Resolution produces linked immutable revisions; unrelated structure stable. +**Checkpoint**: Resolution produces linked immutable revisions; unrelated structure stable. ✅ (58 scenario tests green) ## Phase 6 — Safe Draft Pack