generated from coulomb/repo-seed
106 lines
4.2 KiB
Python
106 lines
4.2 KiB
Python
import json
|
|
from pathlib import Path
|
|
|
|
from phase_memory.adapters import FileBackedMemoryGraphStore, JsonlMemoryEventLog
|
|
from phase_memory.lifecycle import plan_compaction
|
|
from phase_memory.models import MemoryNode, ReviewDecision
|
|
from phase_memory.policy import AUDIT_EVENT_SCHEMA, POLICY_OPERATION_POINTS, REDACTED, make_review_record
|
|
from phase_memory.runtime import PhaseMemoryRuntime
|
|
from phase_memory.utils import stable_digest
|
|
|
|
|
|
FIXTURES = Path(__file__).parent / "fixtures"
|
|
|
|
|
|
def _load(name: str):
|
|
return json.loads((FIXTURES / name).read_text(encoding="utf-8"))
|
|
|
|
|
|
def test_policy_operation_points_cover_runtime_boundaries() -> None:
|
|
assert "profile.import" in POLICY_OPERATION_POINTS
|
|
assert "graph.activation.plan" in POLICY_OPERATION_POINTS
|
|
assert "lifecycle.apply" in POLICY_OPERATION_POINTS
|
|
assert "graph.export" in POLICY_OPERATION_POINTS
|
|
|
|
|
|
def test_runtime_audit_events_use_stable_schema() -> None:
|
|
runtime = PhaseMemoryRuntime()
|
|
|
|
envelope = runtime.plan_profile(_load("memory-profile.json"), source_ref="profile")
|
|
event = envelope["audit_receipt"]["event"]
|
|
|
|
assert event["schema_version"] == AUDIT_EVENT_SCHEMA
|
|
assert event["operation"] == "profile.plan"
|
|
assert event["allowed"] is True
|
|
assert event["policy_decision"]["allowed"] is True
|
|
assert event["source"]["ref"] == "profile"
|
|
|
|
|
|
def test_review_record_approves_review_required_lifecycle_action(tmp_path) -> None:
|
|
store = FileBackedMemoryGraphStore(tmp_path)
|
|
runtime = PhaseMemoryRuntime(graph_store=store, event_log=JsonlMemoryEventLog(tmp_path / "events.jsonl"))
|
|
node = store.save_node(MemoryNode("node.a", "episode", "Trace text"))
|
|
action = plan_compaction([node])
|
|
review = make_review_record(
|
|
reviewed_action_id=f"action:{stable_digest(action.to_dict())}",
|
|
reviewer="reviewer.local",
|
|
approved=True,
|
|
reason="reviewed compacted summary",
|
|
obligations=("retain_sources",),
|
|
source_digests={"node.a": "digest"},
|
|
)
|
|
|
|
envelope = runtime.apply_lifecycle_actions([action], review_record=review)
|
|
|
|
assert review.decision == ReviewDecision.APPROVED
|
|
assert envelope["valid"] is True
|
|
assert envelope["data"]["review_record"]["id"] == review.review_id
|
|
assert store.get_node(action.target_id).metadata["approval_marker"] == review.review_id
|
|
|
|
|
|
def test_rejected_or_mismatched_review_record_denies_apply(tmp_path) -> None:
|
|
store = FileBackedMemoryGraphStore(tmp_path)
|
|
runtime = PhaseMemoryRuntime(graph_store=store, event_log=JsonlMemoryEventLog(tmp_path / "events.jsonl"))
|
|
node = store.save_node(MemoryNode("node.a", "episode", "Trace text"))
|
|
action = plan_compaction([node])
|
|
review = make_review_record(
|
|
reviewed_action_id="action:other",
|
|
reviewer="reviewer.local",
|
|
approved=True,
|
|
reason="wrong action",
|
|
)
|
|
|
|
envelope = runtime.apply_lifecycle_actions([action], review_record=review)
|
|
|
|
assert envelope["valid"] is False
|
|
assert envelope["data"]["denied"][0]["reason"] == "review_required"
|
|
|
|
|
|
def test_activation_policy_denies_and_redacts_nodes() -> None:
|
|
graph = _load("memory-graph.json")
|
|
graph["nodes"][0]["policy"] = {"labels": ["project-local"], "trust_zone": "local"}
|
|
graph["nodes"][1]["policy"] = {"labels": ["project-local"], "trust_zone": "local", "reauthorization": "daily"}
|
|
graph["nodes"][2]["policy"] = {"labels": ["project-local"], "secret": True}
|
|
graph["nodes"][3]["policy"] = {"labels": ["restricted"]}
|
|
runtime = PhaseMemoryRuntime()
|
|
|
|
envelope = runtime.plan_activation(
|
|
graph,
|
|
max_items=4,
|
|
max_tokens=80,
|
|
policy_context={
|
|
"required_labels": ["project-local"],
|
|
"denied_labels": ["restricted"],
|
|
"trust_zone": "local",
|
|
"secrets_allowed": False,
|
|
"approved_reauthorizations": [],
|
|
},
|
|
)
|
|
|
|
selected = envelope["data"]["activation_plan"]["selected_node_ids"]
|
|
denials = envelope["data"]["policy_denials"]
|
|
assert selected == ["decision.boundary"]
|
|
assert [item["id"] for item in denials] == ["event.restart", "artifact.profile", "risk.durable-write"]
|
|
assert all(item["text"] == REDACTED for item in denials)
|
|
assert "activation_policy_denied" in [diagnostic["code"] for diagnostic in envelope["diagnostics"]]
|