Compare commits

..

10 commits

Author SHA1 Message Date
2336b68a9a Add Forgejo CI smoke workflow (enablement template)
All checks were successful
CI Smoke / host-smoke (push) Successful in 1s
CI Smoke / container-smoke (push) Successful in 1m29s
2026-07-08 12:27:45 +02:00
39a053bfd0 Draft capability entry (reuse-surface REUSE-WP-0017-T04, cohort 3)
Honest first-pass maturity vector grounded in README/docs/tests present
in this repo; no invented evidence. Flagged for human review before
publish. See reuse-surface history/2026-07-06-coverage-classification.md.

Co-Authored-By: Claude Sonnet 5 <noreply@anthropic.com>
2026-07-06 19:50:53 +02:00
d72f64d7a0 fix: make install uses uv tool install without requiring a venv
uv pip install needs a virtual environment. Use uv tool install for the
global CLI path and make sync for editable .venv development installs.
2026-07-03 01:34:57 +02:00
fcea33924b chore: sync PMEM-WP-0017 State Hub workstream and task IDs 2026-07-03 01:25:16 +02:00
feb60c1d96 feat(PMEM-WP-0017): federated store management CLI and activity reports
Add phase_memory.management for cross-store discovery and windowed
activity reporting. Extend the phase-memory CLI with stores list and
report commands, plus make report-7d for the default weekly operator view.
2026-07-03 01:24:16 +02:00
0320b112fd Add Makefile install and test targets for local development. 2026-07-03 00:54:25 +02:00
592577f7cd Document implicit default phase-memory activation in ops-warden. 2026-07-03 00:49:39 +02:00
bea81836a1 Remove accidental vim swap file from tracking. 2026-07-02 23:40:55 +02:00
dc699be976 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.
2026-07-02 23:40:45 +02:00
1c3fa03533 Add PMEM-WP-0016 ops-warden cross-runtime memory workplan.
Define phase-memory profile, agent session schemas, and activation contracts
for worker, coding agent, and operator CLI surfaces sharing one canonical store.
2026-07-02 23:31:40 +02:00
26 changed files with 2472 additions and 5 deletions

View file

@ -0,0 +1,29 @@
# Canonical CI smoke template (tier 1 routing drill).
# Copy to: .forgejo/workflows/ci-smoke.yaml in consumer repos.
name: CI Smoke
on:
push:
branches:
- main
workflow_dispatch:
jobs:
host-smoke:
runs-on: self-hosted
steps:
- name: Routing probe (host runner)
run: |
set -eu
echo "repository=${GITHUB_REPOSITORY:-unknown}"
echo "sha=${GITHUB_SHA:-unknown}"
echo "runner=${RUNNER_NAME:-unknown}"
uname -a
container-smoke:
runs-on: ubuntu-latest
steps:
- name: Routing probe (container label)
run: |
set -eu
echo "container-smoke ok for ${GITHUB_REPOSITORY:-unknown}"

1
.gitignore vendored
View file

@ -177,3 +177,4 @@ cython_debug/
# PyPI configuration file
.pypirc
*.swp

42
Makefile Normal file
View file

@ -0,0 +1,42 @@
# phase-memory development and install targets
ROOT := $(dir $(abspath $(lastword $(MAKEFILE_LIST))))
.DEFAULT_GOAL := help
.PHONY: help sync install install-editable check test report-7d report
help:
@echo "phase-memory make targets"
@echo ""
@echo " make install # install phase-memory CLI (uv tool, no venv)"
@echo " make sync # create .venv and editable install (dev)"
@echo " make install-editable # alias for make sync"
@echo " make check # verify import and print version"
@echo " make test # run pytest"
@echo " make report-7d # federated activity report (last 7 days)"
@echo " make report # federated report (REPORT_DAYS=7 default)"
@echo ""
@echo "For ops-warden integration, prefer: cd ../ops-warden && make install-all"
sync:
uv venv --allow-existing
uv pip install -e "$(ROOT)"
install-editable: sync
install:
uv tool install . --force
check:
@PYTHONPATH="$(ROOT)src" python3 -c "import phase_memory; print('phase-memory', phase_memory.__version__, 'ok')"
test:
PYTHONPATH="$(ROOT)src" python3 -m pytest
REPORT_DAYS ?= 7
report-7d:
PYTHONPATH="$(ROOT)src" python3 -m phase_memory.cli report --days 7 --format summary
report:
PYTHONPATH="$(ROOT)src" python3 -m phase_memory.cli report --days $(REPORT_DAYS) --format summary

View file

@ -30,7 +30,15 @@ Adjacent repositories own nearby but separate concerns:
## Quick Start
```bash
python3 -m pytest
make test
```
Install for local development or ops-warden integration:
```bash
make install # global CLI: phase-memory, phase-memory-service
make sync # dev: editable install into .venv
# or from ops-warden: make install-all
```
The default test suite uses only deterministic local fixtures.
@ -47,8 +55,25 @@ PYTHONPATH=src python3 -m phase_memory.cli store import --store .phase-memory-lo
When installed, the package exposes the same commands as `phase-memory`.
Commands emit JSON runtime envelopes by default and accept `--format summary`
for a concise human-readable view. All current commands are dry-run planning
operations; they do not mutate durable memory stores.
for a concise human-readable view. Profile, graph, and store import/export
commands are dry-run planning operations; they do not mutate durable memory
stores by default.
## Management Reports
Inspect activity across every discovered phase-memory store (ops-warden
coordination store, registry entries, and `PHASE_MEMORY_STORE_PATHS`):
```bash
phase-memory stores list --format summary
phase-memory report --days 7 --format summary
phase-memory report --store ops-warden-default --days 30
make report-7d
```
Discovery honours `WARDEN_MEMORY_STORE`, `PHASE_MEMORY_REGISTRY`, and
`PHASE_MEMORY_STORE_PATHS`. Reports are metadata-only and never include secret
values.
## Local Runtime

View file

@ -55,6 +55,20 @@ Routes:
- `POST /operations/{operation}`
- `POST /operations` with `{"operation": "...", "payload": {...}}`
## Cross-Store Activity
Use the phase-memory CLI as the canonical cross-store activity check:
```bash
phase-memory stores list --format summary
phase-memory report --days 7 --format summary
make report-7d
```
Federated reports aggregate ops-warden coordination episodes and local graph
store events/audit lines for the requested window. Drill into one store with
`phase-memory report --store <store_id-or-path> --days N`.
## Readiness Checks
Before accepting traffic:

View file

@ -0,0 +1,80 @@
# Ops-Warden Cross-Runtime Memory Contract
Updated: 2026-07-02
## Purpose
phase-memory provides a shared experiential substrate for ops-warden across:
- `warden worker` ticks with llm-connect / OpenRouter inference
- Coding agent sessions (Claude Code, Codex, Grok, future agents)
- Direct operator CLI use
All runtimes use one canonical store and the same redaction rules.
## Canonical Store
| Setting | Default |
| --- | --- |
| Path | `~/.local/share/warden/memory/` |
| Override | `WARDEN_MEMORY_STORE` |
| Opt-out | `WARDEN_MEMORY=0` |
Layout:
- `metadata.json` — store schema and profile id
- `events.jsonl` — append-only session episodes
## Session Kinds
| Runtime | `session_kind` | Selection |
| --- | --- | --- |
| Worker | `warden.worker` | `WARDEN_SESSION_KIND=warden.worker` during worker ticks |
| Operator CLI | `warden.operator` | default when `WARDEN_AGENT_ID` is unset |
| Agent session | `warden.agent.<id>` | `WARDEN_AGENT_ID=claude|codex|grok|...` |
## Event Schema
Schema: `phase_memory.ops_warden.session_event.v1`
Required metadata fields:
- `session_kind`, `command`, `outcome` (`resolved|escalated|skipped`)
- `need_fingerprint` (digest only — never the raw need text when sensitive)
- `route_id`, `agent_id`, `session_id`, `diagnostic_codes`
Rejected writes containing token-like values or raw URLs.
## Activation
`activate_ops_warden_memory` returns:
- stabilized route matches (verified repeats)
- recent episodes filtered by session kind / need fingerprint
- `llm_calls_avoided` when stabilized memory covers the need
Rigid memory (charter, allowlist) is import-only and never promoted from episodes.
## Default Behavior
Memory is enabled by default (`WARDEN_MEMORY=1`). ops-warden implicitly calls
`ensure_memory_context` on every command; explicit `warden memory activate` is
optional introspection only.
## Agent Session Orientation
Coding agents should:
1. Set `WARDEN_AGENT_ID` to their runtime id (claude, codex, grok, or future).
2. Use normal `warden route` / `warden access` commands; memory loads and episodes
are recorded automatically.
## Evaluation
Ops-warden memory quality is measured by:
- `routing_repeat_accuracy`
- `cross_runtime_continuity`
- `llm_calls_avoided_count`
Use `evaluation_trend_regression_gate` when promoting worker or routing changes.

View file

