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>
82 lines
3.3 KiB
Python
82 lines
3.3 KiB
Python
"""Unit tests for DirectorNode v1 — style directives + Opus planning."""
|
|
|
|
import json
|
|
from harness import HudCapture, make_history, NodeTestRunner
|
|
|
|
from agent.nodes.director_v1 import DirectorNode
|
|
|
|
|
|
async def test_detects_casual_mode():
|
|
"""Director should detect casual chat mode."""
|
|
hud = HudCapture()
|
|
node = DirectorNode(send_hud=hud)
|
|
history = make_history([
|
|
("user", "hey, just hanging out"),
|
|
("assistant", "Hey! What's up?"),
|
|
("user", "not much, just chilling"),
|
|
("assistant", "Nice, enjoy the evening!"),
|
|
])
|
|
await node.update(history, {"user_mood": "happy", "topic": "casual chat"})
|
|
assert node.directive["mode"] == "casual", f"mode={node.directive['mode']}"
|
|
|
|
|
|
async def test_detects_frustrated_style():
|
|
"""Director should adjust style when user is frustrated."""
|
|
hud = HudCapture()
|
|
node = DirectorNode(send_hud=hud)
|
|
history = make_history([
|
|
("user", "this is completely broken, nothing works"),
|
|
("assistant", "Let me help fix that."),
|
|
])
|
|
await node.update(history, {"user_mood": "frustrated", "topic": "debugging"})
|
|
style = node.directive.get("style", "").lower()
|
|
assert any(k in style for k in ["simplif", "patient", "calm", "help", "step"]), \
|
|
f"style doesn't address frustration: {style}"
|
|
|
|
|
|
async def test_produces_plan_for_complex_request():
|
|
"""Director.plan() should produce an investigation plan with Opus."""
|
|
hud = HudCapture()
|
|
node = DirectorNode(send_hud=hud)
|
|
history = make_history([
|
|
("user", "investigate which customers have the most devices"),
|
|
])
|
|
plan = await node.plan(history, {"topic": "database"}, "investigate which customers have the most devices")
|
|
assert plan, "empty plan"
|
|
assert "query_db" in plan.lower() or "select" in plan.lower() or "step" in plan.lower(), \
|
|
f"plan doesn't mention DB tools: {plan[:200]}"
|
|
assert node.current_plan, "plan not stored in current_plan"
|
|
|
|
|
|
async def test_directive_has_required_fields():
|
|
"""Directive should have mode, style, proactive."""
|
|
hud = HudCapture()
|
|
node = DirectorNode(send_hud=hud)
|
|
history = make_history([("user", "hello"), ("assistant", "hi")])
|
|
await node.update(history, {"user_mood": "neutral"})
|
|
assert "mode" in node.directive
|
|
assert "style" in node.directive
|
|
assert "proactive" in node.directive
|
|
|
|
|
|
async def test_context_line_includes_plan():
|
|
"""get_context_line() should include the plan when set."""
|
|
hud = HudCapture()
|
|
node = DirectorNode(send_hud=hud)
|
|
node.current_plan = "Step 1: query kunden table"
|
|
line = node.get_context_line()
|
|
assert "Step 1" in line, f"plan not in context line: {line}"
|
|
assert "DIRECTOR PLAN" in line, f"missing plan header: {line}"
|
|
|
|
|
|
if __name__ == "__main__":
|
|
runner = NodeTestRunner()
|
|
print("\n=== DirectorNode v1 ===")
|
|
runner.test("detects casual mode", test_detects_casual_mode())
|
|
runner.test("detects frustrated style", test_detects_frustrated_style())
|
|
runner.test("produces plan for complex request", test_produces_plan_for_complex_request())
|
|
runner.test("directive has required fields", test_directive_has_required_fields())
|
|
runner.test("context line includes plan", test_context_line_includes_plan())
|
|
p, f = runner.summary()
|
|
print(f"\n {p} passed, {f} failed")
|