feat(038): Phase 5 — US4 immutable resolver

- T022-T025: ScenarioGraph.Resolver.Resolve — typed parameter/selector/manual/
  remove-step resolutions emit immutable revisions linked via parent_revision_hash;
  stale base revision rejected (409 semantics); unrelated step ids/order unchanged
- Selector hints recorded in step description for auditability
- 58 scenario tests pass; ruff clean; regions balanced
This commit is contained in:
2026-07-31 13:00:11 +03:00
parent 9224d1a9ca
commit 31e8524a32
3 changed files with 219 additions and 5 deletions

View File

@@ -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

View File

@@ -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")])

View File

@@ -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