phase-memory/tests/test_service_readiness.py

import json
from pathlib import Path

from phase_memory.models import LifecycleState, MemoryNode
from phase_memory.service import (
    HEALTH_REPORT_SCHEMA,
    KONTEXTUAL_DELEGATION_SCHEMA,
    SERVICE_CONTRACT_SCHEMA,
    LocalServiceRunner,
    RuntimeConfig,
    assert_audit_sink_conformance,
    assert_context_compiler_conformance,
    assert_event_log_conformance,
    assert_graph_store_conformance,
    assert_policy_gateway_conformance,
    assert_runtime_registry_conformance,
    assert_semantic_index_conformance,
    default_conformance_adapters,
    health_report,
    kontextual_delegation_envelope,
    resolve_runtime_adapters,
    runtime_from_config,
    service_contracts,
)

FIXTURES = Path(__file__).parent / "fixtures"


def _load(name: str):
    return json.loads((FIXTURES / name).read_text(encoding="utf-8"))


def test_service_contracts_list_runtime_operations() -> None:
    contracts = service_contracts()

    assert contracts["schema_version"] == SERVICE_CONTRACT_SCHEMA
    assert "profile.plan" in contracts["operations"]
    assert "health.check" in contracts["operations"]


def test_runtime_config_validation_and_health_report() -> None:
    runner = LocalServiceRunner()
    runner.runtime.graph_store.save_node(MemoryNode("node.stale", "episode", lifecycle=LifecycleState.STALE))
    report = health_report(runner.runtime, config=RuntimeConfig.local_default())

    assert report["schema_version"] == HEALTH_REPORT_SCHEMA
    assert report["ok"] is True
    assert report["store"]["stale_memory_count"] == 1
    assert report["adapters"]["graph_store"] == "InMemoryMemoryGraphStore"


def test_service_runner_handles_health() -> None:
    runner = LocalServiceRunner()

    response = runner.handle("health.check")

    assert response["schema_version"] == HEALTH_REPORT_SCHEMA
    assert response["ok"] is True


def test_service_runner_handles_profile_driven_lifecycle_plan() -> None:
    runner = LocalServiceRunner()

    response = runner.handle(
        "graph.lifecycle.plan",
        {
            "profile": _load("memory-profile.json"),
            "graph": _load("memory-graph.json"),
            "parameters": {"refresh_digests": {"event.restart": "new"}},
        },
    )

    actions = {(action["target_id"], action["action"]) for action in response["data"]["dry_run_actions"]}
    assert response["operation"] == "graph.lifecycle.plan"
    assert response["data"]["profile_id"] == "phase-memory-fixture-profile"
    assert ("event.restart", "refresh") in actions


def test_profile_driven_runtime_config_resolves_file_backed_adapters(tmp_path) -> None:
    config = RuntimeConfig.from_profile(
        {
            "schema_version": "markitect.memory.profile.v1",
            "id": "profile.config",
            "stores": {
                "graph_store": "file",
                "event_log": "jsonl",
            },
            "activation": {"semantic_index": "memory"},
            "policy": {"mode": "allow-all", "trust_zones": ["local", "team"]},
            "observability": {"audit_sink": "jsonl", "runtime_registry": "memory"},
            "metadata": {"runtime": {"local_store_path": str(tmp_path / "memory-store")}},
        }
    )

    bundle = resolve_runtime_adapters(config)
    runtime = runtime_from_config(config)
    runtime.graph_store.save_node(MemoryNode("node.config", "decision", "Config-driven node"))

    assert config.adapter_mode("graph_store") == "file"
    assert config.adapter_mode("event_log") == "jsonl"
    assert config.trust_zone_labels == ("local", "team")
    assert bundle.to_dict()["graph_store"] == "FileBackedMemoryGraphStore"
    assert bundle.to_dict()["event_log"] == "JsonlMemoryEventLog"
    assert bundle.to_dict()["semantic_index"] == "InMemorySemanticIndex"
    assert (tmp_path / "memory-store" / "nodes" / "node.config.json").exists()


def test_runtime_config_from_fixture_profile_understands_local_aliases() -> None:
    config = RuntimeConfig.from_profile(_load("memory-profile.json"), local_store_path=".phase-memory-test")

    assert config.adapter_mode("graph_store") == "file"
    assert config.adapter_mode("event_log") == "jsonl"
    assert config.adapter_mode("package_compiler") == "noop"
    assert config.trust_zone_labels == ("project-local",)


def test_missing_external_adapter_blocks_runtime_resolution() -> None:
    registry = RuntimeConfig.local_default().adapter_registry.copy()
    registry["policy_gateway"] = "external"
    config = RuntimeConfig(adapter_registry=registry, policy_mode="external")

    bundle = resolve_runtime_adapters(config)

    assert "external_adapter_declared" in [diagnostic.code for diagnostic in bundle.diagnostics]
    assert "missing_external_adapter" in [diagnostic.code for diagnostic in bundle.diagnostics]
    try:
        runtime_from_config(config)
    except ValueError as exc:
        assert "missing_external_adapter" in str(exc)
    else:
        raise AssertionError("runtime_from_config should require supplied external adapters")


def test_default_adapter_conformance_helpers() -> None:
    adapters = default_conformance_adapters()

    assert_graph_store_conformance(adapters["graph_store"])
    assert_event_log_conformance(adapters["event_log"])
    assert_context_compiler_conformance(adapters["context_compiler"])
    assert_policy_gateway_conformance(adapters["policy_gateway"])
    assert_audit_sink_conformance(adapters["audit_sink"])
    assert_semantic_index_conformance(adapters["semantic_index"])
    assert_runtime_registry_conformance(adapters["runtime_registry"])


def test_kontextual_delegation_envelope_is_explicit() -> None:
    envelope = kontextual_delegation_envelope(
        operation="persist_graph",
        graph_id="graph.a",
        profile_id="profile.a",
        policy_decision={"allowed": True},
        audit_ref="audit.a",
    )

    assert envelope["schema_version"] == KONTEXTUAL_DELEGATION_SCHEMA
    assert "phase_policy" in envelope["phase_memory_owns"]
    assert "durable_records" in envelope["kontextual_owns"]
    assert envelope["imports"]["avoid_circular_imports"] is True