@ -0,0 +1,121 @@
---
id: capability.memory.phase-planning
name: Profile-Driven Memory Phase Planning (phase-memory)
summary: Interprets Markitect memory profiles as runtime plans, modeling memory phases and producing deterministic
dry-run actions for retention, refresh, compaction, stabilization, and activation.
owner: phase-memory
status: draft
domain: communication
tags:
- memory
- agentic
- planning
maturity:
discovery:
current: D3
target: D5
confidence: medium
rationale: README explicitly documents the repo's boundary against adjacent repos (markitect-tool
owns profile/graph/event contracts; kontextual-engine owns durable storage/retrieval; infospace-bench
owns pilots/evaluation) and states the first slice is local-first, dependency-light, dry-run only.
availability:
current: A1
target: A3
confidence: medium
rationale: Python package (`phase-memory`), pip-installable; explicitly does not launch a service
or mutate durable memory stores by default in this first slice.
external_evidence:
completeness:
level: C1
confidence: low
basis: scope_vs_intent_and_consumer_expectations
satisfied_expectations:
- memory phase modeling (retention, refresh, compaction, stabilization, activation)
- deterministic dry-run action generation
- explicit boundary docs vs three adjacent repos
broken_expectations: []
out_of_scope_expectations: []
reliability:
level: R0
confidence: low
basis: consumer_quality_signals
known_reliability_risks:
- first slice is dry-run only; does not yet mutate durable stores
discovery:
intent: Interpret Markitect memory profiles as runtime plans and produce deterministic dry-run actions
for memory-phase transitions, as the operating layer between profile definition (markitect-tool) and
durable storage (kontextual-engine).
includes:
- memory phase modeling and deterministic dry-run planning for retention/refresh/compaction/stabilization/activation
excludes:
- memory profile/graph/event contract definition (markitect-tool)
- durable storage and permission-aware retrieval (kontextual-engine)
- concrete pilots and evaluation (infospace-bench)
assumptions: []
use_cases: []
research_memos: []
availability:
current_level: A1
target_level: A3
current_artifacts:
- Python package (`phase-memory`)
target_artifacts: []
consumption_modes:
- library import
relations:
depends_on: []
supports: []
related_to: []
evidence:
documentation:
- README.md
tests:
- tests/
consumer_feedback: []
bug_reports: []
incidents: []
consumer_guidance:
recommended_for:
- agentic systems needing deterministic memory-phase planning distinct from storage or profile-contract
concerns
not_recommended_for:
- needs for durable memory mutation today (first slice is dry-run only)
known_limitations:
- local-first, dependency-light first slice; no service or durable mutation yet
promotion_history: []
---
# Profile-Driven Memory Phase Planning (phase-memory)
## Overview
`phase-memory` is the profile-driven memory operating layer for agentic systems: it interprets Markitect memory profiles as runtime plans and produces deterministic dry-run actions for memory phases (retention, refresh, compaction, stabilization, activation), explicitly bounded against adjacent repos that own contracts (markitect-tool), durable storage (kontextual-engine), and evaluation (infospace-bench).
## Assessment notes
### Discovery
README explicitly documents the repo's boundary against adjacent repos (markitect-tool owns profile/graph/event contracts; kontextual-engine owns durable storage/retrieval; infospace-bench owns pilots/evaluation) and states the first slice is local-first, dependency-light, dry-run only.
### Availability
Python package (`phase-memory`), pip-installable; explicitly does not launch a service or mutate durable memory stores by default in this first slice.
### Completeness
First-pass honest assessment from the REUSE-WP-0017 coverage campaign
(reuse-surface). No external consumer feedback exists yet; levels reflect
scope-vs-intent documentation quality, not internal code quality.
### Reliability
No production consumer telemetry exists yet; reliability level is
intentionally conservative pending REUSE-WP-0019 reuse-telemetry evidence.
## Promotion checklist
- [x] ID follows `capability.<domain>.<name>` pattern
- [x] Maturity enums match `specs/CapabilityMaturityStandard.md`
- [x] `external_evidence` is populated separately from `maturity`
- [ ] Relations reference valid capability IDs (none yet)
- [x] Index entry added in `registry/indexes/capabilities.yaml`

View file

@ -1,4 +1,19 @@
version: 1
updated: '2026-06-16'
updated: '2026-07-06'
domain: helix_forge
capabilities: []
capabilities:
- id: capability.memory.phase-planning
name: Profile-Driven Memory Phase Planning (phase-memory)
summary: Interprets Markitect memory profiles as runtime plans, modeling memory phases and producing
deterministic dry-run actions for retention, refresh, compaction, stabilization, and activation.
vector: D3 / A1 / C1 / R0
domain: communication
status: draft
owner: phase-memory
path: registry/capabilities/capability.memory.phase-planning.md
tags:
- memory
- agentic
- planning
consumption_modes:
- library import

View file

