276 lines
11 KiB
Python
276 lines
11 KiB
Python
"""Evaluation threshold reports for deterministic scenario fixtures."""
|
|
|
|
from __future__ import annotations
|
|
|
|
import json
|
|
from datetime import datetime, timezone
|
|
from pathlib import Path
|
|
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
|
|
from .utils import stable_digest, utc_now_iso
|
|
|
|
EVALUATION_REPORT_SCHEMA = "phase_memory.evaluation.threshold_report.v1"
|
|
EVALUATION_TREND_SCHEMA = "phase_memory.evaluation.trend_artifact.v1"
|
|
EVALUATION_TREND_HISTORY_SCHEMA = "phase_memory.evaluation.trend_history.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 evaluation_trend_artifact(
|
|
report: dict[str, Any],
|
|
*,
|
|
previous_report: dict[str, Any] | None = None,
|
|
run_metadata: dict[str, Any] | None = None,
|
|
) -> dict[str, Any]:
|
|
run_metadata = {
|
|
"created_at": utc_now_iso(),
|
|
**dict(run_metadata or {}),
|
|
}
|
|
metrics = dict(report.get("metrics") or {})
|
|
thresholds = dict(report.get("thresholds") or {})
|
|
previous_metrics = dict((previous_report or {}).get("metrics") or {})
|
|
threshold_deltas = {
|
|
key: round(float(metrics.get(key) or 0) - float(threshold), 4)
|
|
for key, threshold in sorted(thresholds.items())
|
|
}
|
|
metric_deltas = {
|
|
key: round(float(value or 0) - float(previous_metrics.get(key) or 0), 4)
|
|
for key, value in sorted(metrics.items())
|
|
if key in previous_metrics
|
|
}
|
|
diagnostics = [dict(item) for item in report.get("diagnostics", ())]
|
|
for key, delta in metric_deltas.items():
|
|
if delta < 0:
|
|
diagnostics.append(
|
|
Diagnostic(
|
|
"warn",
|
|
"evaluation_metric_regressed",
|
|
"Evaluation metric declined from the previous report.",
|
|
key,
|
|
{"delta": delta, "current": metrics.get(key), "previous": previous_metrics.get(key)},
|
|
).to_dict()
|
|
)
|
|
artifact_id = f"evaluation-trend:{stable_digest([run_metadata, metrics, thresholds, previous_metrics])}"
|
|
return {
|
|
"schema_version": EVALUATION_TREND_SCHEMA,
|
|
"id": artifact_id,
|
|
"valid": not any(item.get("severity") == "error" for item in diagnostics),
|
|
"run": run_metadata,
|
|
"metrics": metrics,
|
|
"thresholds": thresholds,
|
|
"threshold_deltas": threshold_deltas,
|
|
"metric_deltas": metric_deltas,
|
|
"report": report,
|
|
"previous_report_id": (previous_report or {}).get("id", ""),
|
|
"diagnostics": diagnostics,
|
|
}
|
|
|
|
|
|
def evaluation_trend_history(artifacts: list[dict[str, Any]] | tuple[dict[str, Any], ...]) -> dict[str, Any]:
|
|
ordered = sorted(
|
|
(dict(artifact) for artifact in artifacts),
|
|
key=lambda artifact: (
|
|
str((artifact.get("run") or {}).get("created_at") or ""),
|
|
str((artifact.get("run") or {}).get("run_id") or ""),
|
|
str(artifact.get("id") or ""),
|
|
),
|
|
)
|
|
metric_keys = sorted({str(key) for artifact in ordered for key in (artifact.get("metrics") or {})})
|
|
diagnostics = [
|
|
Diagnostic(
|
|
"error",
|
|
"evaluation_trend_history_invalid_artifact",
|
|
"Trend history can only contain evaluation trend artifacts.",
|
|
f"artifacts.{index}.schema_version",
|
|
{"artifact_id": artifact.get("id", "")},
|
|
).to_dict()
|
|
for index, artifact in enumerate(ordered)
|
|
if artifact.get("schema_version") != EVALUATION_TREND_SCHEMA
|
|
]
|
|
return {
|
|
"schema_version": EVALUATION_TREND_HISTORY_SCHEMA,
|
|
"id": f"evaluation-trend-history:{stable_digest([artifact.get('id', '') for artifact in ordered])}",
|
|
"valid": not diagnostics,
|
|
"count": len(ordered),
|
|
"metric_keys": metric_keys,
|
|
"latest_artifact_id": ordered[-1].get("id", "") if ordered else "",
|
|
"artifacts": ordered,
|
|
"diagnostics": diagnostics,
|
|
}
|
|
|
|
|
|
def load_evaluation_trend_history(path: str | Path) -> dict[str, Any]:
|
|
path = Path(path)
|
|
if not path.exists():
|
|
return evaluation_trend_history(())
|
|
data = json.loads(path.read_text(encoding="utf-8"))
|
|
if data.get("schema_version") == EVALUATION_TREND_HISTORY_SCHEMA:
|
|
return data
|
|
if data.get("schema_version") == EVALUATION_TREND_SCHEMA:
|
|
return evaluation_trend_history((data,))
|
|
return evaluation_trend_history((data,))
|
|
|
|
|
|
def write_evaluation_trend_history(path: str | Path, artifact: dict[str, Any]) -> dict[str, Any]:
|
|
path = Path(path)
|
|
existing = load_evaluation_trend_history(path)
|
|
artifacts = list(existing.get("artifacts") or ())
|
|
artifact_id = str(artifact.get("id") or "")
|
|
if artifact_id and not any(str(item.get("id") or "") == artifact_id for item in artifacts):
|
|
artifacts.append(dict(artifact))
|
|
elif not artifact_id:
|
|
artifacts.append(dict(artifact))
|
|
history = evaluation_trend_history(tuple(artifacts))
|
|
path.parent.mkdir(parents=True, exist_ok=True)
|
|
path.write_text(json.dumps(history, indent=2, sort_keys=True) + "\n", encoding="utf-8")
|
|
return history
|
|
|
|
|
|
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)
|