Implement PMEM-WP-0016 ops-warden cross-runtime memory contracts.

Add ops-warden coordination profile, session event schemas, activation helpers,
adapter pack, evaluation scenarios, and contract documentation for shared memory
across worker, agent session, and operator CLI surfaces.
This commit is contained in:
tegwick 2026-07-02 23:40:45 +02:00
parent 1c3fa03533
commit dc699be976
9 changed files with 891 additions and 8 deletions

View file

@ -126,6 +126,30 @@ from .pilot import (
managed_deployment_pilot_report,
write_live_pilot_evidence,
)
from .ops_warden import (
KNOWN_AGENT_IDS,
OPS_WARDEN_ACTIVATION_SCHEMA,
OPS_WARDEN_ADAPTER_PACK_NAME,
OPS_WARDEN_MEMORY_STATUS_SCHEMA,
OPS_WARDEN_PROFILE_ID,
OPS_WARDEN_RUNTIME_SCHEMA,
OPS_WARDEN_SESSION_EVENT_SCHEMA,
OpsWardenMemoryStore,
activate_ops_warden_memory,
build_session_event,
default_memory_store_path,
memory_enabled,
ops_warden_adapter_pack,
ops_warden_coordination_profile,
ops_warden_evaluation_metrics,
ops_warden_evaluation_report,
record_session_event,
resolve_session_kind,
session_kind_for_agent,
stabilized_route_match,
validate_memory_write,
validate_ops_warden_profile,
)
from .planner import plan_profile_execution
from .runtime import PhaseMemoryRuntime
from .troubleshooting import (
@ -194,7 +218,15 @@ __all__ = [
"LIVE_PILOT_REPORT_SCHEMA",
"LocalMarkitectValidator",
"OptionalMarkitectValidator",
"OPS_WARDEN_ACTIVATION_SCHEMA",
"OPS_WARDEN_ADAPTER_PACK_NAME",
"OPS_WARDEN_MEMORY_STATUS_SCHEMA",
"OPS_WARDEN_PROFILE_ID",
"OPS_WARDEN_RUNTIME_SCHEMA",
"OPS_WARDEN_SESSION_EVENT_SCHEMA",
"OpsWardenMemoryStore",
"PHASE_MEMORY_CREDENTIAL_NEEDS",
"KNOWN_AGENT_IDS",
"abandon_path",
"branch_path",
"compact_path",
@ -234,7 +266,21 @@ __all__ = [
"package_request_from_selection",
"package_response_envelope",
"WordCountTokenEstimator",
"activate_ops_warden_memory",
"activation_quality_report",
"build_session_event",
"default_memory_store_path",
"memory_enabled",
"ops_warden_adapter_pack",
"ops_warden_coordination_profile",
"ops_warden_evaluation_metrics",
"ops_warden_evaluation_report",
"record_session_event",
"resolve_session_kind",
"session_kind_for_agent",
"stabilized_route_match",
"validate_memory_write",
"validate_ops_warden_profile",
"plan_neighborhood_activation",
"retrieve_graph_neighborhood",
"select_event_path",

View file

@ -0,0 +1,497 @@
"""ops-warden cross-runtime memory contract and activation helpers."""
from __future__ import annotations
import json
import os
import re
from dataclasses import dataclass
from pathlib import Path
from typing import Any, Mapping
from .contracts import profile_from_markitect
from .external_adapters import ExternalAdapterPack, adapter_pack_manifest
from .models import Diagnostic
from .utils import stable_digest, utc_now_iso
OPS_WARDEN_RUNTIME_SCHEMA = "phase_memory.ops_warden.runtime.v1"
OPS_WARDEN_SESSION_EVENT_SCHEMA = "phase_memory.ops_warden.session_event.v1"
OPS_WARDEN_ACTIVATION_SCHEMA = "phase_memory.ops_warden.activation.v1"
OPS_WARDEN_MEMORY_STATUS_SCHEMA = "phase_memory.ops_warden.memory_status.v1"
OPS_WARDEN_PROFILE_ID = "ops-warden-coordination"
OPS_WARDEN_ADAPTER_PACK_NAME = "ops-warden-coordination"
SESSION_KIND_WORKER = "warden.worker"
SESSION_KIND_OPERATOR = "warden.operator"
SESSION_KIND_AGENT_PREFIX = "warden.agent."
KNOWN_AGENT_IDS = ("claude", "codex", "grok")
SECRET_FIELD_PATTERN = re.compile(
r"\b("
r"token|secret|password|api[_ ]?key|private[_ ]?key|"
r"vault[_ ]?token|npm_auth_token|client[_ ]?secret|"
r"bearer\s+[a-z0-9._-]{8,}|"
r"https?://[^\s\"']+"
r")\b",
re.IGNORECASE,
)
_OUTCOMES = frozenset({"resolved", "escalated", "skipped"})
def default_memory_store_path(environ: Mapping[str, str] | None = None) -> Path:
environ = environ or os.environ
override = str(environ.get("WARDEN_MEMORY_STORE") or "").strip()
if override:
return Path(override).expanduser()
xdg_data = str(environ.get("XDG_DATA_HOME") or "").strip()
base = Path(xdg_data).expanduser() if xdg_data else Path.home() / ".local" / "share"
return base / "warden" / "memory"
def memory_enabled(environ: Mapping[str, str] | None = None) -> bool:
environ = environ or os.environ
return str(environ.get("WARDEN_MEMORY", "1")).strip().lower() not in {"0", "false", "no", "off"}
def session_kind_for_agent(agent_id: str | None = None) -> str:
agent_id = str(agent_id or "").strip().lower()
if agent_id:
return f"{SESSION_KIND_AGENT_PREFIX}{agent_id}"
return SESSION_KIND_OPERATOR
def resolve_session_kind(environ: Mapping[str, str] | None = None) -> str:
environ = environ or os.environ
explicit = str(environ.get("WARDEN_SESSION_KIND") or "").strip()
if explicit:
return explicit
agent_id = str(environ.get("WARDEN_AGENT_ID") or "").strip()
return session_kind_for_agent(agent_id or None)
def ops_warden_coordination_profile(path: str | Path | None = None) -> dict[str, Any]:
if path is None:
path = Path(__file__).resolve().parents[2] / "tests" / "fixtures" / "ops-warden-coordination-profile.json"
return json.loads(Path(path).read_text(encoding="utf-8"))
def validate_ops_warden_profile(data: dict[str, Any]) -> dict[str, Any]:
result = profile_from_markitect(data)
diagnostics = list(result.diagnostics)
policy = data.get("policy") if isinstance(data.get("policy"), dict) else {}
if policy.get("secrets_allowed") is not False:
diagnostics.append(
Diagnostic(
"error",
"ops_warden_profile_secrets_not_denied",
"Ops-warden coordination profile must deny secret storage.",
"policy.secrets_allowed",
)
)
activation = data.get("activation") if isinstance(data.get("activation"), dict) else {}
if int(activation.get("max_items") or 0) < 1:
diagnostics.append(
Diagnostic(
"error",
"ops_warden_profile_activation_budget_missing",
"Ops-warden coordination profile must declare activation budgets.",
"activation.max_items",
)
)
return {
"schema_version": OPS_WARDEN_RUNTIME_SCHEMA,
"valid": not any(item.severity == "error" for item in diagnostics),
"profile_id": str(data.get("id") or ""),
"diagnostics": [item.to_dict() for item in diagnostics],
}
def validate_memory_write(payload: Mapping[str, Any]) -> tuple[bool, tuple[Diagnostic, ...]]:
diagnostics: list[Diagnostic] = []
serialized = json.dumps(dict(payload), sort_keys=True)
if SECRET_FIELD_PATTERN.search(serialized):
diagnostics.append(
Diagnostic(
"error",
"ops_warden_memory_secret_field_rejected",
"Memory write appears to contain secret values or raw endpoint URLs.",
"payload",
)
)
outcome = str(payload.get("outcome") or "")
if outcome and outcome not in _OUTCOMES:
diagnostics.append(
Diagnostic(
"error",
"ops_warden_memory_invalid_outcome",
"Session event outcome must be resolved, escalated, or skipped.",
"outcome",
{"actual": outcome},
)
)
return not diagnostics, tuple(diagnostics)
def need_fingerprint(need: str) -> str:
return stable_digest(str(need or "").strip().lower())
@dataclass(frozen=True)
class OpsWardenMemoryStore:
root: Path
@classmethod
def open(cls, path: str | Path | None = None, environ: Mapping[str, str] | None = None) -> OpsWardenMemoryStore:
store = cls((Path(path) if path is not None else default_memory_store_path(environ)).expanduser())
store.ensure_layout()
return store
def ensure_layout(self) -> None:
self.root.mkdir(parents=True, exist_ok=True)
if not self.metadata_path.exists():
self.metadata_path.write_text(
json.dumps(
{
"schema_version": OPS_WARDEN_RUNTIME_SCHEMA,
"profile_id": OPS_WARDEN_PROFILE_ID,
"created_at": utc_now_iso(),
},
indent=2,
sort_keys=True,
)
+ "\n",
encoding="utf-8",
)
@property
def metadata_path(self) -> Path:
return self.root / "metadata.json"
@property
def events_path(self) -> Path:
return self.root / "events.jsonl"
def list_events(self) -> list[dict[str, Any]]:
if not self.events_path.exists():
return []
events: list[dict[str, Any]] = []
for line in self.events_path.read_text(encoding="utf-8").splitlines():
if not line.strip():
continue
data = json.loads(line)
if isinstance(data, dict):
events.append(data)
return events
def append_event(self, event: Mapping[str, Any]) -> dict[str, Any]:
payload = dict(event)
payload.setdefault("schema_version", OPS_WARDEN_SESSION_EVENT_SCHEMA)
payload.setdefault("recorded_at", utc_now_iso())
ok, diagnostics = validate_memory_write(payload)
if not ok:
return {
"valid": False,
"diagnostics": [item.to_dict() for item in diagnostics],
}
event_id = str(payload.get("event_id") or f"ops-warden-event:{stable_digest(payload)}")
payload["event_id"] = event_id
with self.events_path.open("a", encoding="utf-8") as handle:
handle.write(json.dumps(payload, sort_keys=True) + "\n")
return {"valid": True, "event": payload, "diagnostics": []}
def status(self) -> dict[str, Any]:
events = self.list_events()
counts: dict[str, int] = {}
last_activation = ""
for event in events:
kind = str(event.get("session_kind") or "unknown")
counts[kind] = counts.get(kind, 0) + 1
if event.get("command") == "memory.activate":
last_activation = str(event.get("recorded_at") or last_activation)
return {
"schema_version": OPS_WARDEN_MEMORY_STATUS_SCHEMA,
"valid": True,
"store_path": str(self.root),
"profile_id": OPS_WARDEN_PROFILE_ID,
"episode_count": len(events),
"episode_counts_by_session_kind": dict(sorted(counts.items())),
"last_activation_at": last_activation,
"diagnostics": [],
}
def build_session_event(
*,
command: str,
session_kind: str,
outcome: str,
need: str = "",
route_id: str = "",
agent_id: str = "",
session_id: str = "",
diagnostic_codes: list[str] | None = None,
metadata: Mapping[str, Any] | None = None,
) -> dict[str, Any]:
return {
"schema_version": OPS_WARDEN_SESSION_EVENT_SCHEMA,
"session_kind": session_kind,
"agent_id": agent_id,
"session_id": session_id,
"command": command,
"need_fingerprint": need_fingerprint(need) if need else "",
"route_id": route_id,
"outcome": outcome,
"diagnostic_codes": list(diagnostic_codes or ()),
"metadata": dict(metadata or {}),
}
def record_session_event(
store: OpsWardenMemoryStore | Path | str,
event: Mapping[str, Any],
) -> dict[str, Any]:
memory = store if isinstance(store, OpsWardenMemoryStore) else OpsWardenMemoryStore.open(store)
return memory.append_event(event)
def stabilized_route_match(
events: list[dict[str, Any]],
*,
need: str = "",
need_fingerprint_value: str = "",
min_confirmations: int = 2,
) -> dict[str, Any] | None:
fingerprint = need_fingerprint_value or (need_fingerprint(need) if need else "")
if not fingerprint:
return None
matches = [
event
for event in events
if event.get("need_fingerprint") == fingerprint
and event.get("outcome") == "resolved"
and event.get("route_id")
]
if len(matches) < min_confirmations:
return None
route_id = str(matches[-1].get("route_id") or "")
return {
"need_fingerprint": fingerprint,
"route_id": route_id,
"confirmations": len(matches),
"source_event_ids": [str(item.get("event_id") or "") for item in matches[-min_confirmations:]],
}
def activate_ops_warden_memory(
store: OpsWardenMemoryStore | Path | str,
*,
session_kind: str,
need: str = "",
session_id: str = "",
) -> dict[str, Any]:
memory = store if isinstance(store, OpsWardenMemoryStore) else OpsWardenMemoryStore.open(store)
profile = ops_warden_coordination_profile()
profile_validation = validate_ops_warden_profile(profile)
events = memory.list_events()
fingerprint = need_fingerprint(need) if need else ""
stabilized = stabilized_route_match(events, need_fingerprint_value=fingerprint) if fingerprint else None
recent = list(reversed(events))[: int(profile.get("activation", {}).get("max_items") or 8)]
selected: list[dict[str, Any]] = []
if stabilized:
selected.append(
{
"kind": "stabilized_route",
"route_id": stabilized["route_id"],
"need_fingerprint": stabilized["need_fingerprint"],
"confirmations": stabilized["confirmations"],
}
)
for event in recent:
if len(selected) >= int(profile.get("activation", {}).get("max_items") or 8):
break
if fingerprint and event.get("need_fingerprint") != fingerprint and event.get("session_kind") != session_kind:
continue
selected.append(
{
"kind": "episode",
"event_id": event.get("event_id", ""),
"session_kind": event.get("session_kind", ""),
"command": event.get("command", ""),
"route_id": event.get("route_id", ""),
"outcome": event.get("outcome", ""),
"diagnostic_codes": list(event.get("diagnostic_codes") or ()),
}
)
activation = {
"schema_version": OPS_WARDEN_ACTIVATION_SCHEMA,
"id": f"ops-warden-activation:{stable_digest([session_kind, fingerprint, selected])}",
"valid": profile_validation["valid"],
"session_kind": session_kind,
"session_id": session_id,
"profile_id": OPS_WARDEN_PROFILE_ID,
"need_fingerprint": fingerprint,
"stabilized_route": stabilized,
"selected_episodes": selected,
"episode_count": len(events),
"llm_calls_avoided": bool(stabilized),
"operator_guidance": {
"worker": "Activate before Brain.plan; record after execute.",
"agent_session": "Call warden memory activate at session start.",
"operator": "WARDEN_AGENT_ID selects warden.agent.<id> session_kind.",
},
"diagnostics": list(profile_validation.get("diagnostics", ())),
}
memory.append_event(
build_session_event(
command="memory.activate",
session_kind=session_kind,
outcome="resolved",
session_id=session_id,
metadata={"selected_count": len(selected), "llm_calls_avoided": bool(stabilized)},
)
)
return activation
def ops_warden_evaluation_report(scenarios_data: dict[str, Any]) -> dict[str, Any]:
scenarios = list(scenarios_data.get("scenarios") or ())
scenario_reports: list[dict[str, Any]] = []
metrics = {
"scenario_count": len(scenarios),
"routing_repeat_accuracy": 0.0,
"cross_runtime_continuity": 0.0,
"llm_calls_avoided_count": 0.0,
}
for scenario in scenarios:
events = list(scenario.get("events") or ())
episode_metrics = ops_warden_evaluation_metrics(events)
expect = dict(scenario.get("expect") or {})
diagnostics: list[dict[str, Any]] = []
if expect.get("stabilized") and not stabilized_route_match(events, need_fingerprint_value=str(events[0].get("need_fingerprint") or "")):
diagnostics.append(
Diagnostic(
"error",
"ops_warden_stabilized_route_missing",
"Scenario expected a stabilized route match.",
"events",
).to_dict()
)
if expect.get("continuity") and episode_metrics["cross_runtime_continuity"] < 1.0:
diagnostics.append(
Diagnostic(
"error",
"ops_warden_cross_runtime_continuity_missing",
"Scenario expected cross-runtime continuity.",
"events",
).to_dict()
)
if expect.get("llm_calls_avoided") and episode_metrics["llm_calls_avoided_count"] < 1.0:
diagnostics.append(
Diagnostic(
"error",
"ops_warden_llm_avoidance_missing",
"Scenario expected an avoided llm-connect call.",
"events",
).to_dict()
)
scenario_reports.append(
{
"id": scenario.get("id", ""),
"metrics": episode_metrics,
"diagnostics": diagnostics,
}
)
for key in ("routing_repeat_accuracy", "cross_runtime_continuity", "llm_calls_avoided_count"):
if key in episode_metrics:
metrics[key] = max(float(metrics[key]), float(episode_metrics[key]))
return {
"schema_version": "phase_memory.ops_warden.evaluation_report.v1",
"valid": not any(item.get("severity") == "error" for report in scenario_reports for item in report.get("diagnostics", ())),
"metrics": metrics,
"scenarios": scenario_reports,
"diagnostics": [
item
for report in scenario_reports
for item in report.get("diagnostics", ())
if item.get("severity") == "error"
],
}
def ops_warden_evaluation_metrics(events: list[dict[str, Any]]) -> dict[str, float]:
total = float(len(events))
resolved = float(sum(1 for event in events if event.get("outcome") == "resolved"))
escalated = float(sum(1 for event in events if event.get("outcome") == "escalated"))
activations = float(sum(1 for event in events if event.get("command") == "memory.activate"))
llm_avoided = float(sum(1 for event in events if (event.get("metadata") or {}).get("llm_calls_avoided")))
agent_sessions = float(
sum(1 for event in events if str(event.get("session_kind") or "").startswith(SESSION_KIND_AGENT_PREFIX))
)
worker_sessions = float(sum(1 for event in events if event.get("session_kind") == SESSION_KIND_WORKER))
continuity = 1.0 if agent_sessions > 0 and worker_sessions > 0 else 0.0
return {
"episode_count": total,
"resolved_episode_count": resolved,
"escalated_episode_count": escalated,
"activation_count": activations,
"llm_calls_avoided_count": llm_avoided,
"cross_runtime_continuity": continuity,
"routing_repeat_accuracy": (resolved / total) if total else 0.0,
}
def ops_warden_adapter_pack() -> ExternalAdapterPack:
from .adapters import InMemoryMemoryEventLog, InMemoryMemoryGraphStore
from .external_adapters import (
ADAPTER_CONFORMANCE_HELPERS,
FakeExternalPolicyGateway,
FakeKontextualRuntimeRegistry,
FakeMarkitectPackageCompiler,
FakeTelemetryAuditSink,
LiveShapedPermissionSemanticIndex,
)
class OpsWardenEventLog(InMemoryMemoryEventLog):
"""Event log shape for ops-warden coordination episodes."""
class OpsWardenGraphStore(InMemoryMemoryGraphStore):
"""Graph store shape for stabilized routing nodes."""
capability_requirements = {
"graph_store": ("ops-warden.coordination.graph",),
"event_log": ("ops-warden.coordination.events",),
"policy_gateway": ("ops-warden.coordination.policy",),
"audit_sink": ("ops-warden.coordination.audit",),
"package_compiler": ("ops-warden.coordination.activation",),
"semantic_index": ("ops-warden.coordination.retrieval",),
"runtime_registry": ("ops-warden.coordination.registry",),
}
ownership_boundaries = {
"graph_store": "phase-memory owns stabilized routing nodes; ops-warden owns episode ingestion",
"event_log": "phase-memory owns coordination episodes; ops-warden owns CLI/worker writers",
"policy_gateway": "ops-warden charter is rigid import-only; phase-memory never relaxes guardrails",
"audit_sink": "phase-memory owns redacted audit schema; ops-warden owns warden activity correlation",
"package_compiler": "phase-memory owns activation packages for worker and agent sessions",
"semantic_index": "phase-memory owns need-fingerprint retrieval for routing memory",
"runtime_registry": "ops-warden owns session_kind routing; phase-memory owns runtime envelopes",
}
adapters = {
"graph_store": OpsWardenGraphStore(),
"event_log": OpsWardenEventLog(),
"policy_gateway": FakeExternalPolicyGateway(),
"audit_sink": FakeTelemetryAuditSink(),
"package_compiler": FakeMarkitectPackageCompiler(),
"semantic_index": LiveShapedPermissionSemanticIndex(),
"runtime_registry": FakeKontextualRuntimeRegistry(),
}
return ExternalAdapterPack(
name=OPS_WARDEN_ADAPTER_PACK_NAME,
adapters=adapters,
capabilities=tuple(sorted({capability for values in capability_requirements.values() for capability in values})),
ownership_boundaries=ownership_boundaries,
required_conformance=dict(ADAPTER_CONFORMANCE_HELPERS),
capability_requirements=capability_requirements,
metadata={"profile_id": OPS_WARDEN_PROFILE_ID, "runtime_schema": OPS_WARDEN_RUNTIME_SCHEMA},
)