Implement live-shaped readiness workplan

src/phase_memory/evaluation.py

@@ -0,0 +1,161 @@
+"""Evaluation threshold reports for deterministic scenario fixtures."""
+
+from __future__ import annotations
+
+from datetime import datetime, timezone
+from typing import Any
+
+from .adapters import InMemorySemanticIndex
+from .contracts import graph_from_markitect
+from .models import Diagnostic, MemoryPath
+from .retrieval import activation_quality_report, select_event_path
+from .runtime import PhaseMemoryRuntime
+
+EVALUATION_REPORT_SCHEMA = "phase_memory.evaluation.threshold_report.v1"
+
+DEFAULT_THRESHOLDS = {
+    "policy_denial_count": 1,
+    "lifecycle_action_count": 3,
+    "path_event_count": 1,
+    "semantic_hit_count": 1,
+    "budget_omission_count": 1,
+    "source_span_coverage": 1.0,
+    "explanation_coverage": 1.0,
+}
+
+
+def evaluation_threshold_report(data: dict[str, Any], *, thresholds: dict[str, float] | None = None) -> dict[str, Any]:
+    thresholds = {**DEFAULT_THRESHOLDS, **dict(thresholds or {})}
+    scenarios = list(data.get("scenarios") or ())
+    metrics = {
+        "scenario_count": len(scenarios),
+        "policy_denial_count": 0,
+        "lifecycle_action_count": 0,
+        "path_event_count": 0,
+        "semantic_hit_count": 0,
+        "budget_omission_count": 0,
+        "source_span_coverage": 0.0,
+        "explanation_coverage": 0.0,
+    }
+    scenario_reports: list[dict[str, Any]] = []
+
+    for scenario in scenarios:
+        scenario_id = str(scenario.get("id") or "")
+        if scenario_id == "policy-denied-activation":
+            report = _policy_scenario(scenario)
+        elif scenario_id == "profile-lifecycle-rules":
+            report = _lifecycle_scenario(scenario)
+        elif scenario_id == "budget-path-and-semantic-hints":
+            report = _budget_scenario(scenario)
+        else:
+            report = {"id": scenario_id, "metrics": {}, "diagnostics": [{"severity": "warn", "code": "unknown_scenario", "message": "Scenario is not recognized by this report."}]}
+        scenario_reports.append(report)
+        for key, value in report.get("metrics", {}).items():
+            if key in metrics and isinstance(value, (int, float)):
+                metrics[key] += value
+
+    diagnostics = _threshold_diagnostics(metrics, thresholds)
+    return {
+        "schema_version": EVALUATION_REPORT_SCHEMA,
+        "valid": not diagnostics,
+        "metrics": metrics,
+        "thresholds": thresholds,
+        "scenarios": scenario_reports,
+        "diagnostics": [diagnostic.to_dict() for diagnostic in diagnostics],
+    }
+
+
+def _policy_scenario(scenario: dict[str, Any]) -> dict[str, Any]:
+    runtime = PhaseMemoryRuntime()
+    response = runtime.plan_activation(
+        scenario["graph"],
+        max_items=int(scenario["profile"].get("activation", {}).get("max_items") or 4),
+        max_tokens=int(scenario["profile"].get("activation", {}).get("max_tokens") or 60),
+        profile_id=scenario["profile"]["id"],
+        policy_context={"denied_labels": ["restricted"], "secrets_allowed": False, "trust_zone": "local"},
+    )
+    denials = response["data"]["policy_denials"]
+    return {
+        "id": scenario["id"],
+        "metrics": {"policy_denial_count": len(denials)},
+        "diagnostics": response["diagnostics"],
+    }
+
+
+def _lifecycle_scenario(scenario: dict[str, Any]) -> dict[str, Any]:
+    runtime = PhaseMemoryRuntime()
+    response = runtime.plan_lifecycle_with_profile(
+        scenario["profile"],
+        scenario["graph"],
+        refresh_digests={"life.decision": "decision-new"},
+        now=datetime(2026, 5, 18, tzinfo=timezone.utc),
+    )
+    return {
+        "id": scenario["id"],
+        "metrics": {"lifecycle_action_count": len(response["data"]["dry_run_actions"])},
+        "diagnostics": response["diagnostics"],
+    }
+
+
+def _budget_scenario(scenario: dict[str, Any]) -> dict[str, Any]:
+    runtime = PhaseMemoryRuntime()
+    graph = graph_from_markitect(scenario["graph"]).value
+    activation = runtime.plan_activation(
+        scenario["graph"],
+        max_items=int(scenario["profile"]["activation"]["max_items"]),
+        max_tokens=int(scenario["profile"]["activation"]["max_tokens"]),
+        profile_id=scenario["profile"]["id"],
+        priority_node_ids=tuple(scenario["expect"]["selected_node_ids"]),
+    )
+    plan = activation["data"]["activation_plan"]
+    quality = activation_quality_report(_activation_plan_from_response(activation), expected_node_ids=tuple(scenario["expect"]["selected_node_ids"]))
+    path_events = select_event_path(graph.events, MemoryPath.from_mapping(scenario["path"]), max_events=2)
+    index = InMemorySemanticIndex()
+    index.upsert_nodes(list(graph.nodes))
+    semantic_hits = index.query(graph_id=graph.graph_id, query="semantic restart", limit=2)
+    return {
+        "id": scenario["id"],
+        "metrics": {
+            "path_event_count": len(path_events),
+            "semantic_hit_count": 1 if semantic_hits and semantic_hits[0]["id"] == scenario["expect"]["semantic_top_id"] else 0,
+            "budget_omission_count": len(plan["omitted"]),
+            "source_span_coverage": quality["source_span_coverage"],
+            "explanation_coverage": quality["explanation_coverage"],
+        },
+        "diagnostics": activation["diagnostics"],
+    }
+
+
+def _activation_plan_from_response(response: dict[str, Any]):
+    from .models import ActivationPlan
+
+    data = response["data"]["activation_plan"]
+    return ActivationPlan(
+        plan_id=data["plan_id"],
+        graph_id=data["graph_id"],
+        selected_node_ids=tuple(data["selected_node_ids"]),
+        selected_event_ids=tuple(data["selected_event_ids"]),
+        omitted=tuple(data["omitted"]),
+        token_estimate=data["token_estimate"],
+        max_items=data["max_items"],
+        max_tokens=data["max_tokens"],
+        selection=response["data"]["package_request"]["selection"],
+        diagnostics=(),
+    )
+
+
+def _threshold_diagnostics(metrics: dict[str, Any], thresholds: dict[str, float]) -> tuple[Diagnostic, ...]:
+    diagnostics: list[Diagnostic] = []
+    for key, threshold in sorted(thresholds.items()):
+        actual = float(metrics.get(key) or 0)
+        if actual < float(threshold):
+            diagnostics.append(
+                Diagnostic(
+                    "error",
+                    "evaluation_threshold_failed",
+                    "Evaluation metric did not meet its threshold.",
+                    key,
+                    {"actual": actual, "threshold": threshold},
+                )
+            )
+    return tuple(diagnostics)