generated from coulomb/repo-seed
tegwick
21c714e286
signals.py: tool_bucket helper + three tool_histogram-based extractors that the outcome/marker signals were blind to — sig_infra_overhead (hub+task+schema share of tool calls over threshold), sig_schema_thrash (repeated ToolSearch), and sig_tool_thrash (one tool dominating). Thresholds in build_context. 8 new tests; suite 88/88 green. Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
81 lines
2.6 KiB
Python
81 lines
2.6 KiB
Python
"""Infra-overhead + thrash signal tests (WP-0005 T02)."""
|
|
|
|
import os
|
|
import sys
|
|
|
|
sys.path.insert(0, os.path.dirname(os.path.dirname(os.path.abspath(__file__))))
|
|
|
|
from session_memory.detect.signals import ( # noqa: E402
|
|
build_context,
|
|
extract_signals,
|
|
sig_infra_overhead,
|
|
sig_schema_thrash,
|
|
sig_tool_thrash,
|
|
tool_bucket,
|
|
)
|
|
|
|
|
|
def _digest(uid="claude:a", repo="r1", tools=None):
|
|
return {"session_uid": uid, "flavor": "claude", "repo": repo, "outcome": "success",
|
|
"cost": {"input_tokens": 1, "output_tokens": 1},
|
|
"markers": {"errors": 0, "retries": 0, "test_runs": 0},
|
|
"tool_histogram": tools or {}}
|
|
|
|
|
|
CTX = {"infra_min_calls": 20, "infra_overhead_threshold": 0.30,
|
|
"schema_thrash_threshold": 5, "tool_thrash_threshold": 80}
|
|
|
|
|
|
def test_tool_bucket_mapping():
|
|
assert tool_bucket("mcp__state-hub__update_task_status") == "statehub_mcp"
|
|
assert tool_bucket("ToolSearch") == "schema_load"
|
|
assert tool_bucket("TaskUpdate") == "task_mgmt"
|
|
assert tool_bucket("Bash") == "shell"
|
|
assert tool_bucket("Edit") == "edit"
|
|
|
|
|
|
def test_infra_overhead_fires_above_share():
|
|
# 18 statehub of 30 total = 60% overhead
|
|
d = _digest(tools={"mcp__state-hub__create_task": 18, "Bash": 8, "Edit": 4})
|
|
sig = sig_infra_overhead(d, CTX)
|
|
assert sig and sig[0].type == "infra_overhead"
|
|
assert sig[0].magnitude >= 0.30
|
|
assert sig[0].detail["statehub"] == 18
|
|
|
|
|
|
def test_infra_overhead_quiet_when_mostly_work():
|
|
d = _digest(tools={"mcp__state-hub__create_task": 3, "Bash": 40, "Edit": 30})
|
|
assert sig_infra_overhead(d, CTX) == []
|
|
|
|
|
|
def test_infra_overhead_ignores_tiny_sessions():
|
|
d = _digest(tools={"mcp__state-hub__create_task": 5}) # below infra_min_calls
|
|
assert sig_infra_overhead(d, CTX) == []
|
|
|
|
|
|
def test_schema_thrash_fires():
|
|
d = _digest(tools={"ToolSearch": 9, "Bash": 5})
|
|
sig = sig_schema_thrash(d, CTX)
|
|
assert sig and sig[0].type == "schema_thrash"
|
|
assert sig[0].detail["tool_searches"] == 9
|
|
|
|
|
|
def test_tool_thrash_fires_on_dominant_tool():
|
|
d = _digest(tools={"Bash": 120, "Edit": 5})
|
|
sig = sig_tool_thrash(d, CTX)
|
|
assert sig and sig[0].locus == "tool:Bash"
|
|
|
|
|
|
def test_extract_signals_includes_infra():
|
|
d = _digest(tools={"mcp__state-hub__create_task": 18, "Bash": 8, "Edit": 4,
|
|
"ToolSearch": 6})
|
|
types = {s.type for s in extract_signals([d])}
|
|
assert "infra_overhead" in types
|
|
assert "schema_thrash" in types
|
|
|
|
|
|
def test_build_context_has_infra_defaults():
|
|
ctx = build_context([])
|
|
assert ctx["infra_overhead_threshold"] == 0.30
|
|
assert ctx["schema_thrash_threshold"] == 5
|