a2bc6347fc
Architecture: - Graph engine (engine.py) loads graph definitions, instantiates nodes - Versioned nodes: input_v1, thinker_v1, output_v1, memorizer_v1, director_v1 - NODE_REGISTRY for dynamic node lookup by name - Graph API: /api/graph/active, /api/graph/list, /api/graph/switch - Graph definition: graphs/v1_current.py (7 nodes, 13 edges, 3 edge types) S3* Audit system: - Workspace mismatch detection (server vs browser controls) - Code-without-tools retry (Thinker wrote code but no tool calls) - Intent-without-action retry (request intent but Thinker only produced text) - Dashboard feedback: browser sends workspace state on every message - Sensor continuous comparison on 5s tick State machines: - create_machine / add_state / reset_machine / destroy_machine via function calling - Local transitions (go:) resolve without LLM round-trip - Button persistence across turns Database tools: - query_db tool via pymysql to MariaDB K3s pod (eras2_production) - Table rendering in workspace (tab-separated parsing) - Director pre-planning with Opus for complex data requests - Error retry with corrected SQL Frontend: - Cytoscape.js pipeline graph with real-time node animations - Overlay scrollbars (CSS-only, no reflow) - Tool call/result trace events - S3* audit events in trace Testing: - 167 integration tests (11 test suites) - 22 node-level unit tests (test_nodes/) - Three test levels: node unit, graph integration, scenario Co-Authored-By: Claude Opus 4.6 (1M context) <noreply@anthropic.com>
89 lines
3.1 KiB
Python
89 lines
3.1 KiB
Python
"""Unit tests for MemorizerNode v1 — fact retention, state distillation."""
|
|
|
|
from harness import HudCapture, make_history, NodeTestRunner
|
|
|
|
from agent.nodes.memorizer_v1 import MemorizerNode
|
|
|
|
|
|
async def test_extracts_mood():
|
|
"""Memorizer should detect user mood from conversation."""
|
|
hud = HudCapture()
|
|
node = MemorizerNode(send_hud=hud)
|
|
history = make_history([
|
|
("user", "this is amazing, I love it!"),
|
|
("assistant", "Glad you're enjoying it!"),
|
|
])
|
|
await node.update(history)
|
|
assert node.state.get("user_mood") in ("happy", "excited", "positive"), \
|
|
f"mood={node.state.get('user_mood')}"
|
|
|
|
|
|
async def test_extracts_language():
|
|
"""Memorizer should detect language switch."""
|
|
hud = HudCapture()
|
|
node = MemorizerNode(send_hud=hud)
|
|
history = make_history([
|
|
("user", "Hallo, wie geht es dir?"),
|
|
("assistant", "Mir geht es gut, danke!"),
|
|
])
|
|
await node.update(history)
|
|
assert node.state.get("language") in ("de", "mixed"), \
|
|
f"language={node.state.get('language')}"
|
|
|
|
|
|
async def test_facts_preserved_across_updates():
|
|
"""Old facts should not be dropped by subsequent updates."""
|
|
hud = HudCapture()
|
|
node = MemorizerNode(send_hud=hud)
|
|
# First update: learn a fact
|
|
history1 = make_history([
|
|
("user", "My dog's name is Bella"),
|
|
("assistant", "Bella is a lovely name!"),
|
|
])
|
|
await node.update(history1)
|
|
assert any("bella" in f.lower() for f in node.state.get("facts", [])), \
|
|
f"Bella not in facts: {node.state.get('facts')}"
|
|
|
|
# Second update: different topic, old fact should survive
|
|
history2 = history1 + make_history([
|
|
("user", "what time is it?"),
|
|
("assistant", "It's 3pm."),
|
|
])
|
|
await node.update(history2)
|
|
assert any("bella" in f.lower() for f in node.state.get("facts", [])), \
|
|
f"Bella dropped after 2nd update: {node.state.get('facts')}"
|
|
|
|
|
|
async def test_topic_tracked():
|
|
"""Memorizer should track the current topic."""
|
|
hud = HudCapture()
|
|
node = MemorizerNode(send_hud=hud)
|
|
history = make_history([
|
|
("user", "let's talk about cooking pasta"),
|
|
("assistant", "Great topic! What kind of pasta?"),
|
|
])
|
|
await node.update(history)
|
|
topic = node.state.get("topic", "")
|
|
assert "pasta" in topic.lower() or "cook" in topic.lower(), f"topic={topic}"
|
|
|
|
|
|
async def test_emits_updated_hud():
|
|
"""Memorizer should emit 'updated' HUD event with state."""
|
|
hud = HudCapture()
|
|
node = MemorizerNode(send_hud=hud)
|
|
history = make_history([("user", "hello"), ("assistant", "hi")])
|
|
await node.update(history)
|
|
assert hud.has("updated"), f"events: {[e.get('event') for e in hud.events]}"
|
|
|
|
|
|
if __name__ == "__main__":
|
|
runner = NodeTestRunner()
|
|
print("\n=== MemorizerNode v1 ===")
|
|
runner.test("extracts mood", test_extracts_mood())
|
|
runner.test("extracts language", test_extracts_language())
|
|
runner.test("facts preserved across updates", test_facts_preserved_across_updates())
|
|
runner.test("topic tracked", test_topic_tracked())
|
|
runner.test("emits updated HUD", test_emits_updated_hud())
|
|
p, f = runner.summary()
|
|
print(f"\n {p} passed, {f} failed")
|