generated from coulomb/repo-seed
tegwick
04929e7981
Add phase-memory bridge, warden memory CLI, route/access/sign recording, memory-aware worker planning with OpenRouter skip, tests, wiki, and AGENTS.md orientation for Claude, Codex, Grok, and future agent sessions.
129 lines
4.4 KiB
Python
129 lines
4.4 KiB
Python
"""Tests for ops-warden phase-memory bridge (WARDEN-WP-0024)."""
|
|
|
|
from __future__ import annotations
|
|
|
|
import json
|
|
|
|
from typer.testing import CliRunner
|
|
|
|
from warden.cli import app
|
|
from warden.memory import activate, enabled, record_command_episode, status, store_path
|
|
from warden.worker import RuleBrain, _plan_with_memory, build_plans
|
|
|
|
runner = CliRunner()
|
|
|
|
|
|
def _msg(**over) -> dict:
|
|
base = {
|
|
"id": "m1",
|
|
"from_agent": "someone",
|
|
"subject": "Where do I get an npm token?",
|
|
"body": "Which subsystem owns this credential — how do I obtain it?",
|
|
}
|
|
base.update(over)
|
|
return base
|
|
|
|
|
|
def test_memory_status_and_activate_round_trip(tmp_path, monkeypatch) -> None:
|
|
monkeypatch.setenv("WARDEN_MEMORY_STORE", str(tmp_path / "memory"))
|
|
monkeypatch.setenv("WARDEN_AGENT_ID", "grok")
|
|
|
|
first = record_command_episode(
|
|
command="route find",
|
|
outcome="resolved",
|
|
need="npm token",
|
|
route_id="openbao-api-key",
|
|
)
|
|
second = record_command_episode(
|
|
command="route find",
|
|
outcome="resolved",
|
|
need="npm token",
|
|
route_id="openbao-api-key",
|
|
)
|
|
payload = activate(need="npm token", agent="grok")
|
|
|
|
assert first["valid"] is True
|
|
assert second["valid"] is True
|
|
assert payload["session_kind"] == "warden.agent.grok"
|
|
assert payload["stabilized_route"]["route_id"] == "openbao-api-key"
|
|
assert status()["episode_count"] >= 3
|
|
|
|
|
|
def test_worker_uses_stabilized_memory_without_llm(tmp_path, monkeypatch) -> None:
|
|
monkeypatch.setenv("WARDEN_MEMORY_STORE", str(tmp_path / "memory"))
|
|
monkeypatch.setenv("WARDEN_SESSION_KIND", "warden.worker")
|
|
record_command_episode(
|
|
command="route find",
|
|
outcome="resolved",
|
|
need="Where do I get an npm token?",
|
|
route_id="openbao-api-key",
|
|
)
|
|
record_command_episode(
|
|
command="route find",
|
|
outcome="resolved",
|
|
need="Where do I get an npm token?",
|
|
route_id="openbao-api-key",
|
|
)
|
|
|
|
plan = _plan_with_memory(_msg(), RuleBrain())
|
|
|
|
assert plan.actions
|
|
assert plan.actions[0].kind == "route_answer"
|
|
assert plan.actions[0].payload.get("memory_stabilized") is True
|
|
|
|
|
|
def test_cross_runtime_continuity_agent_to_worker(tmp_path, monkeypatch) -> None:
|
|
monkeypatch.setenv("WARDEN_MEMORY_STORE", str(tmp_path / "memory"))
|
|
monkeypatch.setenv("WARDEN_AGENT_ID", "codex")
|
|
record_command_episode(
|
|
command="route find",
|
|
outcome="resolved",
|
|
need="openrouter api key",
|
|
route_id="openrouter-llm-connect",
|
|
)
|
|
monkeypatch.setenv("WARDEN_SESSION_KIND", "warden.worker")
|
|
monkeypatch.delenv("WARDEN_AGENT_ID", raising=False)
|
|
activation = activate(need="openrouter api key")
|
|
kinds = {item.get("session_kind") for item in activation["selected_episodes"] if item.get("kind") == "episode"}
|
|
|
|
assert "warden.agent.codex" in kinds
|
|
|
|
|
|
def test_cli_memory_status_and_activate(tmp_path, monkeypatch) -> None:
|
|
monkeypatch.setenv("WARDEN_MEMORY_STORE", str(tmp_path / "memory"))
|
|
status_result = runner.invoke(app, ["memory", "status", "--json"])
|
|
activate_result = runner.invoke(app, ["memory", "activate", "--agent", "claude", "--json"])
|
|
|
|
assert status_result.exit_code == 0
|
|
assert activate_result.exit_code == 0
|
|
assert json.loads(status_result.stdout)["episode_count"] >= 0
|
|
assert json.loads(activate_result.stdout)["session_kind"] == "warden.agent.claude"
|
|
|
|
|
|
def test_route_find_records_memory_episode(tmp_path, monkeypatch) -> None:
|
|
monkeypatch.setenv("WARDEN_MEMORY_STORE", str(tmp_path / "memory"))
|
|
result = runner.invoke(app, ["route", "find", "openrouter api key", "--json"])
|
|
|
|
assert result.exit_code == 0
|
|
payload = status()
|
|
assert payload["episode_count"] >= 1
|
|
|
|
|
|
def test_build_plans_records_worker_outcomes(tmp_path, monkeypatch) -> None:
|
|
monkeypatch.setenv("WARDEN_MEMORY_STORE", str(tmp_path / "memory"))
|
|
plans = build_plans([_msg()], RuleBrain())
|
|
|
|
assert plans[0].actions
|
|
assert status()["episode_count"] >= 1
|
|
|
|
|
|
def test_memory_disabled_skips_recording(monkeypatch) -> None:
|
|
monkeypatch.setenv("WARDEN_MEMORY", "0")
|
|
result = record_command_episode(command="route find", outcome="resolved", need="npm token")
|
|
|
|
assert result.get("skipped") is True
|
|
|
|
|
|
def test_default_store_path_uses_xdg(monkeypatch) -> None:
|
|
monkeypatch.delenv("WARDEN_MEMORY_STORE", raising=False)
|
|
assert str(store_path()).endswith("warden/memory") |