Implement credentialed live hardening workplan

src/phase_memory/evaluation.py

@@ -2,7 +2,9 @@
 
 from __future__ import annotations
 
+import json
 from datetime import datetime, timezone
+from pathlib import Path
 from typing import Any
 
 from .adapters import InMemorySemanticIndex
@@ -14,6 +16,7 @@ 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,
@@ -115,6 +118,68 @@ def evaluation_trend_artifact(
         "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(