generated from coulomb/repo-seed
tegwick
120de64bcb
Load coordination memory by default via ensure_memory_context on app bootstrap and route/access flows; invalidate cache after episode writes. WARDEN_MEMORY=0 remains the opt-out. Document that warden memory activate is optional only.
143 lines
5.0 KiB
Python
143 lines
5.0 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")
|
|
|
|
|
|
def test_route_find_implicitly_activates_memory_without_explicit_command(tmp_path, monkeypatch) -> None:
|
|
from warden.memory import ensure_memory_context
|
|
|
|
monkeypatch.setenv("WARDEN_MEMORY_STORE", str(tmp_path / "memory"))
|
|
monkeypatch.delenv("WARDEN_AGENT_ID", raising=False)
|
|
result = runner.invoke(app, ["route", "find", "ssh tunnel", "--json"])
|
|
|
|
assert result.exit_code == 0
|
|
activation = ensure_memory_context(need="ssh tunnel", implicit=True)
|
|
assert activation is not None
|
|
assert activation.get("implicit") is True
|
|
assert status()["episode_count"] >= 1 |