@ -126,6 +126,39 @@ from .pilot import (
managed_deployment_pilot_report,
write_live_pilot_evidence,
)
from .management import (
FEDERATED_REPORT_SCHEMA,
STORE_LIST_SCHEMA,
STORE_REGISTRY_SCHEMA,
build_federated_report,
build_store_list,
classify_store,
discover_memory_stores,
)
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 (
@ -155,6 +188,9 @@ __all__ = [
"FakeExternalPolicyGateway",
"FakeExternalSemanticIndex",
"FakeKontextualRuntimeRegistry",
"FEDERATED_REPORT_SCHEMA",
"STORE_LIST_SCHEMA",
"STORE_REGISTRY_SCHEMA",
"FakeMarkitectPackageCompiler",
"FakeTelemetryAuditSink",
"LiveShapedKontextualEventLog",
@ -194,7 +230,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 +278,23 @@ __all__ = [
"package_request_from_selection",
"package_response_envelope",
"WordCountTokenEstimator",
"activate_ops_warden_memory",
"activation_quality_report",
"build_federated_report",
"build_session_event",
"build_store_list",
"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",
@ -252,7 +312,9 @@ __all__ = [
"WARDEN_ACCESS_NEED",
"WARDEN_ROUTE_FIND_QUERY",
"build_service_binding",
"classify_store",
"create_wsgi_app",
"discover_memory_stores",
"health_report",
"resolve_credentialed_environ",
"resolve_runtime_adapters",

View file

@ -9,6 +9,7 @@ from pathlib import Path
from typing import Any, Sequence
from .adapters import FileBackedMemoryGraphStore, JsonlAuditSink, JsonlMemoryEventLog
from .management import build_federated_report, build_store_list
from .runtime import PhaseMemoryRuntime
@ -80,6 +81,33 @@ def build_parser() -> argparse.ArgumentParser:
_add_format(store_repair)
store_repair.set_defaults(func=_store_repair)
stores = subparsers.add_parser("stores", help="Federated store discovery")
stores_subparsers = stores.add_subparsers(dest="stores_command")
stores_list = stores_subparsers.add_parser(
"list",
help="List discovered phase-memory stores (defaults, registry, PHASE_MEMORY_STORE_PATHS)",
)
_add_format(stores_list)
stores_list.set_defaults(func=_stores_list)
report = subparsers.add_parser(
"report",
help="Activity report across discovered stores for the last N days",
epilog=(
"Discovery: ops-warden default store, PHASE_MEMORY_REGISTRY, "
"PHASE_MEMORY_STORE_PATHS (colon-separated). Set WARDEN_MEMORY_STORE "
"to override the ops-warden store path."
),
formatter_class=argparse.RawDescriptionHelpFormatter,
)
report.add_argument("--days", type=int, default=7, help="Look back N days (default: 7)")
report.add_argument(
"--store",
help="Focus on one store by registry store_id or absolute path",
)
_add_format(report)
report.set_defaults(func=_report)
return parser
@ -148,6 +176,14 @@ def _store_repair(args: argparse.Namespace, runtime: PhaseMemoryRuntime) -> dict
return _runtime_for_store(args.store).repair_diagnostics(source_ref=str(args.store))
def _stores_list(args: argparse.Namespace, runtime: PhaseMemoryRuntime) -> dict[str, Any]:
return build_store_list()
def _report(args: argparse.Namespace, runtime: PhaseMemoryRuntime) -> dict[str, Any]:
return build_federated_report(days=args.days, focus_store_id=args.store, focus_store_path=args.store)
def _missing_command(args: argparse.Namespace, runtime: PhaseMemoryRuntime) -> dict[str, Any]:
raise SystemExit("Missing command. Use --help for usage.")
@ -200,6 +236,36 @@ def _print_summary(envelope: dict[str, Any]) -> None:
if activation:
print(f"selected_nodes={len(activation.get('selected_node_ids', []))}")
print(f"omitted={len(activation.get('omitted', []))}")
if envelope.get("schema_version") == "phase_memory.management.store_list.v1":
print(f"stores={envelope.get('store_count', 0)}")
for store in envelope.get("stores", ()):
label = store.get("label") or store.get("store_id")
print(f"- {label} [{store.get('store_kind')}] {store.get('path')}")
if envelope.get("schema_version") == "phase_memory.management.federated_report.v1":
aggregate = envelope.get("aggregate") or {}
print(
f"window={envelope.get('window_start')}..{envelope.get('window_end')} "
f"stores={envelope.get('store_count', 0)}"
)
print(
f"episodes={aggregate.get('episode_count', 0)} "
f"audit={aggregate.get('audit_event_count', 0)} "
f"active_stores={aggregate.get('active_store_count', 0)}"
)
for store in envelope.get("stores", ()):
print(
f"- {store.get('store_id')} [{store.get('store_kind')}] "
f"episodes={store.get('episode_count', 0)} "
f"audit={store.get('audit_event_count', 0)} "
f"last={store.get('last_activity_at') or '—'}"
)
detail = envelope.get("store_detail")
if detail:
print(f"detail={detail.get('store_id')} schema={detail.get('schema_version') or '—'}")
if detail.get("repair_diagnostics"):
print(f"repair_diagnostics={len(detail.get('repair_diagnostics', []))}")
if detail.get("episode_timeline"):
print(f"timeline={len(detail.get('episode_timeline', []))}")
if __name__ == "__main__":

View file

@ -0,0 +1,713 @@
"""Federated store discovery and activity reporting."""
from __future__ import annotations
import json
import os
from dataclasses import dataclass, field
from datetime import datetime, timedelta, timezone
from pathlib import Path
from typing import Any, Mapping
from .adapters import LOCAL_STORE_SCHEMA, FileBackedMemoryGraphStore, JsonlAuditSink, JsonlMemoryEventLog
from .models import Diagnostic
from .ops_warden import OPS_WARDEN_PROFILE_ID, OPS_WARDEN_RUNTIME_SCHEMA, OpsWardenMemoryStore, default_memory_store_path
from .utils import parse_iso_datetime, stable_digest, utc_now_iso
STORE_REGISTRY_SCHEMA = "phase_memory.management.store_registry.v1"
STORE_LIST_SCHEMA = "phase_memory.management.store_list.v1"
FEDERATED_REPORT_SCHEMA = "phase_memory.management.federated_report.v1"
STORE_KIND_OPS_WARDEN = "ops_warden_coordination"
STORE_KIND_LOCAL_GRAPH = "local_graph"
DEFAULT_OPS_WARDEN_STORE_ID = "ops-warden-default"
TOP_ROUTE_IDS = 10
DETAIL_EPISODE_LIMIT = 20
@dataclass(frozen=True)
class MemoryStoreDescriptor:
store_id: str
path: Path
store_kind: str
profile_id: str = ""
label: str = ""
source: str = "discovered"
def default_registry_path(environ: Mapping[str, str] | None = None) -> Path:
environ = environ or os.environ
override = str(environ.get("PHASE_MEMORY_REGISTRY") 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 / "phase-memory" / "stores.json"
def classify_store(path: str | Path) -> str | None:
root = Path(path).expanduser().resolve()
local_metadata = root / "phase-memory.json"
if local_metadata.is_file():
try:
data = json.loads(local_metadata.read_text(encoding="utf-8"))
except json.JSONDecodeError:
return None
if str(data.get("schema_version") or "") == LOCAL_STORE_SCHEMA:
return STORE_KIND_LOCAL_GRAPH
ops_metadata = root / "metadata.json"
if ops_metadata.is_file():
try:
data = json.loads(ops_metadata.read_text(encoding="utf-8"))
except json.JSONDecodeError:
return None
if str(data.get("schema_version") or "") == OPS_WARDEN_RUNTIME_SCHEMA:
return STORE_KIND_OPS_WARDEN
return None
def load_store_registry(
path: str | Path | None = None,
environ: Mapping[str, str] | None = None,
) -> dict[str, Any]:
registry_path = Path(path).expanduser() if path is not None else default_registry_path(environ)
diagnostics: list[dict[str, Any]] = []
if not registry_path.is_file():
return {
"schema_version": STORE_REGISTRY_SCHEMA,
"valid": True,
"registry_path": str(registry_path),
"entries": [],
"diagnostics": diagnostics,
}
try:
payload = json.loads(registry_path.read_text(encoding="utf-8"))
except json.JSONDecodeError as exc:
return {
"schema_version": STORE_REGISTRY_SCHEMA,
"valid": False,
"registry_path": str(registry_path),
"entries": [],
"diagnostics": [
Diagnostic(
"warn",
"corrupt_store_registry",
"Store registry file is not valid JSON.",
str(registry_path),
{"error": str(exc)},
).to_dict()
],
}
if not isinstance(payload, dict):
diagnostics.append(
Diagnostic(
"warn",
"invalid_store_registry_shape",
"Store registry must be a JSON object.",
str(registry_path),
).to_dict()
)
return {
"schema_version": STORE_REGISTRY_SCHEMA,
"valid": False,
"registry_path": str(registry_path),
"entries": [],
"diagnostics": diagnostics,
}
schema = str(payload.get("schema_version") or "")
if schema and schema != STORE_REGISTRY_SCHEMA:
diagnostics.append(
Diagnostic(
"warn",
"unknown_store_registry_schema",
"Store registry declares an unknown schema version.",
str(registry_path),
{"schema_version": schema},
).to_dict()
)
entries: list[dict[str, Any]] = []
for item in payload.get("stores") or ():
if not isinstance(item, dict):
continue
store_path = str(item.get("path") or "").strip()
store_kind = str(item.get("store_kind") or "").strip()
store_id = str(item.get("store_id") or "").strip()
if not store_path or not store_kind or not store_id:
diagnostics.append(
Diagnostic(
"warn",
"incomplete_registry_entry",
"Registry entry must include store_id, path, and store_kind.",
store_id or store_path or "unknown",
).to_dict()
)
continue
entries.append(item)
return {
"schema_version": STORE_REGISTRY_SCHEMA,
"valid": not any(item.get("severity") == "error" for item in diagnostics),
"registry_path": str(registry_path),
"entries": entries,
"diagnostics": diagnostics,
}
def _store_id_for_path(path: Path, environ: Mapping[str, str] | None = None) -> str:
resolved = path.expanduser().resolve()
default_ops = default_memory_store_path(environ).expanduser().resolve()
if resolved == default_ops:
return DEFAULT_OPS_WARDEN_STORE_ID
return f"store:{stable_digest(str(resolved))}"
def _descriptor_from_path(
path: Path,
*,
store_kind: str,
source: str,
profile_id: str = "",
label: str = "",
store_id: str = "",
environ: Mapping[str, str] | None = None,
) -> MemoryStoreDescriptor:
resolved = path.expanduser().resolve()
return MemoryStoreDescriptor(
store_id=store_id or _store_id_for_path(resolved, environ),
path=resolved,
store_kind=store_kind,
profile_id=profile_id,
label=label,
source=source,
)
def discover_memory_stores(
environ: Mapping[str, str] | None = None,
*,
registry_path: str | Path | None = None,
extra_paths: tuple[str | Path, ...] = (),
) -> tuple[list[MemoryStoreDescriptor], list[dict[str, Any]]]:
environ = environ or os.environ
diagnostics: list[dict[str, Any]] = []
by_path: dict[str, MemoryStoreDescriptor] = {}
def add_descriptor(descriptor: MemoryStoreDescriptor) -> None:
key = str(descriptor.path)
existing = by_path.get(key)
if existing is None:
by_path[key] = descriptor
return
if descriptor.source == "registry" and existing.source != "registry":
by_path[key] = descriptor
default_ops = default_memory_store_path(environ)
kind = classify_store(default_ops)
if kind:
profile_id = OPS_WARDEN_PROFILE_ID if kind == STORE_KIND_OPS_WARDEN else ""
add_descriptor(
_descriptor_from_path(
default_ops,
store_kind=kind,
source="default",
profile_id=profile_id,
label="ops-warden coordination store",
store_id=DEFAULT_OPS_WARDEN_STORE_ID,
environ=environ,
)
)
registry = load_store_registry(registry_path, environ=environ)
diagnostics.extend(registry.get("diagnostics", ()))
registry_file = Path(str(registry.get("registry_path") or "")).expanduser()
for entry in registry.get("entries", ()):
entry_path = Path(str(entry.get("path") or "")).expanduser()
if not entry_path.is_absolute() and registry_file.is_file():
entry_path = (registry_file.parent / entry_path).resolve()
if not entry_path.is_dir():
diagnostics.append(
Diagnostic(
"warn",
"registry_store_missing",
"Registry entry path does not exist.",
str(entry.get("store_id") or entry_path),
{"path": str(entry_path)},
).to_dict()
)
continue
declared_kind = str(entry.get("store_kind") or "")
actual_kind = classify_store(entry_path)
if actual_kind and actual_kind != declared_kind:
diagnostics.append(
Diagnostic(
"warn",
"registry_store_kind_mismatch",
"Registry store_kind does not match on-disk classification.",
str(entry.get("store_id") or entry_path),
{"declared": declared_kind, "actual": actual_kind},
).to_dict()
)
store_kind = actual_kind or declared_kind
if not store_kind:
continue
add_descriptor(
_descriptor_from_path(
entry_path,
store_kind=store_kind,
source="registry",
profile_id=str(entry.get("profile_id") or ""),
label=str(entry.get("label") or ""),
store_id=str(entry.get("store_id") or ""),
environ=environ,
)
)
env_paths = str(environ.get("PHASE_MEMORY_STORE_PATHS") or "").strip()
path_candidates = [item.strip() for item in env_paths.split(":") if item.strip()]
path_candidates.extend(str(item) for item in extra_paths)
for candidate in path_candidates:
candidate_path = Path(candidate).expanduser()
if not candidate_path.is_dir():
continue
store_kind = classify_store(candidate_path)
if not store_kind:
continue
add_descriptor(
_descriptor_from_path(
candidate_path,
store_kind=store_kind,
source="env",
environ=environ,
)
)
stores = sorted(by_path.values(), key=lambda item: (item.store_kind, item.store_id))
return stores, diagnostics
def resolve_store_reference(
reference: str,
stores: list[MemoryStoreDescriptor],
environ: Mapping[str, str] | None = None,
) -> MemoryStoreDescriptor | None:
ref = str(reference or "").strip()
if not ref:
return None
for store in stores:
if ref == store.store_id or ref == str(store.path):
return store
candidate = Path(ref).expanduser()
if candidate.is_dir():
store_kind = classify_store(candidate)
if store_kind:
return _descriptor_from_path(candidate, store_kind=store_kind, source="explicit", environ=environ)
for store in stores:
if candidate.resolve() == store.path:
return store
return None
def compute_report_window(
*,
days: int,
window_end: datetime | None = None,
) -> tuple[datetime, datetime]:
if days < 1:
raise ValueError("days must be at least 1")
end = window_end or datetime.now(timezone.utc)
if end.tzinfo is None:
end = end.replace(tzinfo=timezone.utc)
else:
end = end.astimezone(timezone.utc)
start = end - timedelta(days=days)
return start, end
def _event_in_window(timestamp: str | None, *, window_start: datetime, window_end: datetime) -> bool | None:
parsed = parse_iso_datetime(timestamp)
if parsed is None:
return None
return window_start <= parsed <= window_end
def _read_jsonl(path: Path) -> tuple[list[dict[str, Any]], list[dict[str, Any]]]:
if not path.is_file():
return [], []
events: list[dict[str, Any]] = []
diagnostics: list[dict[str, Any]] = []
for line_number, raw in enumerate(path.read_text(encoding="utf-8").splitlines(), start=1):
if not raw.strip():
continue
try:
data = json.loads(raw)
except json.JSONDecodeError as exc:
diagnostics.append(
Diagnostic(
"warn",
"malformed_jsonl_line",
"JSONL line is not valid JSON.",
f"{path}:line:{line_number}",
{"error": str(exc)},
).to_dict()
)
continue
if isinstance(data, dict):
events.append(data)
return events, diagnostics
def _increment(counter: dict[str, int], key: str) -> None:
normalized = key or "unknown"
counter[normalized] = counter.get(normalized, 0) + 1
def _top_counter(counter: dict[str, int], *, limit: int) -> dict[str, int]:
return dict(sorted(counter.items(), key=lambda item: (-item[1], item[0]))[:limit])
def aggregate_ops_warden_activity(
store: MemoryStoreDescriptor,
*,
window_start: datetime,
window_end: datetime,
) -> dict[str, Any]:
events_path = store.path / "events.jsonl"
raw_events, line_diagnostics = _read_jsonl(events_path)
diagnostics = list(line_diagnostics)
by_session_kind: dict[str, int] = {}
by_command: dict[str, int] = {}
by_outcome: dict[str, int] = {}
by_route_id: dict[str, int] = {}
episode_timeline: list[dict[str, Any]] = []
first_activity_at = ""
last_activity_at = ""
in_window = 0
missing_timestamp = 0
for event in raw_events:
timestamp = str(event.get("recorded_at") or "")
in_range = _event_in_window(timestamp, window_start=window_start, window_end=window_end)
if in_range is None:
missing_timestamp += 1
diagnostics.append(
Diagnostic(
"warn",
"episode_missing_timestamp",
"Episode lacks a parseable recorded_at timestamp.",
str(event.get("event_id") or "unknown"),
).to_dict()
)
continue
if not in_range:
continue
in_window += 1
_increment(by_session_kind, str(event.get("session_kind") or ""))
_increment(by_command, str(event.get("command") or ""))
_increment(by_outcome, str(event.get("outcome") or ""))
_increment(by_route_id, str(event.get("route_id") or ""))
if not first_activity_at or timestamp < first_activity_at:
first_activity_at = timestamp
if not last_activity_at or timestamp > last_activity_at:
last_activity_at = timestamp
episode_timeline.append(
{
"event_id": str(event.get("event_id") or ""),
"recorded_at": timestamp,
"session_kind": str(event.get("session_kind") or ""),
"command": str(event.get("command") or ""),
"route_id": str(event.get("route_id") or ""),
"outcome": str(event.get("outcome") or ""),
}
)
episode_timeline.sort(key=lambda item: item.get("recorded_at", ""), reverse=True)
metadata_path = store.path / "metadata.json"
schema_version = ""
if metadata_path.is_file():
try:
schema_version = str(json.loads(metadata_path.read_text(encoding="utf-8")).get("schema_version") or "")
except json.JSONDecodeError:
pass
return {
"store_id": store.store_id,
"store_kind": store.store_kind,
"path": str(store.path),
"profile_id": store.profile_id or OPS_WARDEN_PROFILE_ID,
"schema_version": schema_version,
"episode_count": in_window,
"audit_event_count": 0,
"missing_timestamp_count": missing_timestamp,
"first_activity_at": first_activity_at,
"last_activity_at": last_activity_at,
"by_session_kind": dict(sorted(by_session_kind.items())),
"by_command": dict(sorted(by_command.items())),
"by_outcome": dict(sorted(by_outcome.items())),
"by_route_id": _top_counter(by_route_id, limit=TOP_ROUTE_IDS),
"by_operation": {},
"episode_timeline": episode_timeline[:DETAIL_EPISODE_LIMIT],
"repair_diagnostics": [],
"diagnostics": diagnostics,
}
def aggregate_local_graph_activity(
store: MemoryStoreDescriptor,
*,
window_start: datetime,
window_end: datetime,
) -> dict[str, Any]:
events_path = store.path / "events.jsonl"
audit_path = store.path / "audit.jsonl"
raw_events, event_line_diagnostics = _read_jsonl(events_path)
raw_audit, audit_line_diagnostics = _read_jsonl(audit_path)
diagnostics = list(event_line_diagnostics)
diagnostics.extend(audit_line_diagnostics)
by_session_kind: dict[str, int] = {}
by_command: dict[str, int] = {}
by_outcome: dict[str, int] = {}
by_operation: dict[str, int] = {}
episode_timeline: list[dict[str, Any]] = []
first_activity_at = ""
last_activity_at = ""
episode_count = 0
audit_count = 0
missing_timestamp = 0
for event in raw_events:
timestamp = str(event.get("timestamp") or event.get("recorded_at") or "")
in_range = _event_in_window(timestamp, window_start=window_start, window_end=window_end)
if in_range is None:
missing_timestamp += 1
diagnostics.append(
Diagnostic(
"warn",
"episode_missing_timestamp",
"Event lacks a parseable timestamp.",
str(event.get("event_id") or event.get("id") or "unknown"),
).to_dict()
)
continue
if not in_range:
continue
episode_count += 1
kind = str(event.get("kind") or event.get("session_kind") or "")
_increment(by_session_kind, kind)
_increment(by_command, str(event.get("command") or kind))
_increment(by_outcome, str(event.get("outcome") or event.get("state") or ""))
if not first_activity_at or timestamp < first_activity_at:
first_activity_at = timestamp
if not last_activity_at or timestamp > last_activity_at:
last_activity_at = timestamp
episode_timeline.append(
{
"event_id": str(event.get("event_id") or event.get("id") or ""),
"recorded_at": timestamp,
"session_kind": kind,
"command": str(event.get("command") or kind),
"route_id": "",
"outcome": str(event.get("outcome") or event.get("state") or ""),
}
)
for event in raw_audit:
timestamp = str(event.get("timestamp") or event.get("recorded_at") or "")
in_range = _event_in_window(timestamp, window_start=window_start, window_end=window_end)
if in_range is None:
missing_timestamp += 1
diagnostics.append(
Diagnostic(
"warn",
"audit_missing_timestamp",
"Audit event lacks a parseable timestamp.",
str(event.get("operation_id") or event.get("operation") or "unknown"),
).to_dict()
)
continue
if not in_range:
continue
audit_count += 1
_increment(by_operation, str(event.get("operation") or ""))
if not first_activity_at or timestamp < first_activity_at:
first_activity_at = timestamp
if not last_activity_at or timestamp > last_activity_at:
last_activity_at = timestamp
episode_timeline.sort(key=lambda item: item.get("recorded_at", ""), reverse=True)
graph_store = FileBackedMemoryGraphStore(store.path)
repair = graph_store.repair_diagnostics()
repair_diagnostics = [
item.to_dict()
for item in repair
if item.severity in {"error", "warn"}
]
schema_version = ""
try:
schema_version = str(graph_store.metadata().get("schema_version") or "")
except (json.JSONDecodeError, OSError):
pass
return {
"store_id": store.store_id,
"store_kind": store.store_kind,
"path": str(store.path),
"profile_id": store.profile_id,
"schema_version": schema_version,
"episode_count": episode_count,
"audit_event_count": audit_count,
"missing_timestamp_count": missing_timestamp,
"first_activity_at": first_activity_at,
"last_activity_at": last_activity_at,
"by_session_kind": dict(sorted(by_session_kind.items())),
"by_command": dict(sorted(by_command.items())),
"by_outcome": dict(sorted(by_outcome.items())),
"by_route_id": {},
"by_operation": dict(sorted(by_operation.items())),
"episode_timeline": episode_timeline[:DETAIL_EPISODE_LIMIT],
"repair_diagnostics": repair_diagnostics,
"diagnostics": diagnostics,
}
def aggregate_store_activity(
store: MemoryStoreDescriptor,
*,
window_start: datetime,
window_end: datetime,
) -> dict[str, Any]:
if store.store_kind == STORE_KIND_OPS_WARDEN:
return aggregate_ops_warden_activity(store, window_start=window_start, window_end=window_end)
return aggregate_local_graph_activity(store, window_start=window_start, window_end=window_end)
def build_store_list(
environ: Mapping[str, str] | None = None,
*,
registry_path: str | Path | None = None,
extra_paths: tuple[str | Path, ...] = (),
) -> dict[str, Any]:
stores, diagnostics = discover_memory_stores(environ, registry_path=registry_path, extra_paths=extra_paths)
return {
"schema_version": STORE_LIST_SCHEMA,
"valid": not any(item.get("severity") == "error" for item in diagnostics),
"store_count": len(stores),
"stores": [
{
"store_id": store.store_id,
"path": str(store.path),
"store_kind": store.store_kind,
"profile_id": store.profile_id,
"label": store.label,
"source": store.source,
}
for store in stores
],
"diagnostics": diagnostics,
}
def build_federated_report(
*,
days: int = 7,
focus_store_id: str | None = None,
focus_store_path: str | Path | None = None,
environ: Mapping[str, str] | None = None,
registry_path: str | Path | None = None,
extra_paths: tuple[str | Path, ...] = (),
window_end: datetime | None = None,
) -> dict[str, Any]:
environ = environ or os.environ
stores, discovery_diagnostics = discover_memory_stores(
environ,
registry_path=registry_path,
extra_paths=extra_paths,
)
diagnostics = list(discovery_diagnostics)
focus_reference = str(focus_store_id or focus_store_path or "").strip()
focus_store: MemoryStoreDescriptor | None = None
if focus_reference:
focus_store = resolve_store_reference(focus_reference, stores, environ=environ)
if focus_store is None:
diagnostics.append(
Diagnostic(
"error",
"unknown_store_reference",
"Requested store was not discovered and could not be classified.",
focus_reference,
).to_dict()
)
return {
"schema_version": FEDERATED_REPORT_SCHEMA,
"valid": False,
"window_days": days,
"window_start": "",
"window_end": "",
"focus_store_id": focus_reference,
"store_count": 0,
"aggregate": {},
"stores": [],
"store_detail": None,
"diagnostics": diagnostics,
}
stores = [focus_store]
window_start, window_end_dt = compute_report_window(days=days, window_end=window_end)
summaries: list[dict[str, Any]] = []
aggregate = {
"episode_count": 0,
"audit_event_count": 0,
"active_store_count": 0,
"by_outcome": {},
"by_session_kind": {},
"by_operation": {},
}
by_outcome: dict[str, int] = {}
by_session_kind: dict[str, int] = {}
by_operation: dict[str, int] = {}
store_detail: dict[str, Any] | None = None
for store in stores:
summary = aggregate_store_activity(store, window_start=window_start, window_end=window_end_dt)
diagnostics.extend(summary.get("diagnostics", ()))
public_summary = {
key: value
for key, value in summary.items()
if key not in {"episode_timeline", "repair_diagnostics", "diagnostics"}
}
summaries.append(public_summary)
aggregate["episode_count"] += int(summary.get("episode_count") or 0)
aggregate["audit_event_count"] += int(summary.get("audit_event_count") or 0)
if int(summary.get("episode_count") or 0) or int(summary.get("audit_event_count") or 0):
aggregate["active_store_count"] += 1
for key, value in (summary.get("by_outcome") or {}).items():
by_outcome[key] = by_outcome.get(key, 0) + int(value)
for key, value in (summary.get("by_session_kind") or {}).items():
by_session_kind[key] = by_session_kind.get(key, 0) + int(value)
for key, value in (summary.get("by_operation") or {}).items():
by_operation[key] = by_operation.get(key, 0) + int(value)
if focus_store is not None and store.store_id == focus_store.store_id:
store_detail = summary
aggregate["by_outcome"] = dict(sorted(by_outcome.items()))
aggregate["by_session_kind"] = dict(sorted(by_session_kind.items()))
aggregate["by_operation"] = dict(sorted(by_operation.items()))
return {
"schema_version": FEDERATED_REPORT_SCHEMA,
"valid": not any(item.get("severity") == "error" for item in diagnostics),
"window_days": days,
"window_start": window_start.replace(microsecond=0).isoformat(),
"window_end": window_end_dt.replace(microsecond=0).isoformat(),
"focus_store_id": focus_store.store_id if focus_store else "",
"store_count": len(summaries),
"aggregate": aggregate,
"stores": summaries,
"store_detail": store_detail,
"diagnostics": diagnostics,
"generated_at": utc_now_iso(),
}

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},
)

View file

@ -0,0 +1,33 @@
{
"aggregate": {
"active_store_count": 2,
"audit_event_count": 1,
"by_operation": {
"audit.query": 1
},
"by_outcome": {
"resolved": 1,
"skipped": 1,
"unknown": 1
},
"by_session_kind": {
"lifecycle.apply": 1,
"warden.agent.codex": 1,
"warden.operator": 1
},
"episode_count": 3
},
"store_count": 2,
"stores": [
{
"audit_event_count": 1,
"episode_count": 1,
"store_kind": "local_graph"
},
{
"audit_event_count": 0,
"episode_count": 2,
"store_kind": "ops_warden_coordination"
}
]
}

View file

@ -0,0 +1,2 @@
{"operation":"package.compile","operation_id":"op:audit-old","timestamp":"2026-06-02T08:00:00+00:00"}
{"operation":"audit.query","operation_id":"op:audit-in-window","timestamp":"2026-07-02T10:00:00+00:00"}

View file

@ -0,0 +1,2 @@
{"id":"event:old","kind":"path.created","schema_version":"markitect.memory.event.v1","timestamp":"2026-06-01T08:00:00+00:00"}
{"id":"event:in-window","kind":"lifecycle.apply","schema_version":"markitect.memory.event.v1","timestamp":"2026-07-02T09:00:00+00:00"}

View file

@ -0,0 +1,3 @@
{
"schema_version": "phase_memory.local_store.v1"
}

View file

@ -0,0 +1,4 @@
{"agent_id":"grok","command":"route find","diagnostic_codes":[],"event_id":"ops-warden-event:old","need_fingerprint":"abc123","outcome":"resolved","recorded_at":"2026-06-01T10:00:00+00:00","route_id":"openrouter-llm-connect","schema_version":"phase_memory.ops_warden.session_event.v1","session_kind":"warden.agent.grok"}
{"agent_id":"","command":"route find","diagnostic_codes":[],"event_id":"ops-warden-event:in-window-1","need_fingerprint":"def456","outcome":"resolved","recorded_at":"2026-07-01T10:00:00+00:00","route_id":"issue-core-ingestion-api-key","schema_version":"phase_memory.ops_warden.session_event.v1","session_kind":"warden.operator"}
{"agent_id":"codex","command":"access","diagnostic_codes":[],"event_id":"ops-warden-event:in-window-2","need_fingerprint":"ghi789","outcome":"skipped","recorded_at":"2026-07-02T11:00:00+00:00","route_id":"openrouter-llm-connect","schema_version":"phase_memory.ops_warden.session_event.v1","session_kind":"warden.agent.codex"}
{"agent_id":"","command":"memory.activate","diagnostic_codes":[],"event_id":"ops-warden-event:no-ts","need_fingerprint":"","outcome":"resolved","route_id":"","schema_version":"phase_memory.ops_warden.session_event.v1","session_kind":"warden.operator"}

View file

@ -0,0 +1,5 @@
{
"created_at": "2026-07-01T10:00:00+00:00",
"profile_id": "ops-warden-coordination",
"schema_version": "phase_memory.ops_warden.runtime.v1"
}

13
tests/fixtures/management/registry.json vendored Normal file
View file

@ -0,0 +1,13 @@
{
"schema_version": "phase_memory.management.store_registry.v1",
"stores": [
{
"store_id": "fixture-local-graph",
"path": "local-store",
"store_kind": "local_graph",
"profile_id": "local-dev",
"label": "Fixture local graph store",
"registered_at": "2026-07-01T10:00:00+00:00"
}
]
}

View file

@ -0,0 +1,46 @@
{
"schema_version": "markitect.memory.profile.v1",
"id": "ops-warden-coordination",
"title": "Ops-Warden Coordination Memory",
"intent": "Retain routing, coordination, escalation, and pilot-feedback episodes across worker, agent session, and operator CLI runtimes.",
"memory_kinds": ["reasoning", "conversation", "knowledge", "package"],
"stores": {
"reasoning": "local-graph-store",
"conversation": "local-event-log",
"knowledge": "local-graph-store",
"package": "markitect-context-package"
},
"limits": {
"reasoning": {"max_nodes": 200},
"conversation": {"max_nodes": 100},
"package": {"max_items": 8}
},
"retention": {
"conversation": {"stale_after_days": 14, "delete_after_days": 90}
},
"refresh": {
"trigger": "routing-catalog-or-playbook-change"
},
"compaction": {
"strategy": "summarize-coordination-thread-after-review"
},
"activation": {
"max_items": 8,
"max_tokens": 2000
},
"policy": {
"required_labels": ["coordination-local"],
"durable_writes": "review-gated",
"secrets_allowed": false
},
"observability": {
"emit_events": true
},
"failure": {
"missing_runtime_store": "degrade-to-dry-run"
},
"metadata": {
"rigid_import_only": ["charter", "action_allowlist"],
"runtime_modes": ["warden.worker", "warden.operator", "warden.agent.*"]
}
}

View file

@ -0,0 +1,31 @@
{
"schema_version": "phase_memory.evaluation.scenarios.v1",
"scenarios": [
{
"id": "ops-warden-routing-repeat",
"events": [
{"session_kind": "warden.agent.grok", "command": "route find", "need_fingerprint": "abc123", "route_id": "openbao-api-key", "outcome": "resolved"},
{"session_kind": "warden.agent.grok", "command": "route find", "need_fingerprint": "abc123", "route_id": "openbao-api-key", "outcome": "resolved"},
{"session_kind": "warden.worker", "command": "worker run", "need_fingerprint": "abc123", "route_id": "openbao-api-key", "outcome": "resolved"}
],
"expect": {"stabilized": true, "continuity": true}
},
{
"id": "ops-warden-cross-runtime-continuity",
"events": [
{"session_kind": "warden.agent.claude", "command": "route find", "outcome": "resolved"},
{"session_kind": "warden.worker", "command": "worker run", "outcome": "escalated"}
],
"expect": {"continuity": true}
},
{
"id": "ops-warden-llm-avoidance",
"events": [
{"session_kind": "warden.operator", "command": "route find", "need_fingerprint": "def456", "route_id": "ops-bridge-tunnel", "outcome": "resolved"},
{"session_kind": "warden.operator", "command": "route find", "need_fingerprint": "def456", "route_id": "ops-bridge-tunnel", "outcome": "resolved"},
{"session_kind": "warden.worker", "command": "memory.activate", "outcome": "resolved", "metadata": {"llm_calls_avoided": true}}
],
"expect": {"llm_calls_avoided": true}
}
]
}

View file

@ -17,6 +17,7 @@
"EVALUATION_TREND_REGRESSION_GATE_SCHEMA",
"EVALUATION_TREND_SCHEMA",
"ExternalAdapterPack",
"FEDERATED_REPORT_SCHEMA",
"FakeExternalEventLog",
"FakeExternalGraphStore",
"FakeExternalPolicyGateway",
@ -24,6 +25,7 @@
"FakeKontextualRuntimeRegistry",
"FakeMarkitectPackageCompiler",
"FakeTelemetryAuditSink",
"KNOWN_AGENT_IDS",
"LIVE_PILOT_REPORT_SCHEMA",
"LifecycleAction",
"LifecycleActionKind",
@ -52,6 +54,13 @@
"MemoryPath",
"MemoryPathState",
"MemoryPhase",
"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",
"OptionalMarkitectValidator",
"PHASE_MEMORY_CREDENTIAL_NEEDS",
"POLICY_OPERATION_POINTS",
@ -68,6 +77,8 @@
"RuntimeConfig",
"SERVICE_APP_SCHEMA",
"SERVICE_BINDING_SCHEMA",
"STORE_LIST_SCHEMA",
"STORE_REGISTRY_SCHEMA",
"ServiceAppConfig",
"ServiceBinding",
"ServiceResponse",
@ -77,10 +88,15 @@
"WARDEN_ROUTE_FIND_QUERY",
"WordCountTokenEstimator",
"abandon_path",
"activate_ops_warden_memory",
"activation_quality_report",
"adapter_pack_manifest",
"branch_path",
"build_federated_report",
"build_service_binding",
"build_session_event",
"build_store_list",
"classify_store",
"compact_path",
"create_path",
"create_wsgi_app",
@ -88,6 +104,8 @@
"credentialed_drill_config_from_env",
"credentialed_operator_report",
"credentialed_telemetry_retention_drill",
"default_memory_store_path",
"discover_memory_stores",
"evaluation_threshold_report",
"evaluation_trend_artifact",
"evaluation_trend_history",
@ -102,9 +120,14 @@
"make_review_record",
"managed_deployment_manifest",
"managed_deployment_pilot_report",
"memory_enabled",
"merge_path",
"missing_credentialed_adapter_env",
"operator_troubleshooting_matrix",
"ops_warden_adapter_pack",
"ops_warden_coordination_profile",
"ops_warden_evaluation_metrics",
"ops_warden_evaluation_report",
"package_request_from_selection",
"package_response_envelope",
"path_event",
@ -119,17 +142,23 @@
"plan_retention",
"plan_retention_from_rules",
"profile_from_markitect",
"record_session_event",
"resolve_credentialed_environ",
"resolve_runtime_adapters",
"resolve_session_kind",
"retrieve_graph_neighborhood",
"runtime_from_config",
"select_event_path",
"service_app_metadata",
"service_binding_from_config",
"service_contracts",
"session_kind_for_agent",
"stabilized_route_match",
"validate_adapter_pack_manifest",
"validate_managed_deployment_manifest",
"validate_memory_write",
"validate_operator_troubleshooting_matrix",
"validate_ops_warden_profile",
"warden_access_advisory",
"warden_cli_available",
"warden_credential_routing_advisory",

View file

@ -0,0 +1,139 @@
"""Tests for federated store management and reporting."""
from __future__ import annotations
import json
from datetime import datetime, timezone
from pathlib import Path
import pytest
from phase_memory.cli import main
from phase_memory.management import (
FEDERATED_REPORT_SCHEMA,
STORE_KIND_LOCAL_GRAPH,
STORE_KIND_OPS_WARDEN,
build_federated_report,
build_store_list,
classify_store,
discover_memory_stores,
resolve_store_reference,
)
FIXTURES = Path(__file__).resolve().parent / "fixtures" / "management"
WINDOW_END = datetime(2026, 7, 3, 12, 0, tzinfo=timezone.utc)
def _fixture_environ(tmp_path: Path) -> dict[str, str]:
registry = FIXTURES / "registry.json"
ops_store = FIXTURES / "ops-warden-store"
local_store = FIXTURES / "local-store"
return {
"WARDEN_MEMORY_STORE": str(ops_store),
"PHASE_MEMORY_REGISTRY": str(registry),
"PHASE_MEMORY_STORE_PATHS": str(local_store),
"HOME": str(tmp_path),
"XDG_DATA_HOME": str(tmp_path / "xdg"),
}
def test_classify_store_kinds() -> None:
assert classify_store(FIXTURES / "ops-warden-store") == STORE_KIND_OPS_WARDEN
assert classify_store(FIXTURES / "local-store") == STORE_KIND_LOCAL_GRAPH
assert classify_store(FIXTURES / "registry.json") is None
def test_discover_memory_stores_deduplicates_and_merges_registry(tmp_path: Path) -> None:
environ = _fixture_environ(tmp_path)
stores, diagnostics = discover_memory_stores(environ)
assert not any(item.get("severity") == "error" for item in diagnostics)
store_ids = {store.store_id for store in stores}
assert "fixture-local-graph" in store_ids
assert len({str(store.path) for store in stores}) == len(stores)
def test_build_federated_report_aggregate_and_detail(tmp_path: Path) -> None:
environ = _fixture_environ(tmp_path)
report = build_federated_report(days=7, environ=environ, window_end=WINDOW_END)
assert report["schema_version"] == FEDERATED_REPORT_SCHEMA
assert report["valid"] is True
assert report["store_count"] == 2
aggregate = report["aggregate"]
assert aggregate["episode_count"] == 3
assert aggregate["audit_event_count"] == 1
assert aggregate["active_store_count"] == 2
assert aggregate["by_outcome"]["resolved"] == 1
assert aggregate["by_outcome"]["skipped"] == 1
assert aggregate["by_session_kind"]["warden.operator"] == 1
assert aggregate["by_operation"]["audit.query"] == 1
detail = build_federated_report(
days=7,
focus_store_id="fixture-local-graph",
environ=environ,
window_end=WINDOW_END,
)
assert detail["focus_store_id"] == "fixture-local-graph"
assert detail["store_count"] == 1
assert detail["store_detail"]["episode_count"] == 1
assert detail["store_detail"]["audit_event_count"] == 1
def test_resolve_store_reference_by_id_and_path(tmp_path: Path) -> None:
environ = _fixture_environ(tmp_path)
stores, _ = discover_memory_stores(environ)
by_id = resolve_store_reference("fixture-local-graph", stores, environ=environ)
assert by_id is not None
assert by_id.store_kind == STORE_KIND_LOCAL_GRAPH
by_path = resolve_store_reference(str(FIXTURES / "ops-warden-store"), stores, environ=environ)
assert by_path is not None
assert by_path.store_kind == STORE_KIND_OPS_WARDEN
def test_federated_report_aggregate_fixture(tmp_path: Path) -> None:
environ = _fixture_environ(tmp_path)
report = build_federated_report(days=7, environ=environ, window_end=WINDOW_END)
payload = {
"aggregate": report["aggregate"],
"store_count": report["store_count"],
"stores": [
{key: store[key] for key in ("store_id", "store_kind", "episode_count", "audit_event_count")}
for store in report["stores"]
],
}
expected = json.loads((FIXTURES / "federated-report-aggregate.json").read_text(encoding="utf-8"))
assert payload["aggregate"] == expected["aggregate"]
assert payload["store_count"] == expected["store_count"]
actual_stores = sorted(payload["stores"], key=lambda item: item["store_kind"])
expected_stores = sorted(expected["stores"], key=lambda item: item["store_kind"])
for actual, item in zip(actual_stores, expected_stores, strict=True):
assert actual["store_kind"] == item["store_kind"]
assert actual["episode_count"] == item["episode_count"]
assert actual["audit_event_count"] == item["audit_event_count"]
def test_cli_stores_list_and_report(tmp_path: Path, capsys, monkeypatch: pytest.MonkeyPatch) -> None:
environ = _fixture_environ(tmp_path)
monkeypatch.setenv("WARDEN_MEMORY_STORE", environ["WARDEN_MEMORY_STORE"])
monkeypatch.setenv("PHASE_MEMORY_REGISTRY", environ["PHASE_MEMORY_REGISTRY"])
monkeypatch.setenv("PHASE_MEMORY_STORE_PATHS", environ["PHASE_MEMORY_STORE_PATHS"])
assert main(["stores", "list", "--format", "summary"]) == 0
listed = capsys.readouterr().out
assert "Fixture local graph store" in listed
assert main(["report", "--days", "7", "--format", "summary"]) == 0
reported = capsys.readouterr().out
assert "episodes=" in reported
assert "active_stores=" in reported
def test_build_store_list_from_registry_only(tmp_path: Path) -> None:
environ = {
"WARDEN_MEMORY_STORE": str(tmp_path / "missing-ops"),
"PHASE_MEMORY_REGISTRY": str(FIXTURES / "registry.json"),
"PHASE_MEMORY_STORE_PATHS": str(FIXTURES / "local-store"),
}
listing = build_store_list(environ=environ)
assert listing["valid"] is True
assert listing["store_count"] == 1
assert listing["stores"][0]["store_id"] == "fixture-local-graph"

View file

@ -0,0 +1,150 @@
import json
from pathlib import Path
from phase_memory.ops_warden import (
OPS_WARDEN_ACTIVATION_SCHEMA,
OPS_WARDEN_MEMORY_STATUS_SCHEMA,
OPS_WARDEN_PROFILE_ID,
OPS_WARDEN_RUNTIME_SCHEMA,
OPS_WARDEN_SESSION_EVENT_SCHEMA,
activate_ops_warden_memory,
build_session_event,
memory_enabled,
ops_warden_adapter_pack,
ops_warden_coordination_profile,
ops_warden_evaluation_metrics,
record_session_event,
resolve_session_kind,
session_kind_for_agent,
stabilized_route_match,
validate_memory_write,
validate_ops_warden_profile,
)
from phase_memory.external_adapters import validate_adapter_pack_manifest
FIXTURES = Path(__file__).parent / "fixtures"
def test_ops_warden_profile_validates_with_activation_budget() -> None:
profile = ops_warden_coordination_profile(FIXTURES / "ops-warden-coordination-profile.json")
validation = validate_ops_warden_profile(profile)
assert profile["id"] == OPS_WARDEN_PROFILE_ID
assert profile["policy"]["secrets_allowed"] is False
assert validation["valid"] is True
def test_validate_memory_write_rejects_secret_like_payload() -> None:
ok, diagnostics = validate_memory_write(
{
"command": "access",
"outcome": "resolved",
"metadata": {"token": "super-secret-token"},
}
)
assert ok is False
assert diagnostics[0].code == "ops_warden_memory_secret_field_rejected"
def test_session_kind_supports_known_and_future_agents() -> None:
assert session_kind_for_agent("grok") == "warden.agent.grok"
assert session_kind_for_agent(None) == "warden.operator"
assert resolve_session_kind({"WARDEN_AGENT_ID": "claude"}) == "warden.agent.claude"
assert resolve_session_kind({"WARDEN_SESSION_KIND": "warden.worker"}) == "warden.worker"
def test_record_and_activate_ops_warden_memory_round_trip(tmp_path) -> None:
store = tmp_path / "memory"
record_session_event(
store,
build_session_event(
command="route find",
session_kind="warden.agent.grok",
outcome="resolved",
need="openrouter api key",
route_id="openrouter-llm-connect",
agent_id="grok",
session_id="sess-1",
),
)
record_session_event(
store,
build_session_event(
command="route find",
session_kind="warden.agent.grok",
outcome="resolved",
need="openrouter api key",
route_id="openrouter-llm-connect",
agent_id="grok",
session_id="sess-1",
),
)
activation = activate_ops_warden_memory(
store,
session_kind="warden.worker",
need="openrouter api key",
session_id="worker-1",
)
assert activation["schema_version"] == OPS_WARDEN_ACTIVATION_SCHEMA
assert activation["stabilized_route"]["route_id"] == "openrouter-llm-connect"
assert activation["llm_calls_avoided"] is True
def test_cross_runtime_continuity_is_visible_to_worker_activation(tmp_path) -> None:
store = tmp_path / "memory"
record_session_event(
store,
build_session_event(
command="route find",
session_kind="warden.agent.codex",
outcome="resolved",
need="npm token",
route_id="openbao-api-key",
agent_id="codex",
),
)
activation = activate_ops_warden_memory(store, session_kind="warden.worker", need="npm token")
kinds = {item.get("session_kind") for item in activation["selected_episodes"] if item.get("kind") == "episode"}
assert "warden.agent.codex" in kinds
def test_ops_warden_evaluation_metrics_and_adapter_pack() -> None:
scenarios = json.loads((FIXTURES / "ops-warden-evaluation-scenarios.json").read_text(encoding="utf-8"))
events = scenarios["scenarios"][0]["events"]
metrics = ops_warden_evaluation_metrics(events)
pack = ops_warden_adapter_pack()
diagnostics = validate_adapter_pack_manifest(pack)
assert metrics["cross_runtime_continuity"] == 1.0
assert metrics["routing_repeat_accuracy"] > 0
assert not any(item.severity == "error" for item in diagnostics)
assert pack.name == "ops-warden-coordination"
def test_stabilized_route_match_requires_confirmations() -> None:
events = [
{"need_fingerprint": "abc", "route_id": "openbao-api-key", "outcome": "resolved"},
{"need_fingerprint": "abc", "route_id": "openbao-api-key", "outcome": "resolved"},
]
assert stabilized_route_match(events, need_fingerprint_value="abc")["route_id"] == "openbao-api-key"
assert stabilized_route_match(events[:1], need_fingerprint_value="abc") is None
def test_memory_enabled_honors_opt_out() -> None:
assert memory_enabled({"WARDEN_MEMORY": "1"}) is True
assert memory_enabled({"WARDEN_MEMORY": "0"}) is False
def test_ops_warden_evaluation_report_passes_fixture_scenarios() -> None:
from phase_memory.ops_warden import ops_warden_evaluation_report
scenarios = json.loads((FIXTURES / "ops-warden-evaluation-scenarios.json").read_text(encoding="utf-8"))
report = ops_warden_evaluation_report(scenarios)
assert report["valid"] is True
assert report["metrics"]["scenario_count"] == 3

View file

@ -0,0 +1,236 @@
---
id: PMEM-WP-0016
type: workplan
title: "Ops-Warden Cross-Runtime Memory Profile And Contracts"
domain: communication
repo: phase-memory
status: finished
owner: codex
topic_slug: phase-memory
created: "2026-07-02"
updated: "2026-07-02"
state_hub_workstream_id: "0f817f83-bb43-4e60-b3a7-b7cca0e9e6ed"
---
# PMEM-WP-0016: Ops-Warden Cross-Runtime Memory Profile And Contracts
## Goal
Define the phase-memory profile, event schemas, activation packs, and evaluation
gates that let **ops-warden learn from experience** across every runtime surface:
- **Direct ops-warden use** — `warden worker` with llm-connect / OpenRouter
inference, scheduled ticks, and operator CLI (`route`, `access`, `sign`).
- **Coding agent sessions** — Claude Code, Codex, Grok, and future agents that
invoke warden commands during an interactive session.
Memory must be **present** (discoverable, activatable, writable) in all three
modes without duplicating stores or leaking secrets.
## Current Evidence
`PMEM-WP-0015` added credential-safe pilot artifacts, ops-warden routing
advisories, evaluation trend regression gates, and redacted operator reports.
`WARDEN-WP-0020` shipped `warden worker` with `RuleBrain` and `LlmConnectBrain`,
fixed guardrails, and metadata-only audit via `warden activity`.
The gap: no shared memory substrate connects worker ticks, agent sessions, and
operator CLI usage. Each invocation starts cold.
## Non-Goals
- Store secret values, tokens, endpoint URLs, or prompt bodies with credentials.
- Override ops-warden's fixed charter, action allowlist, or no-secret invariant
through learned memory.
- Own llm-connect, OpenRouter configuration, or ops-warden worker scheduling.
- Replace State Hub as the coordination transport.
## Cross-Runtime Model
```mermaid
flowchart LR
subgraph surfaces [Runtime surfaces]
Worker[warden worker tick]
Agent[Agent session<br/>claude / codex / grok]
Operator[Operator CLI]
end
subgraph contract [Shared contract]
Store[(warden memory store)]
Profile[ops-warden profile]
Activate[activation plan]
end
Worker --> Store
Agent --> Store
Operator --> Store
Store --> Profile
Profile --> Activate
```
| Runtime mode | `session_kind` | Typical trigger |
| --- | --- | --- |
| `worker` | `warden.worker` | Scheduled tick, `warden worker run` |
| `agent_session` | `warden.agent.<agent_id>` | Coding agent invokes `warden route` / `warden access` |
| `operator` | `warden.operator` | Human shell without agent wrapper |
All modes read/write the **same canonical store** (path owned by ops-warden,
schema owned by phase-memory). Modes differ only in `session_kind` metadata and
activation budget — not in storage location.
## T01 - Define ops-warden memory profile
```task
id: PMEM-WP-0016-T01
status: done
priority: high
state_hub_task_id: "53dbc3f0-d2cb-4e0a-b558-64166ab976f3"
```
Author the `ops-warden-coordination` memory profile covering routing,
coordination, escalation, and pilot-feedback memory kinds.
Acceptance:
- Profile declares memory kinds, retention, compaction, activation budgets, and
policy (`secrets_allowed: false`, `durable_writes: review-gated`).
- Rigid memory kinds are documented as **import-only** (charter, allowlist) and
never promoted from learned episodes.
- Profile validates through existing `profile_from_markitect` / runtime planning
paths with deterministic diagnostics.
- Fixture lives at `tests/fixtures/ops-warden-coordination-profile.json`.
## T02 - Cross-runtime memory contract
```task
id: PMEM-WP-0016-T02
status: done
priority: high
state_hub_task_id: "6933918e-3ae0-4e0b-aade-1fbcf88ef0a1"
```
Define the `phase_memory.ops_warden.runtime.v1` contract: canonical store path
resolution, `session_kind` enum, read/write envelopes, and redaction rules.
Acceptance:
- Contract specifies canonical store default (XDG-based, overridable via
`WARDEN_MEMORY_STORE`) without embedding secrets.
- `session_kind` covers `warden.worker`, `warden.agent.*`, and `warden.operator`.
- Write envelopes reject payloads containing secret-field patterns (tokens, raw
URLs with credentials, key material).
- Contract is documented in `docs/ops-warden-memory-contract.md` and covered by
unit tests.
## T03 - Agent session event schema
```task
id: PMEM-WP-0016-T03
status: done
priority: high
state_hub_task_id: "9d436ccb-11df-49f1-82bd-eb75b1834018"
```
Define metadata-only event schemas for coding agent sessions (Claude, Codex,
Grok, extensible `agent_id`).
Acceptance:
- Events capture: `agent_id`, `session_id`, `command` (`route find`, `access`,
`worker run`, etc.), `need_fingerprint`, `route_id`, `outcome`
(`resolved|escalated|skipped`), and `diagnostic_codes`.
- Schema supports unknown future `agent_id` values without code changes.
- No event field may contain secret values or full credential payloads.
- Conformance fixtures cover at least claude, codex, and grok agent ids.
## T04 - Activation pack for routing memory
```task
id: PMEM-WP-0016-T04
status: done
priority: high
state_hub_task_id: "2f1f65a3-9a82-4854-83bb-20ef5eb3bac8"
```
Build an ops-warden activation pack that retrieves relevant coordination
episodes before inference or routing decisions.
Acceptance:
- Pack retrieves stabilized routing patterns, recent fluid thread summaries, and
applicable troubleshooting rows within profile token/item budgets.
- Activation output is a bounded context package suitable for llm-connect
injection and for agent session orientation.
- Pack works without network access (local store only).
- Tests prove deterministic selection for fixture graphs with duplicate need
fingerprints.
## T05 - Evaluation scenarios and regression gate
```task
id: PMEM-WP-0016-T05
status: done
priority: medium
state_hub_task_id: "ba0d35b3-2de1-483f-9761-cec5cbe3e8da"
```
Add evaluation scenarios and trend metrics for ops-warden memory quality across
runtime modes.
Acceptance:
- Scenarios cover: routing repeat accuracy, escalation precision, cross-runtime
continuity (agent session write → worker read), and OpenRouter call avoidance
when stabilized memory matches.
- Metrics integrate with `evaluation_trend_regression_gate`.
- Scenario fixture at `tests/fixtures/ops-warden-evaluation-scenarios.json`.
- Threshold report runs in default CI without live OpenRouter access.
## T06 - Ops-warden adapter pack conformance
```task
id: PMEM-WP-0016-T06
status: done
priority: medium
state_hub_task_id: "d430906b-6852-4328-8bc9-ad8c79fa8b03"
```
Publish a live-shaped ops-warden adapter pack and conformance suite for
ops-warden to consume.
Acceptance:
- Adapter pack manifest declares graph store, event log, audit sink, and
activation helpers required by `WARDEN-WP-0024`.
- `validate_adapter_pack_manifest` passes for the ops-warden pack.
- Public exports and API snapshot updated if new symbols are added.
- Pack documentation explains how ops-warden depends on phase-memory without
circular repo ownership.
## Acceptance Criteria
- ops-warden can activate phase-memory consistently from worker, agent session,
and operator CLI modes using one canonical store.
- Learned memory improves routing suggestions without weakening security
guardrails.
- Evaluation gates can block promotion when cross-runtime continuity or routing
accuracy regresses.
## Dependencies
- `WARDEN-WP-0024` (ops-warden repo) — consumes the contract and adapter pack.
- `WARDEN-WP-0020` — worker brain and guardrail foundation (finished).
- `PMEM-WP-0015` — pilot evidence and regression gate patterns (finished).
## Closure Review
Implemented cross-runtime memory contracts and activation helpers:
- `ops-warden-coordination` profile fixture with activation budgets and
`secrets_allowed: false`.
- `phase_memory.ops_warden.runtime.v1` contract with canonical store path,
`session_kind` coverage, and secret-field rejection.
- Agent session event schema supporting claude, codex, grok, and future agent ids.
- `activate_ops_warden_memory`, stabilized route matching, and ops-warden adapter
pack with evaluation scenarios and report helpers.
- Contract documentation in `docs/ops-warden-memory-contract.md`.

View file

@ -0,0 +1,109 @@
---
id: PMEM-WP-0017
type: workplan
title: "Federated Store Management CLI And Activity Reports"
domain: communication
repo: phase-memory
status: finished
owner: codex
topic_slug: phase-memory
created: "2026-07-03"
updated: "2026-07-03"
state_hub_workstream_id: "4e3fa1d9-0e9f-446f-a677-6038e153a316"
---
# PMEM-WP-0017: Federated Store Management CLI And Activity Reports
## Goal
Extend the `phase-memory` CLI into a **management interface** for every on-disk
memory store the runtime has produced — not only the ops-warden coordination
store. Operators and agents need one command to answer:
- How much memory activity happened in the last *N* days **across all stores**?
- Which store contributed what (episodes, audit events, session kinds, outcomes)?
- What does a **single store** look like in detail for the same window?
LLM interactivity stays out of scope here. When memory is "brought to life"
through ops-warden (worker, route/access, coding-agent sessions), chat already
happens in that surface. This workplan adds **inspectable reporting** only.
## Implementation Update - 2026-07-03
Shipped federated store discovery, windowed activity aggregation, and CLI
reporting:
- `src/phase_memory/management.py` — discovery, classification, aggregation,
`build_federated_report`, `build_store_list`
- `phase-memory stores list` and `phase-memory report [--days N] [--store …]`
- `make report-7d` and `make report REPORT_DAYS=N`
- Tests in `tests/test_management_cli.py` with `tests/fixtures/management/`
Validation: `make test` → 117 passed, 1 skipped.
## T01 - Store discovery and classification
```task
id: PMEM-WP-0017-T01
status: done
priority: high
state_hub_task_id: "7c0d7977-bd81-4aab-ab8d-5acfce7c9ba5"
```
Implement `phase_memory.management` with `discover_memory_stores`,
`classify_store`, and registry read path.
## T02 - Windowed activity aggregation
```task
id: PMEM-WP-0017-T02
status: done
priority: high
state_hub_task_id: "f58aec01-077f-431b-8eb4-482cbd8ef200"
```
Aggregate ops-warden episodes and local graph events/audit for a time window.
## T03 - Federated report envelope
```task
id: PMEM-WP-0017-T03
status: done
priority: high
state_hub_task_id: "e242b3b6-6c20-440e-82e7-c4e877096fe5"
```
`build_federated_report` returns `phase_memory.management.federated_report.v1`.
## T04 - CLI commands
```task
id: PMEM-WP-0017-T04
status: done
priority: high
state_hub_task_id: "a43154ca-5fb2-4791-84d7-c92dd222e681"
```
`stores list` and `report` wired in `cli.py`.
## T05 - Makefile, README, and operator runbook
```task
id: PMEM-WP-0017-T05
status: done
priority: medium
state_hub_task_id: "cdde3cfc-e1d5-40d2-a936-e54d6145bd79"
```
`make report-7d`, README management section, runbook cross-store activity.
## T06 - Tests and evaluation hook
```task
id: PMEM-WP-0017-T06
status: done
priority: medium
state_hub_task_id: "19672f3e-2b95-4bae-8d76-6ac5c6cdb446"
```
`tests/test_management_cli.py` and management fixtures; public API snapshot updated.