coulomb/phase-memory
4
0
Fork 0
Code Issues Pull requests Projects Releases Packages Wiki Activity Actions

Implement PMEM-WP-0015 credentialed live pilot with ops-warden routing.

Browse source 
Add credential routing advisories via warden route/access, live pilot evidence
helpers, managed deployment pilot probes, evaluation trend regression gates,
and expanded troubleshooting. Update operator runbook and maturity scorecard.
This commit is contained in:
tegwick 2026-07-02 23:24:35 +02:00
parent bff90ec1ed
commit 29f893b905
15 changed files with 913 additions and 38 deletions
Download patch file Download diff file Expand all files Collapse all files

81
src/phase_memory/evaluation.py
View file

@ -17,6 +17,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"
EVALUATION_TREND_REGRESSION_GATE_SCHEMA = "phase_memory.evaluation.trend_regression_gate.v1"
DEFAULT_THRESHOLDS = {
"policy_denial_count": 1,
@ -165,6 +166,86 @@ def load_evaluation_trend_history(path: str | Path) -> dict[str, Any]:
return evaluation_trend_history((data,))
def evaluation_trend_regression_gate(
history: dict[str, Any],
*,
min_artifacts: int = 1,
) -> dict[str, Any]:
artifacts = list(history.get("artifacts") or ())
diagnostics: list[dict[str, Any]] = []
if history.get("schema_version") != EVALUATION_TREND_HISTORY_SCHEMA:
diagnostics.append(
Diagnostic(
"error",
"evaluation_trend_history_invalid",
"Regression gate requires a valid evaluation trend history artifact.",
"schema_version",
{"expected": EVALUATION_TREND_HISTORY_SCHEMA},
).to_dict()
)
if len(artifacts) < min_artifacts:
diagnostics.append(
Diagnostic(
"warn",
"evaluation_trend_history_insufficient",
"Regression gate needs at least one persisted trend artifact.",
"count",
{"actual": len(artifacts), "minimum": min_artifacts},
).to_dict()
)
latest = artifacts[-1] if artifacts else {}
previous = artifacts[-2] if len(artifacts) > 1 else {}
latest_metrics = dict(latest.get("metrics") or {})
previous_metrics = dict(previous.get("metrics") or {})
regressions = {
key: round(float(latest_metrics.get(key) or 0) - float(previous_metrics.get(key) or 0), 4)
for key in sorted(set(latest_metrics) & set(previous_metrics))
if float(latest_metrics.get(key) or 0) < float(previous_metrics.get(key) or 0)
}
for key, delta in regressions.items():
diagnostics.append(
Diagnostic(
"warn",
"evaluation_metric_regressed",
"Evaluation metric declined from the previous trend artifact.",
key,
{
"delta": delta,
"current": latest_metrics.get(key),
"previous": previous_metrics.get(key),
},
).to_dict()
)
for diagnostic in latest.get("diagnostics", ()):
if isinstance(diagnostic, dict) and diagnostic.get("code") == "evaluation_metric_regressed":
diagnostics.append(dict(diagnostic))
threshold_failures = [
dict(item)
for item in (latest.get("report") or {}).get("diagnostics", ())
if isinstance(item, dict) and item.get("code") == "evaluation_threshold_failed"
]
for failure in threshold_failures:
diagnostics.append(failure)
return {
"schema_version": EVALUATION_TREND_REGRESSION_GATE_SCHEMA,
"id": f"evaluation-trend-regression-gate:{stable_digest([history.get('id', ''), latest.get('id', ''), regressions])}",
"valid": not any(item.get("severity") == "error" for item in diagnostics)
and not threshold_failures
and not regressions,
"artifact_count": len(artifacts),
"latest_artifact_id": latest.get("id", ""),
"previous_artifact_id": previous.get("id", ""),
"metric_regressions": regressions,
"threshold_failures": threshold_failures,
"operator_guidance": {
"compare": "Diff the latest evaluation-trend-history.json artifact metrics against the previous run id.",
"gate": "Block promotion when metric_regressions or threshold_failures are non-empty.",
"history_path": "reports/evaluation-trend-history.json",
},
"diagnostics": diagnostics,
}
def write_evaluation_trend_history(path: str | Path, artifact: dict[str, Any]) -> dict[str, Any]:
path = Path(path)
existing = load_evaluation_trend_history(path)