STR-39: expand source-aware whitebox workflows and wiki memory
This commit is contained in:
28
tests/llm/test_source_aware_whitebox.py
Normal file
28
tests/llm/test_source_aware_whitebox.py
Normal file
@@ -0,0 +1,28 @@
|
||||
from strix.llm.config import LLMConfig
|
||||
from strix.llm.llm import LLM
|
||||
|
||||
|
||||
def test_llm_config_whitebox_defaults_to_false(monkeypatch) -> None:
|
||||
monkeypatch.setenv("STRIX_LLM", "openai/gpt-5")
|
||||
config = LLMConfig()
|
||||
assert config.is_whitebox is False
|
||||
|
||||
|
||||
def test_llm_config_whitebox_can_be_enabled(monkeypatch) -> None:
|
||||
monkeypatch.setenv("STRIX_LLM", "openai/gpt-5")
|
||||
config = LLMConfig(is_whitebox=True)
|
||||
assert config.is_whitebox is True
|
||||
|
||||
|
||||
def test_whitebox_prompt_loads_source_aware_coordination_skill(monkeypatch) -> None:
|
||||
monkeypatch.setenv("STRIX_LLM", "openai/gpt-5")
|
||||
|
||||
whitebox_llm = LLM(LLMConfig(scan_mode="quick", is_whitebox=True), agent_name="StrixAgent")
|
||||
assert "<source_aware_whitebox>" in whitebox_llm.system_prompt
|
||||
assert "Begin with fast source triage" in whitebox_llm.system_prompt
|
||||
assert "You MUST begin at the very first step by running the code and testing live." not in (
|
||||
whitebox_llm.system_prompt
|
||||
)
|
||||
|
||||
non_whitebox_llm = LLM(LLMConfig(scan_mode="quick", is_whitebox=False), agent_name="StrixAgent")
|
||||
assert "<source_aware_whitebox>" not in non_whitebox_llm.system_prompt
|
||||
107
tests/tools/test_agents_graph_whitebox.py
Normal file
107
tests/tools/test_agents_graph_whitebox.py
Normal file
@@ -0,0 +1,107 @@
|
||||
from types import SimpleNamespace
|
||||
|
||||
import strix.agents as agents_module
|
||||
from strix.llm.config import LLMConfig
|
||||
from strix.tools.agents_graph import agents_graph_actions
|
||||
|
||||
|
||||
def test_create_agent_inherits_parent_whitebox_flag(monkeypatch) -> None:
|
||||
monkeypatch.setenv("STRIX_LLM", "openai/gpt-5")
|
||||
|
||||
agents_graph_actions._agent_graph["nodes"].clear()
|
||||
agents_graph_actions._agent_graph["edges"].clear()
|
||||
agents_graph_actions._agent_messages.clear()
|
||||
agents_graph_actions._running_agents.clear()
|
||||
agents_graph_actions._agent_instances.clear()
|
||||
agents_graph_actions._agent_states.clear()
|
||||
|
||||
parent_id = "parent-agent"
|
||||
parent_llm = LLMConfig(timeout=123, scan_mode="standard", is_whitebox=True)
|
||||
agents_graph_actions._agent_instances[parent_id] = SimpleNamespace(
|
||||
llm_config=parent_llm,
|
||||
non_interactive=True,
|
||||
)
|
||||
|
||||
captured_config: dict[str, object] = {}
|
||||
|
||||
class FakeStrixAgent:
|
||||
def __init__(self, config: dict[str, object]):
|
||||
captured_config["agent_config"] = config
|
||||
|
||||
class FakeThread:
|
||||
def __init__(self, target, args, daemon, name):
|
||||
self.target = target
|
||||
self.args = args
|
||||
self.daemon = daemon
|
||||
self.name = name
|
||||
|
||||
def start(self) -> None:
|
||||
return None
|
||||
|
||||
monkeypatch.setattr(agents_module, "StrixAgent", FakeStrixAgent)
|
||||
monkeypatch.setattr(agents_graph_actions.threading, "Thread", FakeThread)
|
||||
|
||||
agent_state = SimpleNamespace(
|
||||
agent_id=parent_id,
|
||||
get_conversation_history=list,
|
||||
)
|
||||
result = agents_graph_actions.create_agent(
|
||||
agent_state=agent_state,
|
||||
task="source-aware child task",
|
||||
name="SourceAwareChild",
|
||||
inherit_context=False,
|
||||
)
|
||||
|
||||
assert result["success"] is True
|
||||
llm_config = captured_config["agent_config"]["llm_config"]
|
||||
assert isinstance(llm_config, LLMConfig)
|
||||
assert llm_config.timeout == 123
|
||||
assert llm_config.scan_mode == "standard"
|
||||
assert llm_config.is_whitebox is True
|
||||
|
||||
|
||||
def test_delegation_prompt_includes_wiki_memory_instruction_in_whitebox(monkeypatch) -> None:
|
||||
monkeypatch.setenv("STRIX_LLM", "openai/gpt-5")
|
||||
|
||||
agents_graph_actions._agent_graph["nodes"].clear()
|
||||
agents_graph_actions._agent_graph["edges"].clear()
|
||||
agents_graph_actions._agent_messages.clear()
|
||||
agents_graph_actions._running_agents.clear()
|
||||
agents_graph_actions._agent_instances.clear()
|
||||
agents_graph_actions._agent_states.clear()
|
||||
|
||||
parent_id = "parent-1"
|
||||
child_id = "child-1"
|
||||
agents_graph_actions._agent_graph["nodes"][parent_id] = {"name": "Parent", "status": "running"}
|
||||
agents_graph_actions._agent_graph["nodes"][child_id] = {"name": "Child", "status": "running"}
|
||||
|
||||
class FakeState:
|
||||
def __init__(self) -> None:
|
||||
self.agent_id = child_id
|
||||
self.agent_name = "Child"
|
||||
self.parent_id = parent_id
|
||||
self.task = "analyze source risks"
|
||||
self.stop_requested = False
|
||||
self.messages: list[tuple[str, str]] = []
|
||||
|
||||
def add_message(self, role: str, content: str) -> None:
|
||||
self.messages.append((role, content))
|
||||
|
||||
def model_dump(self) -> dict[str, str]:
|
||||
return {"agent_id": self.agent_id}
|
||||
|
||||
class FakeAgent:
|
||||
def __init__(self) -> None:
|
||||
self.llm_config = LLMConfig(is_whitebox=True)
|
||||
|
||||
async def agent_loop(self, _task: str) -> dict[str, bool]:
|
||||
return {"ok": True}
|
||||
|
||||
state = FakeState()
|
||||
agent = FakeAgent()
|
||||
result = agents_graph_actions._run_agent_in_thread(agent, state, inherited_messages=[])
|
||||
|
||||
assert result["result"] == {"ok": True}
|
||||
task_messages = [msg for role, msg in state.messages if role == "user"]
|
||||
assert task_messages
|
||||
assert 'list_notes(category="wiki")' in task_messages[-1]
|
||||
100
tests/tools/test_notes_wiki.py
Normal file
100
tests/tools/test_notes_wiki.py
Normal file
@@ -0,0 +1,100 @@
|
||||
from pathlib import Path
|
||||
|
||||
from strix.telemetry.tracer import Tracer, get_global_tracer, set_global_tracer
|
||||
from strix.tools.notes import notes_actions
|
||||
|
||||
|
||||
def _reset_notes_state() -> None:
|
||||
notes_actions._notes_storage.clear()
|
||||
notes_actions._loaded_notes_run_dir = None
|
||||
|
||||
|
||||
def test_wiki_notes_are_persisted_and_removed(tmp_path: Path, monkeypatch) -> None:
|
||||
monkeypatch.chdir(tmp_path)
|
||||
_reset_notes_state()
|
||||
|
||||
previous_tracer = get_global_tracer()
|
||||
tracer = Tracer("wiki-test-run")
|
||||
set_global_tracer(tracer)
|
||||
|
||||
try:
|
||||
created = notes_actions.create_note(
|
||||
title="Repo Map",
|
||||
content="## Architecture\n- monolith",
|
||||
category="wiki",
|
||||
tags=["source-map"],
|
||||
)
|
||||
assert created["success"] is True
|
||||
note_id = created["note_id"]
|
||||
assert isinstance(note_id, str)
|
||||
|
||||
note = notes_actions._notes_storage[note_id]
|
||||
wiki_filename = note.get("wiki_filename")
|
||||
assert isinstance(wiki_filename, str)
|
||||
|
||||
wiki_path = tmp_path / "strix_runs" / "wiki-test-run" / "wiki" / wiki_filename
|
||||
assert wiki_path.exists()
|
||||
assert "## Architecture" in wiki_path.read_text(encoding="utf-8")
|
||||
|
||||
updated = notes_actions.update_note(
|
||||
note_id=note_id,
|
||||
content="## Architecture\n- service-oriented",
|
||||
)
|
||||
assert updated["success"] is True
|
||||
assert "service-oriented" in wiki_path.read_text(encoding="utf-8")
|
||||
|
||||
deleted = notes_actions.delete_note(note_id=note_id)
|
||||
assert deleted["success"] is True
|
||||
assert wiki_path.exists() is False
|
||||
finally:
|
||||
_reset_notes_state()
|
||||
set_global_tracer(previous_tracer) # type: ignore[arg-type]
|
||||
|
||||
|
||||
def test_notes_jsonl_replay_survives_memory_reset(tmp_path: Path, monkeypatch) -> None:
|
||||
monkeypatch.chdir(tmp_path)
|
||||
_reset_notes_state()
|
||||
|
||||
previous_tracer = get_global_tracer()
|
||||
tracer = Tracer("notes-replay-run")
|
||||
set_global_tracer(tracer)
|
||||
|
||||
try:
|
||||
created = notes_actions.create_note(
|
||||
title="Auth findings",
|
||||
content="initial finding",
|
||||
category="findings",
|
||||
tags=["auth"],
|
||||
)
|
||||
assert created["success"] is True
|
||||
note_id = created["note_id"]
|
||||
assert isinstance(note_id, str)
|
||||
|
||||
notes_path = tmp_path / "strix_runs" / "notes-replay-run" / "notes" / "notes.jsonl"
|
||||
assert notes_path.exists() is True
|
||||
|
||||
_reset_notes_state()
|
||||
listed = notes_actions.list_notes(category="findings")
|
||||
assert listed["success"] is True
|
||||
assert listed["total_count"] == 1
|
||||
assert listed["notes"][0]["note_id"] == note_id
|
||||
|
||||
updated = notes_actions.update_note(note_id=note_id, content="updated finding")
|
||||
assert updated["success"] is True
|
||||
|
||||
_reset_notes_state()
|
||||
listed_after_update = notes_actions.list_notes(search="updated finding")
|
||||
assert listed_after_update["success"] is True
|
||||
assert listed_after_update["total_count"] == 1
|
||||
assert listed_after_update["notes"][0]["note_id"] == note_id
|
||||
|
||||
deleted = notes_actions.delete_note(note_id=note_id)
|
||||
assert deleted["success"] is True
|
||||
|
||||
_reset_notes_state()
|
||||
listed_after_delete = notes_actions.list_notes(category="findings")
|
||||
assert listed_after_delete["success"] is True
|
||||
assert listed_after_delete["total_count"] == 0
|
||||
finally:
|
||||
_reset_notes_state()
|
||||
set_global_tracer(previous_tracer) # type: ignore[arg-type]
|
||||
Reference in New Issue
Block a user