wehub-resource-sync
212 lines
7.7 KiB
Python
212 lines
7.7 KiB
Python
"""Equivalence harness — policy-pipeline self-comparison.
|
|
|
|
Both control and candidate are :func:`opensquilla.tools.dispatch.build_tool_handler`
|
|
(the policy pipeline). This is a determinism/infrastructure smoke-test: two
|
|
independent factory calls must produce byte-for-byte identical results for every
|
|
corpus case. Any shared mutable state, contextvar leak, or non-deterministic
|
|
ordering will surface here.
|
|
|
|
The legacy dispatch module was removed. The permanent behavioural contract
|
|
lives in:
|
|
- test_dispatch_corpus_snapshots.py (golden envelopes)
|
|
- test_dispatch_properties.py (idempotence, ordering invariants)
|
|
- test_dispatch_legacy_coverage.py (line coverage gate, now targets dispatch.py)
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import copy
|
|
import json
|
|
from typing import Any
|
|
|
|
import pytest
|
|
import structlog.testing
|
|
from laboratory import Experiment
|
|
from test_tools.dispatch_corpus import ALL_CASES, CorpusCase # noqa: E402
|
|
|
|
from opensquilla.tool_boundary import ToolResult
|
|
from opensquilla.tools.dispatch import build_tool_handler as _build_candidate
|
|
|
|
# Both sides use the policy-pipeline factory.
|
|
from opensquilla.tools.dispatch import build_tool_handler as _build_control
|
|
from opensquilla.tools.types import ToolContext, current_tool_context
|
|
|
|
# -----------------------------------------------------------------------
|
|
# Envelope comparator
|
|
# -----------------------------------------------------------------------
|
|
|
|
def _parse_content(content: Any) -> dict[str, Any] | str:
|
|
if isinstance(content, str):
|
|
try:
|
|
return json.loads(content)
|
|
except (TypeError, ValueError):
|
|
return content
|
|
return content
|
|
|
|
|
|
def _envelope_eq(a: ToolResult, b: ToolResult) -> bool:
|
|
"""Compare two ToolResult envelopes for dispatch-contract equivalence."""
|
|
from opensquilla.execution_status import normalize_execution_status
|
|
|
|
if type(a) is not type(b):
|
|
return False
|
|
if a.tool_use_id != b.tool_use_id:
|
|
return False
|
|
if a.tool_name != b.tool_name:
|
|
return False
|
|
if a.is_error != b.is_error:
|
|
return False
|
|
a_es = (
|
|
normalize_execution_status(a.execution_status)
|
|
if a.execution_status is not None
|
|
else None
|
|
)
|
|
b_es = (
|
|
normalize_execution_status(b.execution_status)
|
|
if b.execution_status is not None
|
|
else None
|
|
)
|
|
if a_es != b_es:
|
|
return False
|
|
if _parse_content(a.content) != _parse_content(b.content):
|
|
return False
|
|
if a.artifacts != b.artifacts:
|
|
return False
|
|
return True
|
|
|
|
|
|
# -----------------------------------------------------------------------
|
|
# Log record normaliser
|
|
# -----------------------------------------------------------------------
|
|
|
|
def _normalise_log_records(records: list[dict[str, Any]]) -> set[tuple[str, str, str]]:
|
|
result: set[tuple[str, str, str]] = set()
|
|
for rec in records:
|
|
result.add((rec.get("_logger", ""), rec.get("log_level", ""), rec.get("event", "")))
|
|
return result
|
|
|
|
|
|
# -----------------------------------------------------------------------
|
|
# Side-effect capture helper
|
|
# -----------------------------------------------------------------------
|
|
|
|
class _DispatchRun:
|
|
def __init__(self) -> None:
|
|
self.result: ToolResult | None = None
|
|
self.artifact_snapshot: list[dict[str, Any]] = []
|
|
self.log_records: set[tuple[str, str, str]] = set()
|
|
|
|
|
|
async def _run_one_side(
|
|
build_fn: Any,
|
|
case: CorpusCase,
|
|
ctx: ToolContext | None,
|
|
registry: Any,
|
|
) -> _DispatchRun:
|
|
run = _DispatchRun()
|
|
handler = build_fn(
|
|
registry,
|
|
ctx,
|
|
known_skill_names=set(case.known_skill_names) if case.known_skill_names else None,
|
|
)
|
|
token = current_tool_context.set(None)
|
|
try:
|
|
with structlog.testing.capture_logs() as logs:
|
|
run.result = await handler(case.tool_call)
|
|
run.log_records = _normalise_log_records(logs)
|
|
run.artifact_snapshot = list(ctx.published_artifacts) if ctx is not None else []
|
|
finally:
|
|
current_tool_context.reset(token)
|
|
return run
|
|
|
|
|
|
# -----------------------------------------------------------------------
|
|
# Experiment subclass
|
|
# -----------------------------------------------------------------------
|
|
|
|
class _MismatchRecorder:
|
|
def __init__(self) -> None:
|
|
self.mismatches: list[str] = []
|
|
|
|
|
|
class _DispatchExperiment(Experiment):
|
|
def __init__(self, name: str, recorder: _MismatchRecorder) -> None:
|
|
super().__init__(name=name, raise_on_mismatch=False)
|
|
self._recorder = recorder
|
|
|
|
def compare(self, control: Any, candidate: Any) -> bool: # type: ignore[override]
|
|
ctrl_run: _DispatchRun = control.value
|
|
cand_run: _DispatchRun = candidate.value
|
|
if ctrl_run.result is None or cand_run.result is None:
|
|
self._recorder.mismatches.append("One side returned None result")
|
|
return False
|
|
if not _envelope_eq(ctrl_run.result, cand_run.result):
|
|
self._recorder.mismatches.append(
|
|
f"Envelope mismatch: is_error={ctrl_run.result.is_error} vs "
|
|
f"{cand_run.result.is_error}, "
|
|
f"content={_parse_content(ctrl_run.result.content)!r} vs "
|
|
f"{_parse_content(cand_run.result.content)!r}"
|
|
)
|
|
return False
|
|
if ctrl_run.artifact_snapshot != cand_run.artifact_snapshot:
|
|
self._recorder.mismatches.append(
|
|
f"Artifact mismatch: {ctrl_run.artifact_snapshot!r} vs "
|
|
f"{cand_run.artifact_snapshot!r}"
|
|
)
|
|
return False
|
|
if ctrl_run.log_records != cand_run.log_records:
|
|
self._recorder.mismatches.append(
|
|
f"Log record mismatch: {ctrl_run.log_records!r} vs "
|
|
f"{cand_run.log_records!r}"
|
|
)
|
|
return False
|
|
return True
|
|
|
|
def publish(self, result: Any) -> None: # type: ignore[override]
|
|
pass
|
|
|
|
|
|
# -----------------------------------------------------------------------
|
|
# Async harness runner
|
|
# -----------------------------------------------------------------------
|
|
|
|
async def _run_equivalence(case: CorpusCase) -> _MismatchRecorder:
|
|
recorder = _MismatchRecorder()
|
|
ctrl_ctx = copy.deepcopy(case.ctx_factory())
|
|
cand_ctx = copy.deepcopy(case.ctx_factory())
|
|
ctrl_registry = case.registry_factory()
|
|
cand_registry = case.registry_factory()
|
|
if case.setup is not None:
|
|
case.setup()
|
|
try:
|
|
ctrl_run = await _run_one_side(_build_control, case, ctrl_ctx, ctrl_registry)
|
|
cand_run = await _run_one_side(_build_candidate, case, cand_ctx, cand_registry)
|
|
finally:
|
|
if case.teardown is not None:
|
|
case.teardown()
|
|
experiment = _DispatchExperiment(name=case.name, recorder=recorder)
|
|
experiment.control(lambda: ctrl_run)
|
|
experiment.candidate(lambda: cand_run)
|
|
experiment.conduct(randomize=False)
|
|
ctx_val = current_tool_context.get()
|
|
if ctx_val is not None and case.contextvar_must_be_none_after:
|
|
recorder.mismatches.append(
|
|
f"Contextvar leak after dispatch: current_tool_context.get() = {ctx_val!r}"
|
|
)
|
|
return recorder
|
|
|
|
|
|
# -----------------------------------------------------------------------
|
|
# Parametrised test
|
|
# -----------------------------------------------------------------------
|
|
|
|
@pytest.mark.parametrize("case", ALL_CASES, ids=[c.name for c in ALL_CASES])
|
|
@pytest.mark.asyncio
|
|
async def test_dispatch_equivalence_corpus(case: CorpusCase) -> None:
|
|
"""Self-comparison: two independent factory calls must produce identical results."""
|
|
recorder = await _run_equivalence(case)
|
|
assert not recorder.mismatches, (
|
|
f"Equivalence failures for case '{case.name}':\n"
|
|
+ "\n".join(f" - {m}" for m in recorder.mismatches)
|
|
)
|