Shared memory and context tools for agentic work.
Code Rooms
import importlib.util
import json
import pathlib
import tempfile
import unittest
from types import SimpleNamespace
REPO_ROOT = pathlib.Path(__file__).resolve().parents[1]
BUG_HUNT_PATH = REPO_ROOT / "scripts" / "benchmark" / "bug_hunt_round.py"
def load_bug_hunt_module():
spec = importlib.util.spec_from_file_location("bug_hunt_round", BUG_HUNT_PATH)
module = importlib.util.module_from_spec(spec)
assert spec.loader is not None
spec.loader.exec_module(module)
return module
class BugHuntMissionControlTests(unittest.TestCase):
@classmethod
def setUpClass(cls):
cls.bug_hunt = load_bug_hunt_module()
def fill_answer_key(self, out_dir, bug_count=1):
bugs = []
for index in range(bug_count):
bugs.append(
{
"id": f"seeded-{index + 1}",
"severity": "medium",
"file": "index.js",
"symbol": "demo",
"expected": "demo seeded bug",
"match_terms": [["demo"]],
}
)
payload = {
"schema": self.bug_hunt.ANSWER_KEY_SCHEMA,
"round_id": "round",
"repo": "repo",
"source_commit": None,
"seeded_bug_count": bug_count,
"bugs": bugs,
"non_claims": [],
answer_key = pathlib.Path(out_dir) / "operator-only" / "answer-key.json"
answer_key.write_text(json.dumps(payload), encoding="utf-8")
def test_structured_mission_control_usage_completes_loop(self):
result = {
"mission_control_usage": {
"mission_control_unavailable": False,
"mission_start_called": True,
"mission_event_count": 2,
"mission_next_count": 3,
"mission_verify_count": 1,
"mission_handoff_called": True,
"mission_close_called": True,
"do_not_guardrails_observed": ["call activate"],
"verified_claims": ["claim-1"],
"direct_proof_switches": ["step-2"],
"coverage_sweeps": ["boundary sweep"],
"event_digest": "hash64:abc",
"handoff_summary": "resume at close",
"proof_packet_summary": "closed with proof",
summary = self.bug_hunt.summarize_mission_control(
result, [], "m1nd-mission-control"
self.assertTrue(summary["applicable"])
self.assertTrue(summary["loop_complete"])
self.assertFalse(summary["unavailable"])
self.assertEqual(summary["mission_start_count"], 1)
self.assertEqual(summary["mission_event_count"], 2)
self.assertEqual(summary["mission_next_count"], 3)
self.assertEqual(summary["mission_verify_count"], 1)
self.assertEqual(summary["mission_handoff_count"], 1)
self.assertEqual(summary["mission_close_count"], 1)
self.assertEqual(summary["completed_step_count"], 6)
self.assertEqual(summary["required_step_count"], 6)
self.assertEqual(summary["adherence_rate"], 1.0)
self.assertEqual(summary["do_not_guardrail_count"], 1)
self.assertEqual(summary["verified_claim_signal_count"], 1)
self.assertEqual(summary["direct_proof_switch_count"], 1)
self.assertEqual(summary["coverage_sweep_count"], 1)
self.assertTrue(summary["event_digest_present"])
self.assertTrue(summary["handoff_summary_present"])
self.assertTrue(summary["proof_packet_summary_present"])
self.assertTrue(summary["evidence_packet_complete"])
def test_default_template_does_not_create_false_unavailable_or_loop(self):
round_payload = {"round_id": "round", "repo": "repo"}
lane = {"lane_id": "audit-01", "instruction_mode": "m1nd-mission-control"}
result = self.bug_hunt.result_template(round_payload, lane)
self.assertIn("tool_usage_summary", result)
self.assertEqual(result["tool_usage_summary"]["total_observed_action_count"], 0)
self.assertFalse(summary["loop_complete"])
self.assertEqual(summary["mission_start_count"], 0)
self.assertEqual(summary["mission_event_count"], 0)
self.assertEqual(summary["mission_next_count"], 0)
self.assertEqual(summary["mission_verify_count"], 0)
self.assertEqual(summary["mission_handoff_count"], 0)
self.assertEqual(summary["mission_close_count"], 0)
self.assertEqual(summary["completed_step_count"], 0)
self.assertEqual(summary["adherence_rate"], 0.0)
self.assertFalse(summary["evidence_packet_complete"])
def test_tool_usage_summary_falls_back_from_template_zeros(self):
"commands_run": ["rg bug", "PYTHONPATH=src python3 - <<'PY'"],
"files_inspected": ["src/app.py", "tests/test_app.py"],
"m1nd_usage": [{"tool": "search"}],
"tool_usage_summary": {
"shell_command_count": 0,
"file_read_count": 0,
"test_or_probe_count": 0,
"runtime_probe_count": 0,
"m1nd_call_count": 0,
"mission_control_call_count": 0,
"total_observed_action_count": 0,
},
mission_control = {
"applicable": True,
"mission_start_count": 1,
"mission_next_count": 1,
"mission_handoff_count": 1,
"mission_close_count": 1,
events = [
{"event_type": "runtime_probe", "event_source": "agent"},
]
timing = {
"first_finding_event_elapsed_seconds": 12.5,
"first_seeded_finding_event_elapsed_seconds": None,
usage = self.bug_hunt.summarize_tool_usage(
result, events, mission_control, timing
self.assertEqual(usage["shell_command_count"], 2)
self.assertEqual(usage["file_read_count"], 2)
self.assertEqual(usage["test_or_probe_count"], 1)
self.assertEqual(usage["runtime_probe_count"], 1)
self.assertEqual(usage["m1nd_call_count"], 1)
self.assertEqual(usage["mission_control_call_count"], 7)
self.assertEqual(usage["total_observed_action_count"], 14)
self.assertEqual(usage["first_good_finding_seconds"], 12.5)
def test_tool_usage_does_not_count_m1nd_helper_probe_as_runtime_probe(self):
"commands_run": [
"probe_m1nd.py call mission_start",
"probe_m1nd.py call mission_next",
"PYTHONPATH=src python3 focused runtime probe",
],
"files_inspected": [],
"m1nd_usage": [{"tool": "mission_start"}, {"tool": "mission_next"}],
mission_control = {"applicable": False}
"first_finding_event_elapsed_seconds": None,
usage = self.bug_hunt.summarize_tool_usage(result, [], mission_control, timing)
self.assertEqual(usage["shell_command_count"], 3)
self.assertEqual(usage["runtime_probe_count"], 0)
self.assertEqual(usage["m1nd_call_count"], 2)
def test_source_backed_finding_requires_file_evidence_and_repro(self):
self.assertTrue(
self.bug_hunt.finding_is_source_backed(
"file": "src/app.py",
"evidence": ["src/app.py:10"],
"reproduction_or_test": "pytest tests/test_app.py",
self.assertFalse(
"evidence": "src/app.py:10",
def test_non_mission_control_lane_does_not_emit_mission_metrics(self):
"notes": "mission_start mission_next mission_verify mission_close",
"mission_control_unavailable": True,
summary = self.bug_hunt.summarize_mission_control(result, [], "m1nd-trained")
self.assertFalse(summary["applicable"])
self.assertIsNone(summary["unavailable"])
self.assertIsNone(summary["loop_complete"])
self.assertIsNone(summary["mission_next_count"])
self.assertIsNone(summary["adherence_rate"])
self.assertIsNone(summary["coverage_sweep_count"])
self.assertIsNone(summary["evidence_packet_complete"])
def test_structured_unavailable_mission_control_does_not_count_text_tokens(self):
self.assertTrue(summary["unavailable"])
def test_mission_control_validity_marks_partial_lanes_not_evaluable(self):
lanes = [
"lane_id": "audit-01",
"instruction_mode": "m1nd-mission-control",
"completed": True,
"mission_control": {"loop_complete": True, "unavailable": False},
"lane_id": "audit-02",
"mission_control": {"loop_complete": False, "unavailable": False},
"lane_id": "audit-03",
"completed": False,
"lane_id": "audit-04",
"instruction_mode": "direct",
validity = self.bug_hunt.mission_control_validity(lanes)
self.assertTrue(validity["present"])
self.assertFalse(validity["all_completed_lanes_evaluable"])
self.assertEqual(validity["lane_count"], 3)
self.assertEqual(validity["evaluable_lane_ids"], ["audit-01"])
self.assertEqual(validity["partial_or_unavailable_lane_ids"], ["audit-02"])
self.assertEqual(validity["missing_result_lane_ids"], ["audit-03"])
def test_next_actions_keep_exposure_fix_for_unevaluable_mission_control(self):
report = {
"comparability": {
"mission_control_validity": {
"present": True,
"all_completed_lanes_evaluable": False,
actions = self.bug_hunt.next_product_actions(report)
self.assertIn(
"Fix worker-host Mission Control exposure before rerunning Mission Control comparisons.",
actions,
self.assertNotIn(
"Analyze conservative Mission Control misses and tune mission_next/mission_verify stopping rules.",
def test_next_actions_tune_mission_control_when_evaluable(self):
"all_completed_lanes_evaluable": True,
"arms": {"m1nd-mission-control": {"seeded_recall_rate": 0.9}},
def test_next_actions_repeat_when_coverage_sweep_round_succeeds(self):
"arms": {
"m1nd-mission-control": {
"seeded_recall_rate": 1.0,
"median_coverage_sweep_count": 1.0,
"Repeat the Mission Control direct-sweep calibration on another fixture before treating it as a generalized improvement.",
def test_preflight_accepts_mission_control_round_scaffold(self):
with tempfile.TemporaryDirectory() as temp_dir:
out_dir = pathlib.Path(temp_dir)
args = SimpleNamespace(
out_dir=out_dir,
round_id="round",
repo="repo",
source_repo=None,
seeded_repo=None,
workspace_root=None,
source_commit=None,
seeded_bug_count=1,
lanes_full_spec=0,
lanes_temponizer_full=0,
lanes_temponizer_compact=0,
lanes_temponizer=0,
lanes_short_audit=0,
lanes_mission_control=1,
lanes_trained=0,
lanes_basic=0,
lanes_direct=1,
self.bug_hunt.init_round(args)
self.fill_answer_key(out_dir)
preflight = self.bug_hunt.preflight_round(
out_dir / "round.json",
required_modes=["m1nd-mission-control", "direct"],
self.assertTrue(preflight["ok"])
self.assertEqual(preflight["blockers"], [])
self.assertEqual(preflight["arm_lane_counts"]["m1nd-mission-control"], 1)
self.assertEqual(preflight["arm_lane_counts"]["direct"], 1)
def test_init_materializes_seeded_lane_workspaces(self):
root = pathlib.Path(temp_dir)
seeded = root / "seeded"
seeded.mkdir()
(seeded / "index.js").write_text("export default 1;\n", encoding="utf-8")
out_dir = root / "round"
workspace_root = root / "workspaces"
seeded_repo=seeded,
workspace_root=workspace_root,
lanes_direct=0,
no_materialize_workspaces=False,
round_payload = self.bug_hunt.init_round(args)
repo_path = pathlib.Path(round_payload["lanes"][0]["repo_path"])
self.assertTrue((repo_path / "index.js").exists())
self.assertEqual(round_payload["materialized_workspaces"][0]["status"], "created")
any("copied the seeded repo" in item for item in round_payload["non_claims"])
def test_preflight_requires_live_mission_control_tools(self):
fake_mcp = self.write_fake_mcp(out_dir, ["trust_selftest", "seek"])
required_modes=["m1nd-mission-control"],
require_live_mission_control=True,
m1nd_binary=str(fake_mcp),
self.assertFalse(preflight["ok"])
self.assertEqual(preflight["mission_control_surface"]["tool_count"], 2)
any(
"live Mission Control tools unavailable" in item
for item in preflight["blockers"]
def test_preflight_accepts_live_mission_control_tools(self):
fake_mcp = self.write_fake_mcp(
out_dir,
[
"mission_start",
"mission_event",
"mission_next",
"mission_verify",
"mission_handoff",
"mission_close",
"trust_selftest",
preflight["mission_control_surface"]["mission_control_available"]
def test_preflight_rejects_broken_mission_control_prompt(self):
prompt_path = pathlib.Path(round_payload["lanes"][0]["prompt"])
prompt_path.write_text("broken prompt", encoding="utf-8")
"mission-control prompt missing tokens" in item
def test_preflight_rejects_empty_answer_key(self):
any("answer key has no bugs" in item for item in preflight["blockers"])
def write_fake_mcp(self, out_dir, tool_names):
fake_mcp = pathlib.Path(out_dir) / "fake-m1nd-mcp.py"
script = f"""#!/usr/bin/env python3
import sys
TOOLS = {json.dumps(tool_names)}
for line in sys.stdin:
request = json.loads(line)
method = request.get("method")
response = {{"jsonrpc": "2.0", "id": request.get("id")}}
if method == "initialize":
response["result"] = {{"serverInfo": {{"name": "fake-m1nd-mcp"}}}}
elif method == "tools/list":
response["result"] = {{"tools": [{{"name": name}} for name in TOOLS]}}
else:
response["error"] = {{"code": -32601, "message": "not found"}}
print(json.dumps(response), flush=True)
"""
fake_mcp.write_text(script, encoding="utf-8")
fake_mcp.chmod(0o755)
return fake_mcp
if __name__ == "__main__":
unittest.main()
