from __future__ import annotations
import json
import os
import subprocess
import sys
import tempfile
import unittest
from pathlib import Path
from rlm import harness as package_harness
from rlm import rlm as callable_rlm
from rlm.harness import HarnessState, get_harness_state
PYTHON_REFERENCE = {
"type": "python",
"import": "agent_skills.example",
"callable": "run",
"call_pattern": "await run(...)",
}
class HarnessStateTest(unittest.TestCase):
def test_crud_for_all_entry_kinds(self) -> None:
with tempfile.TemporaryDirectory() as temp_dir:
state = HarnessState(Path(temp_dir) / "harness_state.json")
created = {
"prompt": state.create_prompt_note(
"Prompt note",
"Prompt content",
id="prompt_entry",
path="prompt/path",
metadata={"kind": "prompt"},
),
"memory": state.create_memory(
"Memory",
"Memory content",
id="memory_entry",
path="memory/path",
metadata={"kind": "memory"},
),
"skill": state.create_skill(
"Skill",
"Skill content",
id="skill_entry",
path="skill/path",
reference=PYTHON_REFERENCE,
arguments={"target": {"type": "string", "required": True}},
metadata={"kind": "skill"},
),
"subagent": state.create_subagent(
"Subagent",
"Subagent content",
id="subagent_entry",
path="subagent/path",
metadata={"kind": "subagent"},
),
}
for kind, entry in created.items():
self.assertEqual(entry.kind, kind)
self.assertIn("content", state.get(kind, entry.id).content.lower())
self.assertIn(entry, state.list(kind))
state.update_prompt_note("prompt_entry", "Prompt note", "Prompt content updated")
state.update_memory("memory_entry", "Memory", "Memory content updated")
state.update_skill(
"skill_entry",
"Skill",
"Skill content updated",
reference=PYTHON_REFERENCE,
arguments={"target": {"type": "string", "required": True}, "mode": {"type": "string"}},
)
state.update_subagent("subagent_entry", "Subagent", "Subagent content updated")
for kind in ("prompt", "memory", "skill", "subagent"):
entry_id = f"{kind}_entry"
self.assertEqual(state.get(kind, entry_id).version, 2)
self.assertIn("updated", state.get(kind, entry_id).content)
delete_method = getattr(state, f"delete_{'prompt_note' if kind == 'prompt' else kind}")
self.assertTrue(delete_method(entry_id))
self.assertIsNone(state.get(kind, entry_id))
self.assertFalse(delete_method(entry_id))
self.assertEqual(state.list(), [])
def test_persists_entries_and_refinements(self) -> None:
with tempfile.TemporaryDirectory() as temp_dir:
state = HarnessState(Path(temp_dir) / "harness_state.json")
memory = state.create_memory(
"Prefer focused patches",
"Small harness updates are easier to validate than broad rewrites.",
path="engineering",
)
skill = state.create_skill(
"Check failures first",
"Inspect current failure evidence before editing code.",
id="failure_first",
reference=PYTHON_REFERENCE,
arguments={"failure_log": {"type": "string", "description": "Current failure evidence."}},
)
subagent = state.create_subagent(
"Reviewer",
"Review the proposed patch for regressions and missing tests.",
metadata={"max_turns": 3},
)
state.create_prompt_note("Refinement cadence", "Refine only after repeated evidence.")
event = state.record_refinement(
"skill failed twice",
["updated failure_first skill", "added reviewer subagent"],
evidence="two failed validations",
outcome="next validation passed",
)
reloaded = HarnessState(state.file_path)
self.assertEqual(reloaded.get("memory", memory.id).content, memory.content)
self.assertEqual(reloaded.get("skill", skill.id).version, 1)
self.assertEqual(reloaded.get("skill", skill.id).arguments["failure_log"]["type"], "string")
self.assertEqual(reloaded.get("subagent", subagent.id).metadata["max_turns"], 3)
self.assertEqual(reloaded.refinements[0].id, event.id)
self.assertIn("Prefer focused patches", reloaded.overview())
self.assertIn(
"Call contract: installed Python skills use await <skill_import>(...)",
reloaded.overview(),
)
overview = reloaded.overview()
self.assertIn("handle = await rlm('sub-task', model='provider/model')", overview)
self.assertIn("never the child's answer", overview)
self.assertIn("receiver_role='parent'", overview)
self.assertIn("await rlm.list_subagents()", overview)
self.assertIn("receiver_role='child'", overview)
self.assertIn("refinements: 1", reloaded.overview())
def test_load_ignores_unknown_json_keys(self) -> None:
with tempfile.TemporaryDirectory() as temp_dir:
state_path = Path(temp_dir) / "harness_state.json"
state_path.write_text(
json.dumps(
{
"schema": 1,
"entries": {
"memory": {
"known": {
"id": "mismatched",
"kind": "skill",
"title": "Known memory",
"content": "Loaded despite extra keys.",
"path": 123,
"source": None,
"version": "2",
"metadata": "not a dict",
"unexpected": True,
},
"missing_content": {
"title": "Missing content",
}
}
},
"refinements": [
{
"id": "refine_extra",
"trigger": "extra keys",
"changes": [1, "loaded"],
"ignored": "value",
},
{
"id": "refine_missing_changes",
"trigger": "missing changes",
}
],
}
),
encoding="utf-8",
)
state = HarnessState(state_path)
self.assertEqual(state.get("memory", "known").content, "Loaded despite extra keys.")
self.assertEqual(state.get("memory", "known").id, "known")
self.assertEqual(state.get("memory", "known").kind, "memory")
self.assertEqual(state.get("memory", "known").path, "general")
self.assertEqual(state.get("memory", "known").source, "agent")
self.assertIsNone(state.get("memory", "mismatched"))
self.assertEqual(state.get("memory", "known").version, 2)
self.assertEqual(state.get("memory", "known").metadata, {})
self.assertIsNone(state.get("memory", "missing_content"))
self.assertEqual(state.refinements[0].id, "refine_extra")
self.assertEqual(state.refinements[0].changes, ["1", "loaded"])
self.assertEqual(len(state.refinements), 1)
self.assertIn("1, loaded", state.overview())
updated = state.update_memory("known", "Known memory", "Updated content.")
self.assertEqual(updated.version, 3)
def test_skill_arguments_are_first_class(self) -> None:
with tempfile.TemporaryDirectory() as temp_dir:
state = HarnessState(Path(temp_dir) / "harness_state.json")
created = state.create_skill(
"Edit file",
"Apply a targeted edit.",
id="edit_file",
reference={
"type": "python",
"import": "agent_skills.file_edit",
"callable": "file_edit",
"call_pattern": "await file_edit(path=..., find=..., replace=...)",
},
arguments={
"path": {"type": "string", "required": True},
"find": {"type": "string", "required": True},
"replace": {"type": "string", "required": True},
},
)
updated = state.update_skill(
"edit_file",
"Edit file",
"Apply a targeted edit after reading context.",
reference={
"type": "python",
"import": "agent_skills.file_edit",
"callable": "file_edit",
"call_pattern": "await file_edit(path=..., find=..., replace=...)",
},
arguments={
"path": {"type": "string", "required": True},
"find": {"type": "string", "required": True},
"replace": {"type": "string", "required": True},
"validate": {"type": "boolean", "default": True},
},
)
reloaded = HarnessState(state.file_path)
self.assertEqual(created.arguments["path"]["required"], True)
self.assertEqual(created.reference["type"], "python")
self.assertEqual(updated.version, 2)
self.assertEqual(reloaded.get("skill", "edit_file").arguments["validate"]["default"], True)
self.assertEqual(reloaded.get("skill", "edit_file").reference["import"], "agent_skills.file_edit")
self.assertIn('"path"', reloaded.overview())
self.assertIn("agent_skills", reloaded.overview())
def test_skill_references_must_be_python(self) -> None:
with tempfile.TemporaryDirectory() as temp_dir:
state = HarnessState(Path(temp_dir) / "harness_state.json")
with self.assertRaisesRegex(ValueError, "Python reference"):
state.create_skill("No reference", "missing", arguments={})
with self.assertRaisesRegex(ValueError, "reference.type must be 'python'"):
state.create_skill(
"Shell reference",
"bad",
reference={"type": "shell", "command": "edit"},
arguments={},
)
with self.assertRaisesRegex(ValueError, "Python import"):
state.create_skill("No import", "bad", reference={"type": "python", "callable": "run"}, arguments={})
with self.assertRaisesRegex(ValueError, "callable or call_pattern"):
state.create_skill(
"No callable",
"bad",
reference={"type": "python", "import": "agent_skills.bad"},
arguments={},
)
def test_load_tolerates_corrupt_or_non_object_state(self) -> None:
for payload in ("not json at all", "null", "[]", '"a string"', "123"):
with tempfile.TemporaryDirectory() as temp_dir:
state_path = Path(temp_dir) / "harness_state.json"
state_path.write_text(payload, encoding="utf-8")
state = HarnessState(state_path)
self.assertEqual(state.list(), [])
self.assertEqual(state.refinements, [])
# The store must remain usable and self-heal on the next write.
created = state.create_memory("Recovered", "Works after corruption.", id="recovered")
self.assertEqual(HarnessState(state_path).get("memory", "recovered").content, created.content)
def test_update_skill_preserves_omitted_arguments(self) -> None:
with tempfile.TemporaryDirectory() as temp_dir:
state = HarnessState(Path(temp_dir) / "harness_state.json")
state.create_skill(
"Edit file",
"Apply an edit.",
id="edit_file",
reference=PYTHON_REFERENCE,
arguments={"path": {"type": "string", "required": True}},
)
# Updating only title/content (arguments omitted) must keep the contract.
state.update_skill("edit_file", "Edit file", "Apply an edit carefully.", reference=PYTHON_REFERENCE)
self.assertEqual(state.get("skill", "edit_file").arguments, {"path": {"type": "string", "required": True}})
# An explicit empty dict still clears it.
state.update_skill("edit_file", "Edit file", "Now argument-free.", reference=PYTHON_REFERENCE, arguments={})
self.assertEqual(state.get("skill", "edit_file").arguments, {})
def test_update_skill_without_reference_preserves_contract(self) -> None:
with tempfile.TemporaryDirectory() as temp_dir:
state = HarnessState(Path(temp_dir) / "harness_state.json")
state.create_skill(
"Edit file",
"Apply an edit.",
id="edit_file",
reference=PYTHON_REFERENCE,
arguments={"path": {"type": "string", "required": True}},
)
# A title/content-only update must not require re-sending the reference,
# and must preserve the existing reference and arguments.
updated = state.update_skill("edit_file", "Edit file", "Apply an edit carefully.")
self.assertEqual(updated.version, 2)
self.assertEqual(updated.reference, PYTHON_REFERENCE)
self.assertEqual(updated.arguments, {"path": {"type": "string", "required": True}})
self.assertEqual(updated.content, "Apply an edit carefully.")
def test_update_preserves_omitted_path(self) -> None:
with tempfile.TemporaryDirectory() as temp_dir:
state = HarnessState(Path(temp_dir) / "harness_state.json")
state.create_memory("Grouped", "content", id="grouped", path="repo/testing")
# Updating without a path keeps the custom grouping path.
state.update_memory("grouped", "Grouped", "new content")
self.assertEqual(state.get("memory", "grouped").path, "repo/testing")
# An explicit path still moves it.
state.update_memory("grouped", "Grouped", "newer", path="repo/other")
self.assertEqual(state.get("memory", "grouped").path, "repo/other")
def test_in_memory_state_never_touches_disk(self) -> None:
previous = os.environ.get("RLM_HARNESS_STATE_DIR")
previous_global = os.environ.get("RLM_GLOBAL_HARNESS_STATE_DIR")
with tempfile.TemporaryDirectory() as temp_dir:
os.environ["RLM_HARNESS_STATE_DIR"] = temp_dir
os.environ.pop("RLM_GLOBAL_HARNESS_STATE_DIR", None)
try:
state = HarnessState(in_memory=True)
created = state.create_memory("Volatile", "in memory only", id="volatile")
state.record_refinement("trigger", ["change"])
self.assertIsNone(state.file_path)
self.assertEqual(created.content, "in memory only")
self.assertEqual(state.get("memory", "volatile").content, "in memory only")
# Local in-memory operations do not resolve or persist a path.
self.assertEqual(list(Path(temp_dir).iterdir()), [])
finally:
if previous is None:
os.environ.pop("RLM_HARNESS_STATE_DIR", None)
else:
os.environ["RLM_HARNESS_STATE_DIR"] = previous
if previous_global is None:
os.environ.pop("RLM_GLOBAL_HARNESS_STATE_DIR", None)
else:
os.environ["RLM_GLOBAL_HARNESS_STATE_DIR"] = previous_global
def test_in_memory_state_global_flag_uses_global_env_store(self) -> None:
previous_global = os.environ.get("RLM_GLOBAL_HARNESS_STATE_DIR")
with tempfile.TemporaryDirectory() as temp_dir:
global_dir = Path(temp_dir) / "global"
os.environ["RLM_GLOBAL_HARNESS_STATE_DIR"] = str(global_dir)
try:
state = HarnessState(in_memory=True)
global_entry = state.create_memory("Global note", "persisted", id="global_note", global_=True)
finally:
if previous_global is None:
os.environ.pop("RLM_GLOBAL_HARNESS_STATE_DIR", None)
else:
os.environ["RLM_GLOBAL_HARNESS_STATE_DIR"] = previous_global
self.assertIsNone(state.file_path)
self.assertEqual(global_entry.scope, "global")
self.assertEqual(global_entry.content, "persisted")
self.assertIsNone(state.get("memory", "global_note"))
self.assertEqual(
HarnessState(global_dir / "harness_state.json", scope="global").get("memory", "global_note").content,
"persisted",
)
def test_in_memory_state_global_flag_uses_default_global_store(self) -> None:
previous_agent_dir = os.environ.get("PRIME_AGENT_CODING_AGENT_DIR")
previous_global = os.environ.get("RLM_GLOBAL_HARNESS_STATE_DIR")
with tempfile.TemporaryDirectory() as temp_dir:
agent_dir = Path(temp_dir) / "agent"
os.environ["PRIME_AGENT_CODING_AGENT_DIR"] = str(agent_dir)
os.environ.pop("RLM_GLOBAL_HARNESS_STATE_DIR", None)
try:
state = HarnessState(in_memory=True)
global_entry = state.create_memory("Default global", "persisted", id="default_global", global_=True)
finally:
if previous_agent_dir is None:
os.environ.pop("PRIME_AGENT_CODING_AGENT_DIR", None)
else:
os.environ["PRIME_AGENT_CODING_AGENT_DIR"] = previous_agent_dir
if previous_global is None:
os.environ.pop("RLM_GLOBAL_HARNESS_STATE_DIR", None)
else:
os.environ["RLM_GLOBAL_HARNESS_STATE_DIR"] = previous_global
self.assertIsNone(state.file_path)
self.assertEqual(global_entry.scope, "global")
self.assertIsNone(state.get("memory", "default_global"))
self.assertEqual(
HarnessState(agent_dir / "harness" / "harness_state.json", scope="global")
.get("memory", "default_global")
.content,
"persisted",
)
def test_reloads_external_writes_before_mutating(self) -> None:
with tempfile.TemporaryDirectory() as temp_dir:
state_path = Path(temp_dir) / "harness_state.json"
kernel_state = HarnessState(state_path)
kernel_state.create_memory("Kernel note", "Written from the kernel.", id="kernel")
# Simulate the host /refine command rewriting the same file from another
# process. A second instance loads the current file, adds an entry, saves.
host_state = HarnessState(state_path)
host_state.create_memory("Host note", "Written by /refine.", id="host")
# Guarantee the mtime advances even on coarse-resolution filesystems.
future = state_path.stat().st_mtime + 5
os.utime(state_path, (future, future))
# A read on the long-lived kernel state must observe the host write.
self.assertEqual(kernel_state.get("memory", "host").content, "Written by /refine.")
# A mutation must merge onto the host write instead of clobbering it.
kernel_state.create_memory("Second kernel note", "Written later.", id="kernel_2")
reloaded = HarnessState(state_path)
self.assertIsNotNone(reloaded.get("memory", "kernel"))
self.assertIsNotNone(reloaded.get("memory", "host"))
self.assertIsNotNone(reloaded.get("memory", "kernel_2"))
def test_create_detects_externally_written_entry(self) -> None:
with tempfile.TemporaryDirectory() as temp_dir:
state_path = Path(temp_dir) / "harness_state.json"
state = HarnessState(state_path)
# Another process creates the same entry on disk after our last load.
other = HarnessState(state_path)
other.create_memory("External", "Written elsewhere.", id="dup")
future = state_path.stat().st_mtime + 5
os.utime(state_path, (future, future))
# create() must observe the external entry and honor create-or-fail.
with self.assertRaisesRegex(ValueError, "already exists"):
state.create_memory("Local", "Should not overwrite.", id="dup")
self.assertEqual(state.get("memory", "dup").content, "Written elsewhere.")
def test_explicit_create_and_update_enforce_entry_existence(self) -> None:
with tempfile.TemporaryDirectory() as temp_dir:
state = HarnessState(Path(temp_dir) / "harness_state.json")
first = state.create_skill("Triage", "old", id="triage", reference=PYTHON_REFERENCE, arguments={})
with self.assertRaisesRegex(ValueError, "already exists"):
state.create_skill("Triage", "duplicate", id="triage", reference=PYTHON_REFERENCE, arguments={})
with self.assertRaisesRegex(ValueError, "does not exist"):
state.update_skill("missing", "Missing", "missing", reference=PYTHON_REFERENCE, arguments={})
second = state.update_skill("triage", "Triage", "new", reference=PYTHON_REFERENCE, arguments={})
self.assertEqual(first.id, second.id)
self.assertEqual(second.content, "new")
self.assertEqual(second.version, 2)
def test_explicit_state_dir_cache_uses_harness_state_file(self) -> None:
with tempfile.TemporaryDirectory() as temp_dir:
state = get_harness_state(temp_dir)
again = get_harness_state(temp_dir)
self.assertIs(state, again)
self.assertEqual(state.file_path, Path(temp_dir).resolve() / "harness_state.json")
def test_explicit_state_dir_global_flag_uses_matching_state_file(self) -> None:
previous_global = os.environ.get("RLM_GLOBAL_HARNESS_STATE_DIR")
with tempfile.TemporaryDirectory() as temp_dir:
explicit_dir = Path(temp_dir) / "explicit"
env_global_dir = Path(temp_dir) / "env-global"
os.environ["RLM_GLOBAL_HARNESS_STATE_DIR"] = str(env_global_dir)
try:
state = get_harness_state(explicit_dir)
global_entry = state.create_memory("Scoped global", "custom dir", id="scoped_global", global_=True)
finally:
if previous_global is None:
os.environ.pop("RLM_GLOBAL_HARNESS_STATE_DIR", None)
else:
os.environ["RLM_GLOBAL_HARNESS_STATE_DIR"] = previous_global
self.assertEqual(global_entry.scope, "global")
self.assertIsNotNone(
HarnessState(explicit_dir / "harness_state.json", scope="global").get("memory", "scoped_global")
)
self.assertFalse((env_global_dir / "harness_state.json").exists())
def test_env_default_state_keeps_env_global_target_after_explicit_dir_cache_hit(self) -> None:
previous_local = os.environ.get("RLM_HARNESS_STATE_DIR")
previous_global = os.environ.get("RLM_GLOBAL_HARNESS_STATE_DIR")
with tempfile.TemporaryDirectory() as temp_dir:
local_dir = Path(temp_dir) / "local"
env_global_dir = Path(temp_dir) / "env-global"
os.environ["RLM_HARNESS_STATE_DIR"] = str(local_dir)
os.environ["RLM_GLOBAL_HARNESS_STATE_DIR"] = str(env_global_dir)
try:
cached_from_env = get_harness_state()
# An explicit state_dir that aliases the env local dir must not
# redirect the env-default singleton's global target.
cached_from_explicit = get_harness_state(local_dir)
global_entry = cached_from_env.create_memory(
"Env global",
"still targets the env global dir",
id="env_global_after_hit",
global_=True,
)
finally:
if previous_local is None:
os.environ.pop("RLM_HARNESS_STATE_DIR", None)
else:
os.environ["RLM_HARNESS_STATE_DIR"] = previous_local
if previous_global is None:
os.environ.pop("RLM_GLOBAL_HARNESS_STATE_DIR", None)
else:
os.environ["RLM_GLOBAL_HARNESS_STATE_DIR"] = previous_global
self.assertIs(cached_from_env, cached_from_explicit)
self.assertEqual(global_entry.scope, "global")
self.assertIsNotNone(
HarnessState(env_global_dir / "harness_state.json", scope="global").get(
"memory", "env_global_after_hit"
)
)
self.assertIsNone(
HarnessState(local_dir / "harness_state.json").get("memory", "env_global_after_hit")
)
def test_local_state_requires_local_path(self) -> None:
previous_local = os.environ.get("RLM_HARNESS_STATE_DIR")
previous_session = os.environ.get("RLM_SESSION_DIR")
try:
os.environ.pop("RLM_HARNESS_STATE_DIR", None)
os.environ.pop("RLM_SESSION_DIR", None)
with self.assertRaisesRegex(RuntimeError, "Local harness state requires"):
HarnessState()
finally:
if previous_local is None:
os.environ.pop("RLM_HARNESS_STATE_DIR", None)
else:
os.environ["RLM_HARNESS_STATE_DIR"] = previous_local
if previous_session is None:
os.environ.pop("RLM_SESSION_DIR", None)
else:
os.environ["RLM_SESSION_DIR"] = previous_session
def test_default_state_uses_global_harness_env_dir(self) -> None:
previous = os.environ.get("RLM_HARNESS_STATE_DIR")
with tempfile.TemporaryDirectory() as temp_dir:
os.environ["RLM_HARNESS_STATE_DIR"] = temp_dir
try:
state = HarnessState()
finally:
if previous is None:
os.environ.pop("RLM_HARNESS_STATE_DIR", None)
else:
os.environ["RLM_HARNESS_STATE_DIR"] = previous
self.assertEqual(state.file_path, Path(temp_dir).resolve() / "harness_state.json")
def test_global_scope_default_state_uses_global_harness_env_dir(self) -> None:
previous_local = os.environ.get("RLM_HARNESS_STATE_DIR")
previous_global = os.environ.get("RLM_GLOBAL_HARNESS_STATE_DIR")
with tempfile.TemporaryDirectory() as temp_dir:
local_dir = Path(temp_dir) / "local"
global_dir = Path(temp_dir) / "global"
os.environ["RLM_HARNESS_STATE_DIR"] = str(local_dir)
os.environ["RLM_GLOBAL_HARNESS_STATE_DIR"] = str(global_dir)
try:
state = HarnessState(scope="global")
finally:
if previous_local is None:
os.environ.pop("RLM_HARNESS_STATE_DIR", None)
else:
os.environ["RLM_HARNESS_STATE_DIR"] = previous_local
if previous_global is None:
os.environ.pop("RLM_GLOBAL_HARNESS_STATE_DIR", None)
else:
os.environ["RLM_GLOBAL_HARNESS_STATE_DIR"] = previous_global
self.assertEqual(state.scope, "global")
self.assertEqual(state.file_path, global_dir.resolve() / "harness_state.json")
def test_default_state_is_local_and_global_flag_targets_global_store(self) -> None:
previous_local = os.environ.get("RLM_HARNESS_STATE_DIR")
previous_global = os.environ.get("RLM_GLOBAL_HARNESS_STATE_DIR")
with tempfile.TemporaryDirectory() as temp_dir:
local_dir = Path(temp_dir) / "local"
global_dir = Path(temp_dir) / "global"
os.environ["RLM_HARNESS_STATE_DIR"] = str(local_dir)
os.environ["RLM_GLOBAL_HARNESS_STATE_DIR"] = str(global_dir)
try:
state = get_harness_state()
global_state = get_harness_state(global_=True)
local_entry = state.create_memory("Local note", "Only this session.", id="local_note")
global_entry = state.create_memory("Global note", "All sessions.", id="global_note", global_=True)
kwargs_entry = state.create_memory(
"Kwargs global note",
"All sessions via kwargs.",
id="kwargs_global_note",
**{"global": True},
)
finally:
if previous_local is None:
os.environ.pop("RLM_HARNESS_STATE_DIR", None)
else:
os.environ["RLM_HARNESS_STATE_DIR"] = previous_local
if previous_global is None:
os.environ.pop("RLM_GLOBAL_HARNESS_STATE_DIR", None)
else:
os.environ["RLM_GLOBAL_HARNESS_STATE_DIR"] = previous_global
self.assertEqual(state.file_path, local_dir.resolve() / "harness_state.json")
self.assertEqual(global_state.file_path, global_dir.resolve() / "harness_state.json")
self.assertEqual(local_entry.scope, "local")
self.assertEqual(global_entry.scope, "global")
self.assertEqual(kwargs_entry.scope, "global")
self.assertIsNotNone(HarnessState(local_dir / "harness_state.json").get("memory", "local_note"))
self.assertIsNone(HarnessState(local_dir / "harness_state.json").get("memory", "global_note"))
self.assertIsNotNone(HarnessState(global_dir / "harness_state.json", scope="global").get("memory", "global_note"))
self.assertIsNotNone(
HarnessState(global_dir / "harness_state.json", scope="global").get("memory", "kwargs_global_note")
)
def test_global_kwarg_must_be_boolean(self) -> None:
with tempfile.TemporaryDirectory() as temp_dir:
state = HarnessState(Path(temp_dir) / "harness_state.json")
with self.assertRaisesRegex(TypeError, "global must be a bool"):
state.create_memory("Bad global flag", "bad", id="bad_global", **{"global": "false"})
def test_state_cache_keeps_scope_distinct_when_local_and_global_share_a_file(self) -> None:
previous_local = os.environ.get("RLM_HARNESS_STATE_DIR")
previous_global = os.environ.get("RLM_GLOBAL_HARNESS_STATE_DIR")
with tempfile.TemporaryDirectory() as temp_dir:
os.environ["RLM_HARNESS_STATE_DIR"] = temp_dir
os.environ["RLM_GLOBAL_HARNESS_STATE_DIR"] = temp_dir
try:
state = get_harness_state()
global_state = get_harness_state(global_=True)
local_entry = state.create_memory("Local note", "Only this session.", id="local_note")
global_entry = state.create_memory("Global note", "All sessions.", id="global_note", global_=True)
finally:
if previous_local is None:
os.environ.pop("RLM_HARNESS_STATE_DIR", None)
else:
os.environ["RLM_HARNESS_STATE_DIR"] = previous_local
if previous_global is None:
os.environ.pop("RLM_GLOBAL_HARNESS_STATE_DIR", None)
else:
os.environ["RLM_GLOBAL_HARNESS_STATE_DIR"] = previous_global
self.assertIsNot(state, global_state)
self.assertEqual(state.file_path, global_state.file_path)
self.assertEqual(state.scope, "local")
self.assertEqual(global_state.scope, "global")
self.assertEqual(local_entry.scope, "local")
self.assertEqual(global_entry.scope, "global")
reloaded = HarnessState(Path(temp_dir) / "harness_state.json")
self.assertEqual(reloaded.get("memory", "local_note").scope, "local")
self.assertEqual(reloaded.get("memory", "global_note").scope, "global")
def test_scope_prefixed_ids_route_to_the_displayed_scope(self) -> None:
previous_local = os.environ.get("RLM_HARNESS_STATE_DIR")
previous_global = os.environ.get("RLM_GLOBAL_HARNESS_STATE_DIR")
with tempfile.TemporaryDirectory() as temp_dir:
local_dir = Path(temp_dir) / "local"
global_dir = Path(temp_dir) / "global"
os.environ["RLM_HARNESS_STATE_DIR"] = str(local_dir)
os.environ["RLM_GLOBAL_HARNESS_STATE_DIR"] = str(global_dir)
try:
state = get_harness_state()
state.create_memory("Global note", "v1", id="routed", global_=True)
# The overview displays [global:routed]; that id must be usable as-is
# and imply the global scope without passing global_.
updated = state.update_memory("global:routed", "Global note", "v2")
self.assertEqual(updated.scope, "global")
self.assertEqual(state.get("memory", "global:routed").content, "v2")
self.assertIsNone(state.get("memory", "routed"))
state.create_memory("Local note", "local", id="local_note")
self.assertEqual(state.get("memory", "local:local_note").content, "local")
self.assertTrue(state.delete_memory("local:local_note"))
self.assertIsNone(state.get("memory", "local_note"))
finally:
if previous_local is None:
os.environ.pop("RLM_HARNESS_STATE_DIR", None)
else:
os.environ["RLM_HARNESS_STATE_DIR"] = previous_local
if previous_global is None:
os.environ.pop("RLM_GLOBAL_HARNESS_STATE_DIR", None)
else:
os.environ["RLM_GLOBAL_HARNESS_STATE_DIR"] = previous_global
self.assertEqual(
HarnessState(global_dir / "harness_state.json", scope="global").get("memory", "routed").content,
"v2",
)
def test_create_with_prefixed_id_does_not_mint_literal_id(self) -> None:
previous_local = os.environ.get("RLM_HARNESS_STATE_DIR")
previous_global = os.environ.get("RLM_GLOBAL_HARNESS_STATE_DIR")
with tempfile.TemporaryDirectory() as temp_dir:
local_dir = Path(temp_dir) / "local"
global_dir = Path(temp_dir) / "global"
os.environ["RLM_HARNESS_STATE_DIR"] = str(local_dir)
os.environ["RLM_GLOBAL_HARNESS_STATE_DIR"] = str(global_dir)
try:
state = get_harness_state()
entry = state.create_memory("Validation", "content", id="global:validation")
finally:
if previous_local is None:
os.environ.pop("RLM_HARNESS_STATE_DIR", None)
else:
os.environ["RLM_HARNESS_STATE_DIR"] = previous_local
if previous_global is None:
os.environ.pop("RLM_GLOBAL_HARNESS_STATE_DIR", None)
else:
os.environ["RLM_GLOBAL_HARNESS_STATE_DIR"] = previous_global
self.assertEqual(entry.id, "validation")
self.assertEqual(entry.scope, "global")
global_store = HarnessState(global_dir / "harness_state.json", scope="global")
self.assertIsNotNone(global_store.get("memory", "validation"))
self.assertIsNone(global_store.get("memory", "global:validation"))
self.assertFalse((local_dir / "harness_state.json").exists())
def test_module_harness_binds_lazily_to_env_set_after_import(self) -> None:
# Forkserver scenario: rlm is imported in the template process without the
# per-session env; the child applies env after fork. rlm.harness must then
# resolve against the new env instead of a store frozen at import time.
previous_local = os.environ.get("RLM_HARNESS_STATE_DIR")
previous_session = os.environ.get("RLM_SESSION_DIR")
with tempfile.TemporaryDirectory() as temp_dir:
try:
os.environ.pop("RLM_HARNESS_STATE_DIR", None)
os.environ.pop("RLM_SESSION_DIR", None)
# Without local env, local writes fail loudly instead of vanishing.
with self.assertRaisesRegex(RuntimeError, "global_=True"):
package_harness.create_memory("Volatile", "pre-env", id="pre_env")
os.environ["RLM_HARNESS_STATE_DIR"] = temp_dir
entry = package_harness.create_memory("Session note", "persisted", id="session_note")
self.assertIsNone(package_harness.get("memory", "pre_env"))
finally:
if previous_local is None:
os.environ.pop("RLM_HARNESS_STATE_DIR", None)
else:
os.environ["RLM_HARNESS_STATE_DIR"] = previous_local
if previous_session is None:
os.environ.pop("RLM_SESSION_DIR", None)
else:
os.environ["RLM_SESSION_DIR"] = previous_session
self.assertEqual(entry.scope, "local")
reloaded = HarnessState(Path(temp_dir) / "harness_state.json")
self.assertEqual(reloaded.get("memory", "session_note").content, "persisted")
def test_module_harness_without_env_raises_on_local_writes_and_reads_work(self) -> None:
previous_local = os.environ.get("RLM_HARNESS_STATE_DIR")
previous_session = os.environ.get("RLM_SESSION_DIR")
try:
os.environ.pop("RLM_HARNESS_STATE_DIR", None)
os.environ.pop("RLM_SESSION_DIR", None)
for mutate in (
lambda: package_harness.create_memory("Lost", "content", id="lost"),
lambda: package_harness.update_memory("lost", "Lost", "content"),
lambda: package_harness.delete_memory("lost"),
lambda: package_harness.upsert("memory", "Lost", "content", id="lost"),
lambda: package_harness.record_refinement("trigger", ["change"]),
):
with self.assertRaisesRegex(RuntimeError, "Local harness state requires.*global_=True"):
mutate()
# Reads keep working against an empty view.
self.assertIsNone(package_harness.get("memory", "lost"))
self.assertEqual(package_harness.list(), [])
self.assertIn("memory: 0", package_harness.overview())
self.assertEqual(package_harness.snapshot()["refinements"], [])
finally:
if previous_local is None:
os.environ.pop("RLM_HARNESS_STATE_DIR", None)
else:
os.environ["RLM_HARNESS_STATE_DIR"] = previous_local
if previous_session is None:
os.environ.pop("RLM_SESSION_DIR", None)
else:
os.environ["RLM_SESSION_DIR"] = previous_session
def test_module_harness_without_env_still_routes_global_writes(self) -> None:
previous_local = os.environ.get("RLM_HARNESS_STATE_DIR")
previous_session = os.environ.get("RLM_SESSION_DIR")
previous_global = os.environ.get("RLM_GLOBAL_HARNESS_STATE_DIR")
with tempfile.TemporaryDirectory() as temp_dir:
global_dir = Path(temp_dir) / "global"
try:
os.environ.pop("RLM_HARNESS_STATE_DIR", None)
os.environ.pop("RLM_SESSION_DIR", None)
os.environ["RLM_GLOBAL_HARNESS_STATE_DIR"] = str(global_dir)
entry = package_harness.create_memory("Lesson", "keep me", id="no_session_lesson", global_=True)
finally:
if previous_local is None:
os.environ.pop("RLM_HARNESS_STATE_DIR", None)
else:
os.environ["RLM_HARNESS_STATE_DIR"] = previous_local
if previous_session is None:
os.environ.pop("RLM_SESSION_DIR", None)
else:
os.environ["RLM_SESSION_DIR"] = previous_session
if previous_global is None:
os.environ.pop("RLM_GLOBAL_HARNESS_STATE_DIR", None)
else:
os.environ["RLM_GLOBAL_HARNESS_STATE_DIR"] = previous_global
self.assertEqual(entry.scope, "global")
self.assertEqual(
HarnessState(global_dir / "harness_state.json", scope="global").get("memory", "no_session_lesson").content,
"keep me",
)
def test_import_rlm_without_env_does_not_raise(self) -> None:
env = dict(os.environ)
env.pop("RLM_HARNESS_STATE_DIR", None)
env.pop("RLM_SESSION_DIR", None)
env["PYTHONPATH"] = str(Path(__file__).resolve().parents[1] / "src")
result = subprocess.run(
[sys.executable, "-c", "import rlm; repr(rlm.harness); rlm.harness.overview(); rlm.harness.create_memory"],
env=env,
capture_output=True,
text=True,
)
self.assertEqual(result.returncode, 0, result.stderr)
def test_empty_local_state_dir_env_is_treated_as_unset(self) -> None:
previous_local = os.environ.get("RLM_HARNESS_STATE_DIR")
previous_session = os.environ.get("RLM_SESSION_DIR")
with tempfile.TemporaryDirectory() as temp_dir:
try:
# Empty local dir must not fall through to the global agent-dir default.
os.environ["RLM_HARNESS_STATE_DIR"] = ""
os.environ.pop("RLM_SESSION_DIR", None)
with self.assertRaisesRegex(RuntimeError, "Local harness state requires"):
HarnessState()
# With a session dir it takes the session fallback instead.
os.environ["RLM_SESSION_DIR"] = temp_dir
state = HarnessState()
self.assertEqual(state.file_path, Path(temp_dir).resolve() / "harness" / "harness_state.json")
# A whitespace-only session dir is also unset.
os.environ["RLM_SESSION_DIR"] = " "
with self.assertRaisesRegex(RuntimeError, "Local harness state requires"):
HarnessState()
finally:
if previous_local is None:
os.environ.pop("RLM_HARNESS_STATE_DIR", None)
else:
os.environ["RLM_HARNESS_STATE_DIR"] = previous_local
if previous_session is None:
os.environ.pop("RLM_SESSION_DIR", None)
else:
os.environ["RLM_SESSION_DIR"] = previous_session
def test_explicit_dir_aliasing_env_local_dir_keeps_env_global_target(self) -> None:
previous_local = os.environ.get("RLM_HARNESS_STATE_DIR")
previous_global = os.environ.get("RLM_GLOBAL_HARNESS_STATE_DIR")
with tempfile.TemporaryDirectory() as temp_dir:
local_dir = Path(temp_dir) / "local"
env_global_dir = Path(temp_dir) / "env-global"
os.environ["RLM_HARNESS_STATE_DIR"] = str(local_dir)
os.environ["RLM_GLOBAL_HARNESS_STATE_DIR"] = str(env_global_dir)
try:
# First construction happens via an explicit dir that merely aliases
# the env local dir; global writes must still hit the env global dir.
state = get_harness_state(local_dir)
global_entry = state.create_memory("Aliased", "still global", id="alias_global", global_=True)
finally:
if previous_local is None:
os.environ.pop("RLM_HARNESS_STATE_DIR", None)
else:
os.environ["RLM_HARNESS_STATE_DIR"] = previous_local
if previous_global is None:
os.environ.pop("RLM_GLOBAL_HARNESS_STATE_DIR", None)
else:
os.environ["RLM_GLOBAL_HARNESS_STATE_DIR"] = previous_global
self.assertEqual(global_entry.scope, "global")
self.assertIsNotNone(
HarnessState(env_global_dir / "harness_state.json", scope="global").get("memory", "alias_global")
)
self.assertIsNone(
HarnessState(local_dir / "harness_state.json").get("memory", "alias_global")
)
def test_callable_rlm_exposes_harness_state_helpers(self) -> None:
self.assertIs(callable_rlm.harness, package_harness)
self.assertIs(callable_rlm.get_harness_state, get_harness_state)
def test_record_refinement_accepts_single_change_string(self) -> None:
with tempfile.TemporaryDirectory() as temp_dir:
state = HarnessState(Path(temp_dir) / "harness_state.json")
event = state.record_refinement("manual cli test", "single change")
self.assertEqual(event.changes, ["single change"])
self.assertEqual(state.refinements[0].changes, ["single change"])
def test_unknown_kind_rejected(self) -> None:
with tempfile.TemporaryDirectory() as temp_dir:
state = HarnessState(Path(temp_dir) / "harness_state.json")
with self.assertRaisesRegex(ValueError, "unknown harness kind"):
state.upsert("tool", "Tool", "Tool content")
with self.assertRaisesRegex(ValueError, "unknown harness kind"):
state.get("tool", "tool")
with self.assertRaisesRegex(ValueError, "unknown harness kind"):
state.delete("tool", "tool")
with self.assertRaisesRegex(ValueError, "unknown harness kind"):
state.list("tool")
if __name__ == "__main__":
unittest.main()