Files
phase-memory/tests/test_markitect_bridge.py

93 lines
3.7 KiB
Python

import json
from pathlib import Path
from phase_memory.bridge import (
MARKITECT_PACKAGE_REQUEST_SCHEMA,
MARKITECT_PACKAGE_RESPONSE_SCHEMA,
LocalMarkitectValidator,
OptionalMarkitectValidator,
package_response_envelope,
)
from phase_memory.contracts import graph_from_markitect
from phase_memory.runtime import PhaseMemoryRuntime
FIXTURES = Path(__file__).parent / "fixtures"
def _load(name: str):
return json.loads((FIXTURES / name).read_text(encoding="utf-8"))
def test_local_markitect_validator_reports_contract_diagnostics() -> None:
validator = LocalMarkitectValidator()
invalid_profile = validator.validate_profile(_load("markitect-invalid-profile.json"))
invalid_graph = validator.validate_graph(_load("markitect-invalid-graph.json"))
assert not invalid_profile.valid
assert [diagnostic.code for diagnostic in invalid_profile.diagnostics] == ["unsupported_profile_schema", "missing_profile_id"]
assert not invalid_graph.valid
assert [diagnostic.code for diagnostic in invalid_graph.diagnostics] == ["missing_edge_target"]
def test_optional_validator_falls_back_to_local_boundary() -> None:
validator = OptionalMarkitectValidator()
result = validator.validate_selection({"schema_version": "markitect.memory.selection.v1", "id": "selection.a"})
assert result.valid
assert result.subject_kind == "memory_selection"
def test_activation_preserves_metadata_for_package_request() -> None:
graph_data = _load("memory-graph.json")
graph_data["nodes"][0]["confidence"] = 0.95
graph_data["nodes"][0]["policy"] = {"labels": ["project-local"], "trust_zone": "local"}
graph_data["nodes"][0]["provenance"] = [{"source": "architecture"}]
graph = graph_from_markitect(graph_data).value
runtime = PhaseMemoryRuntime()
envelope = runtime.plan_activation(
graph.to_dict(),
max_items=1,
max_tokens=20,
profile_id="phase-memory-fixture-profile",
priority_node_ids=("decision.boundary",),
)
request = envelope["data"]["package_request"]
item = request["provenance"]["selected_items"]["decision.boundary"]
assert request["schema_version"] == MARKITECT_PACKAGE_REQUEST_SCHEMA
assert request["selection_id"] == envelope["data"]["activation_plan"]["plan_id"]
assert request["graph_id"] == "phase-memory-fixture-graph"
assert request["profile_id"] == "phase-memory-fixture-profile"
assert request["selected_nodes"] == ["decision.boundary"]
assert item["source_spans"] == [{"path": "docs/architecture.md", "line_start": 1}]
assert item["provenance"] == [{"source": "architecture"}]
assert item["confidence"] == 0.95
assert item["policy"]["labels"] == ["project-local"]
assert item["reason_selected"] == "priority"
def test_package_response_envelope_keeps_markitect_response_opaque() -> None:
response = _load("markitect-package-response.json")
envelope = package_response_envelope(response, request_id="request.a")
assert envelope["schema_version"] == MARKITECT_PACKAGE_RESPONSE_SCHEMA
assert envelope["request_id"] == "request.a"
assert envelope["package_ref"] == "package:activation-fixture"
assert envelope["response"] == response
def test_runtime_compile_package_uses_bridge_response_envelope() -> None:
runtime = PhaseMemoryRuntime()
selection = {"schema_version": "markitect.memory.selection.v1", "id": "selection.a", "nodes": ["node.a"], "events": []}
envelope = runtime.compile_package(selection)
assert envelope["data"]["package_request"]["schema_version"] == MARKITECT_PACKAGE_REQUEST_SCHEMA
assert envelope["data"]["package_response"]["schema_version"] == MARKITECT_PACKAGE_RESPONSE_SCHEMA
assert envelope["data"]["package_response"]["package_ref"] == "package:selection.a"