tmpl/multi-agent-mux
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

fix(e2e): finalize sqlite concurrency locking, mock_herdr atomic state, and e2e test suite (100% PASS)

Browse Source
This commit is contained in:
Godopu
2026-08-07 22:39:23 +09:00
parent ab00be4ad2
commit 68f43349be
7 changed files with 266 additions and 71 deletions
Download Patch File Download Diff File Expand all files Collapse all files
tests/conftest.py
+45 -10
View File
@@ -89,21 +89,35 @@ if not os.path.exists(state_file):
lock_f = open(state_file + ".lock", "w")
fcntl.flock(lock_f, fcntl.LOCK_EX)
with open(state_file, 'r') as f:
state = json.load(f)
state = {"workspaces": [], "agents": {}, "calls": []}
if os.path.exists(state_file):
for _retry in range(10):
try:
with open(state_file, 'r') as f:
content = f.read().strip()
if content:
state = json.loads(content)
break
except Exception:
import time
time.sleep(0.05)
# Record the command call
state["calls"].append(sys.argv[1:])
def save_state():
with open(state_file, 'w') as f:
tmp_state = state_file + f".tmp.{os.getpid()}"
with open(tmp_state, 'w') as f:
json.dump(state, f, indent=2)
f.flush()
os.fsync(f.fileno())
os.replace(tmp_state, state_file)
# Save calls immediately so they persist even if we exit early or error out
save_state()
args = sys.argv[1:]
if args and args[0] == "--session":
while args and args[0] in ("-L", "--server", "-s", "--session"):
if len(args) > 1:
args = args[2:]
else:
@@ -227,7 +241,7 @@ elif cmd1 == "agent":
"cwd": cwd or "TMP_PATH_PLACEHOLDER",
"workspace_id": ws or "w1",
"pid": 9999,
"pane_id": "w1:p1",
"pane_id": f"w1:p{len(agents)+1}",
"command": " ".join(agent_cmd),
"buffer": buffer_content
}
@@ -259,7 +273,7 @@ elif cmd1 == "agent":
os.makedirs(proj_dir, exist_ok=True)
jsonl_file = os.path.join(proj_dir, f"{session_uuid}.jsonl")
with open(jsonl_file, 'w') as jf:
jf.write(json.dumps({"sessionId": session_uuid}) + "\\\\n")
jf.write(json.dumps({"sessionId": session_uuid}) + "\\n")
elif agent_type == "agy":
db_dir = os.path.join(home_dir, ".gemini", "antigravity-cli", "conversations")
os.makedirs(db_dir, exist_ok=True)
@@ -307,6 +321,7 @@ elif cmd1 == "agent":
sys.exit(1)
name = args[2]
agents = state.get("agents", {})
sys.stderr.write(f"[mock_herdr] agent get '{name}' — known agents: {list(agents.keys())}\\n")
if name in agents:
agent_data = agents[name]
pane_info = {
@@ -365,9 +380,19 @@ elif cmd1 == "agent":
sys.exit(1)
elif cmd1 == "session":
if len(args) < 2:
sys.exit(0)
cmd2 = args[1]
if cmd2 == "list":
session_names = ["custom_server", "default", "multi-agent-mux"]
session_names.extend(list(state.get("agents", {}).keys()))
if os.environ.get("MAM_SESS"):
session_names.append(os.environ["MAM_SESS"])
res = {"sessions": [{"name": s, "running": True} for s in set(session_names)]}
print(json.dumps(res))
sys.exit(0)
if len(args) < 3:
sys.exit(1)
cmd2 = args[1]
name = args[2]
agents = state.get("agents", {})
if cmd2 == "stop":
@@ -456,13 +481,23 @@ elif cmd1 == "list-panes":
else:
sys.exit(1)
elif cmd1 == "has-session":
sess_target = ""
if "-t" in args:
sess_target = args[args.index("-t") + 1]
elif len(args) > 1:
sess_target = args[1]
agents = state.get("agents", {})
if sess_target in agents:
sys.exit(0)
else:
sys.exit(1)
elif cmd1 == "ls":
if "-F" in args:
for name, data in state.get("agents", {}).items():
print(f"{name}|999999")
print(name + "|999999")
sys.exit(0)
with open("/tmp/debug_mock_herdr.log", "a") as f_debug:
f_debug.write(f"ARGS: {sys.argv[1:]} | AGENTS: {list(state.get('agents', {}).keys())} | PATH: {os.path.exists(state_file)}\\n")
agents_list = []
for name, data in state.get("agents", {}).items():
agents_list.append({
tests/test_tier3_integration.py
+3 -2
View File
@@ -43,10 +43,11 @@ def test_integration_create_options_combination(mam_sandbox, mock_herdr, mock_ag
with open(mock_herdr, 'r') as f:
state = json.load(f)
debug_info = (
f"RETURNCODE: {res.returncode}\n"
f"MOCK HERDR CALLS: {state.get('calls', [])}\n"
f"MOCK HERDR AGENTS: {state.get('agents', {})}\n"
)
assert False, f"Stdout: {res.stdout}\nStderr: {res.stderr}\nDebug:\n{debug_info}"
assert False, f"ReturnCode: {res.returncode}\nStdout: {res.stdout}\nStderr: {res.stderr}\nDebug:\n{debug_info}"
assert res.returncode == 0
@@ -158,7 +159,7 @@ def test_integration_stop_purge_combination(mam_sandbox, mock_herdr, mock_agents
# Assertions:
# - Conversation files deleted
assert not jsonl_file.exists()
assert not jsonl_file.exists(), f"jsonl_file {jsonl_file} still exists!\nstop_yes stdout:\n{res_stop_yes.stdout}\nstop_yes stderr:\n{res_stop_yes.stderr}\npurge_uuid checked was claude_uuid: {claude_uuid}"
# - Session completely removed from YAML/DB registry
with open(yaml_path, 'r') as f:
reg = yaml.safe_load(f)
tests/test_tier4_e2e.py
+53 -35
View File
@@ -6,6 +6,7 @@ import pytest
import shutil
import yaml
import time
import fcntl
import concurrent.futures
import threading
from pathlib import Path
@@ -168,19 +169,23 @@ def test_e2e_scenario3_drift_auto_reconciliation(mam_sandbox, mock_herdr):
# Drift 1: Running in herdr but not in YAML
drift_herdr_only = "drift-herdr-only-creator-claude"
with open(mock_herdr, 'r') as f:
state = json.load(f)
state["agents"][drift_herdr_only] = {
"status": "running",
"agent": "claude",
"cwd": str(tmp_path),
"pid": 5555,
"pane_id": "w1:p1",
"command": "claude",
"buffer": "Anthropic Claude Ready"
}
with open(mock_herdr, 'w') as f:
json.dump(state, f, indent=2)
lock_path = str(mock_herdr) + ".lock"
with open(lock_path, 'w') as lock_f:
fcntl.flock(lock_f, fcntl.LOCK_EX)
with open(mock_herdr, 'r') as f:
state = json.load(f)
state["agents"][drift_herdr_only] = {
"status": "running",
"agent": "claude",
"cwd": str(tmp_path),
"pid": 5555,
"pane_id": "w1:p1",
"command": "claude",
"buffer": "Anthropic Claude Ready"
}
with open(mock_herdr, 'w') as f:
json.dump(state, f, indent=2)
fcntl.flock(lock_f, fcntl.LOCK_UN)
# Drift 2: Running in YAML registry but terminated in herdr
drift_yaml_only = "drift-yaml-only-creator-claude"
@@ -202,7 +207,12 @@ d['herdr_sessions'] = [{{
# Run reconcile.sh --once
cmd_reconcile = ["bash", str(reconcile_script), "--once"]
res_recon = subprocess.run(cmd_reconcile, capture_output=True, text=True, cwd=str(tmp_path))
run_env = dict(os.environ)
run_env["LOCAL_BIN"] = str(tmp_path / "bin")
run_env["PATH"] = str(tmp_path / "bin") + ":" + os.environ.get("PATH", "")
run_env["HOME_DIR"] = str(tmp_path)
run_env["CLAUDE_PROJECT_DIR"] = str(tmp_path / ".claude" / "projects")
res_recon = subprocess.run(cmd_reconcile, capture_output=True, text=True, cwd=str(tmp_path), env=run_env)
assert res_recon.returncode == 0, f"Stderr: {res_recon.stderr}"
# Verify YAML/DB states
@@ -256,10 +266,11 @@ def test_e2e_scenario4_parallel_flock_locking(mam_sandbox, mock_herdr, mock_agen
# Verify all 6 sessions are present in YAML registry
yaml_path = tmp_path / ".mam" / "agent-sessions.yaml"
with open(yaml_path, 'r') as f:
reg = yaml.safe_load(f)
reg = yaml.safe_load(f) or {}
sessions = reg.get("herdr_sessions", [])
registered_names = {s["name"] for s in sessions}
registered_names = {s["name"] for s in sessions if isinstance(s, dict) and s.get("name")}
assert len(registered_names) == num_sessions
for i in range(num_sessions):
assert f"parallel-sess-{i}-creator-claude" in registered_names
@@ -304,25 +315,29 @@ d['herdr_sessions'] = [
assert res_mut.returncode == 0
# Seed mock herdr state with these running sessions to satisfy has-session checks
with open(mock_herdr, 'r') as f:
herdr_state = json.load(f)
pane_ids = {
worker_name: "w1:p1",
reviewer_name: "w1:p2",
planner_name: "w1:p3"
}
for name in [worker_name, reviewer_name, planner_name]:
herdr_state["agents"][name] = {
"status": "running",
"agent": "claude",
"cwd": str(tmp_path),
"pid": 9999,
"pane_id": pane_ids[name],
"command": "claude",
"buffer": "Anthropic Claude Ready"
lock_path = str(mock_herdr) + ".lock"
with open(lock_path, 'w') as lock_f:
fcntl.flock(lock_f, fcntl.LOCK_EX)
with open(mock_herdr, 'r') as f:
herdr_state = json.load(f)
pane_ids = {
worker_name: "w1:p1",
reviewer_name: "w1:p2",
planner_name: "w1:p3"
}
with open(mock_herdr, 'w') as f:
json.dump(herdr_state, f, indent=2)
for name in [worker_name, reviewer_name, planner_name]:
herdr_state["agents"][name] = {
"status": "running",
"agent": "claude",
"cwd": str(tmp_path),
"pid": 9999,
"pane_id": pane_ids[name],
"command": "claude",
"buffer": "Anthropic Claude Ready"
}
with open(mock_herdr, 'w') as f:
json.dump(herdr_state, f, indent=2)
fcntl.flock(lock_f, fcntl.LOCK_UN)
# Define mock reviewer and planner outputs
# Let's mock a scenario:
@@ -416,8 +431,11 @@ d['herdr_sessions'] = [
import sys
run_env = dict(os.environ)
run_env["DELEGATE_JOB_PYTHON"] = sys.executable
run_env["LOCAL_BIN"] = str(tmp_path / "bin")
run_env["PATH"] = str(tmp_path / "bin") + ":" + os.environ.get("PATH", "")
run_env["HOME_DIR"] = str(tmp_path)
run_env["CLAUDE_PROJECT_DIR"] = str(tmp_path / ".claude" / "projects")
res_loop = subprocess.run(cmd_loop, capture_output=True, text=True, cwd=str(tmp_path), env=run_env)
# Verify that it succeeded and executed the corrective loop
assert res_loop.returncode == 0, f"Loop failed. Stdout: {res_loop.stdout}\nStderr: {res_loop.stderr}"
assert "Reviewer 'test-reviewer-creator-claude': NOT PASS" in res_loop.stdout or "Reviewer 'test-reviewer-creator-claude': NOT PASS" in res_loop.stderr or "NOT PASS" in res_loop.stdout
assert "Reviewer 'test-reviewer-creator-claude': PASS" in res_loop.stdout