GodopuandClaude Opus 5.5
feccfd5e03
- BaseAgentAdapter: id_lifecycle (pre_assigned: claude, grok / post_discovered) and capture_active_session_id - exactly one verified candidate newer than the session that no other row records, else a reason (no-candidate, ambiguous, ...). - agy: verify_artifact (discover) rejects an id last_conversations.json maps to another workspace and always excludes ids other rows record (a cache hit no longer bypasses it). capture uses the cache only as a proposal: the conversation's presence lock or DB must be held by the row's pane process or a descendant (lsof/ps); anything unprovable is not pinned. - lib_py/agents/autopin.py: handle_completed, run by the side that verified the job's completed - the delegate-job wrapper (direct), create --onboard and the loop's wait_for_job (recovery, as a subprocess: G9). Local herdr jobs only; one (job, dispatch token, attempt) is handled once (job-scoped flock + marker in the job dir); writes go through atomic_dump_yaml; a pin records session_id_source=discovered / session_id_verified=true; a failure leaves the reason on the row and never changes the job's result. - reconcile.sh drift C for agy, hermes, opencode and codex calls the same capture (the agy *.db glob is gone); _pin_and_verify_resume uses pin_row. - Tests: tests/test_issue40_auto_pin.py; issue64 :407 now expects False; the static reconcile-source checks follow the move to the adapters; the two tier2 drift-C component tests create with --no-onboard (onboarding now pins). Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
600 lines
28 KiB
Python
600 lines
28 KiB
Python
"""Issue #40: a post-discovered session id (agy, opencode, codex, hermes) is
|
|
pinned once a local job's `completed` is verified, by the receiving side
|
|
(lib_py.agents.autopin), and agy never records another workspace's - or
|
|
another session's - conversation.
|
|
|
|
Completion conditions:
|
|
1) an unpinned agy row never records the id of a conversation that only
|
|
another workspace updated: test_agy_verify_rejects_id_cached_for_another_workspace,
|
|
test_reconcile_does_not_pin_another_workspaces_agy_conversation,
|
|
test_autopin_does_not_pin_another_workspaces_agy_conversation
|
|
2) the first completed job pins the session's own id, no daemon:
|
|
test_completed_onboarding_pins_agy_without_daemon (+ reconcile, lib.sh and
|
|
create --onboard variants), test_completed_first_job_pins_opencode_codex_hermes
|
|
3) one pin per job/token/attempt; ambiguity pins nothing and leaves a reason:
|
|
test_wrapper_and_loop_recovery_pin_once, test_ambiguous_candidates_leave_reason_without_pin
|
|
4) pre-assigned rows and rows with an id are untouched:
|
|
test_preassigned_and_existing_ids_are_untouched
|
|
|
|
Everything runs against temporary homes, registries and a mock herdr.
|
|
"""
|
|
import json
|
|
import os
|
|
import sqlite3
|
|
import subprocess
|
|
import time
|
|
from pathlib import Path
|
|
|
|
import pytest
|
|
import yaml
|
|
|
|
REPO = Path(__file__).resolve().parents[1]
|
|
SKILLS = REPO / ".agents" / "skills"
|
|
|
|
UA = "a4000001-0000-4000-8000-000000000001"
|
|
UB = "b4000002-0000-4000-8000-000000000002"
|
|
UC = "c4000003-0000-4000-8000-000000000003"
|
|
|
|
|
|
# --------------------------------------------------------------------------- helpers
|
|
|
|
def _agy_conv(home, uuid, mtime=None):
|
|
db = Path(home) / ".gemini" / "antigravity-cli" / "conversations" / f"{uuid}.db"
|
|
db.parent.mkdir(parents=True, exist_ok=True)
|
|
conn = sqlite3.connect(str(db))
|
|
conn.execute("CREATE TABLE steps (id INTEGER)")
|
|
conn.execute("INSERT INTO steps VALUES (1)")
|
|
conn.commit()
|
|
conn.close()
|
|
if mtime is not None:
|
|
os.utime(db, (mtime, mtime))
|
|
return db
|
|
|
|
|
|
def _agy_cache(home, mapping):
|
|
lc = Path(home) / ".gemini" / "antigravity-cli" / "cache" / "last_conversations.json"
|
|
lc.parent.mkdir(parents=True, exist_ok=True)
|
|
lc.write_text(json.dumps({str(k): v for k, v in mapping.items()}))
|
|
|
|
|
|
def _row(name, agent, cwd, epoch, pid=4242, **extra):
|
|
own = {"agy": "agy_conversation_id_own", "opencode": "opencode_session_id_own",
|
|
"codex": "codex_session_id_own", "hermes": "hermes_conversation_id_own",
|
|
"claude": "claude_session_id_own", "grok": "grok_session_id_own"}[agent]
|
|
row = {"name": name, "status": "running", "role": "creator", "agent": agent,
|
|
"herdr_session_epoch": epoch, own: None,
|
|
"session_id_source": "pending-discovery", "session_id_verified": False,
|
|
"last_visible_status": "unverified",
|
|
"pane": {"cwd": str(cwd), "pid": pid, "cmd": agent}}
|
|
row.update(extra)
|
|
return row
|
|
|
|
|
|
def _control_plane(root, rows):
|
|
mam = Path(root) / ".mam"
|
|
(mam / "jobs").mkdir(parents=True, exist_ok=True)
|
|
yaml_path = mam / "agent-sessions.yaml"
|
|
yaml_path.write_text(yaml.safe_dump({"herdr_sessions": rows}, sort_keys=False))
|
|
return yaml_path, mam / "jobs"
|
|
|
|
|
|
def _job(jobs_dir, job_id, session, status="completed", token="tok-1", attempt=1,
|
|
transport="herdr", agent="agy"):
|
|
rec = {"job_id": job_id, "status": status, "agent": agent, "agent_session": session,
|
|
"dispatch_token": token, "attempt_id": attempt, "transport": transport}
|
|
(Path(jobs_dir) / f"{job_id}.json").write_text(json.dumps(rec))
|
|
|
|
|
|
def _rows(yaml_path):
|
|
return {r["name"]: r for r in yaml.safe_load(Path(yaml_path).read_text())["herdr_sessions"]}
|
|
|
|
|
|
def _handle(jobs_dir, job_id, yaml_path, home, **kw):
|
|
from lib_py.agents import autopin
|
|
kw.setdefault("viewport", lambda name, agent, cwd: 0)
|
|
kw.setdefault("resume", lambda name, agent, cwd: "resume_verified")
|
|
return autopin.handle_completed(str(jobs_dir), job_id, yaml_path=str(yaml_path), home_dir=str(home), **kw)
|
|
|
|
|
|
@pytest.fixture
|
|
def owners(monkeypatch):
|
|
"""agy presence ownership without lsof/ps: {uuid: [holder pids]} and a
|
|
pid -> ppid table the test fills in."""
|
|
from lib_py.agents.adapters import agy
|
|
held, parents = {}, {}
|
|
|
|
def holders(paths):
|
|
if held.get("__lsof__") is None and "__lsof__" in held:
|
|
return None
|
|
return sorted({p for path in paths for u, pids in held.items() if u in path for p in pids})
|
|
|
|
monkeypatch.setattr(agy, "_holder_pids", holders)
|
|
monkeypatch.setattr(agy, "_process_parents", lambda: dict(parents) or None)
|
|
return held, parents
|
|
|
|
|
|
@pytest.fixture
|
|
def fake_opencode(tmp_path, monkeypatch):
|
|
"""No real opencode is ever asked for its paths: the adapter falls back
|
|
to <home>/.local/share/opencode/opencode.db."""
|
|
bindir = tmp_path / "fakebin"
|
|
bindir.mkdir(exist_ok=True)
|
|
exe = bindir / "opencode"
|
|
exe.write_text("#!/bin/sh\nexit 1\n")
|
|
exe.chmod(0o755)
|
|
monkeypatch.setenv("PATH", f"{bindir}:{os.environ.get('PATH', '')}")
|
|
monkeypatch.delenv("XDG_DATA_HOME", raising=False)
|
|
return exe
|
|
|
|
|
|
def _seed_class_a(agent, home, ws, ws_other, sid, other_sid, started):
|
|
"""One session of `agent` in ws (sid) and one in another workspace (other_sid), both new."""
|
|
if agent == "opencode":
|
|
db = Path(home) / ".local" / "share" / "opencode" / "opencode.db"
|
|
db.parent.mkdir(parents=True, exist_ok=True)
|
|
conn = sqlite3.connect(str(db))
|
|
conn.execute("CREATE TABLE IF NOT EXISTS session_v2 (id TEXT PRIMARY KEY, parent_id TEXT, directory TEXT, time_created INTEGER)")
|
|
for u, w, t in ((sid, ws, started), (other_sid, ws_other, started + 5)):
|
|
conn.execute("INSERT OR REPLACE INTO session_v2 VALUES (?, NULL, ?, ?)", (u, str(w), int(t * 1000)))
|
|
conn.commit()
|
|
conn.close()
|
|
elif agent == "hermes":
|
|
db = Path(home) / ".hermes" / "state.db"
|
|
db.parent.mkdir(parents=True, exist_ok=True)
|
|
conn = sqlite3.connect(str(db))
|
|
conn.execute("CREATE TABLE IF NOT EXISTS sessions (id TEXT PRIMARY KEY, cwd TEXT, started_at REAL)")
|
|
for u, w, t in ((sid, ws, started), (other_sid, ws_other, started + 5)):
|
|
conn.execute("INSERT OR REPLACE INTO sessions VALUES (?, ?, ?)", (u, str(w), t))
|
|
conn.commit()
|
|
conn.close()
|
|
elif agent == "codex":
|
|
codex = Path(home) / ".codex"
|
|
sess = codex / "sessions" / "2026" / "10" / "20"
|
|
sess.mkdir(parents=True, exist_ok=True)
|
|
conn = sqlite3.connect(str(codex / "state_5.sqlite"))
|
|
conn.execute("CREATE TABLE IF NOT EXISTS threads (id TEXT PRIMARY KEY, cwd TEXT, rollout_path TEXT, created_at INTEGER, updated_at INTEGER)")
|
|
for u, w, t in ((sid, ws, started), (other_sid, ws_other, started + 5)):
|
|
rollout = sess / f"rollout-2026-10-20T00-00-00-{u}.jsonl"
|
|
rollout.write_text(json.dumps({"payload": {"id": u, "cwd": str(w)}}) + "\n")
|
|
conn.execute("INSERT OR REPLACE INTO threads VALUES (?, ?, ?, ?, ?)", (u, str(w), str(rollout), int(t), int(t)))
|
|
conn.commit()
|
|
conn.close()
|
|
|
|
|
|
# --------------------------------------------------------------------------- adapter contract
|
|
|
|
def test_adapter_id_lifecycle_classification():
|
|
from lib_py.agents.base import POST_DISCOVERED, PRE_ASSIGNED
|
|
from lib_py.agents.registry import get_adapter
|
|
assert {a: get_adapter(a).id_lifecycle for a in ("agy", "opencode", "codex", "hermes", "claude", "grok")} == {
|
|
"agy": POST_DISCOVERED, "opencode": POST_DISCOVERED, "codex": POST_DISCOVERED,
|
|
"hermes": POST_DISCOVERED, "claude": PRE_ASSIGNED, "grok": PRE_ASSIGNED}
|
|
|
|
|
|
# --------------------------------------------------------------------------- 1) another workspace's agy id
|
|
|
|
def test_agy_verify_rejects_id_cached_for_another_workspace(tmp_path):
|
|
from lib_py.verify_session import verify_session_uuid
|
|
home, ws_a, ws_b = tmp_path / "home", tmp_path / "ws_a", tmp_path / "ws_b"
|
|
ws_a.mkdir()
|
|
ws_b.mkdir()
|
|
epoch = int(time.time()) - 100
|
|
_agy_conv(home, UB)
|
|
_agy_cache(home, {ws_b: UB})
|
|
row_a = _row("ws-a-creator-agy", "agy", ws_a, epoch)
|
|
assert verify_session_uuid(str(ws_a), "agy", UB, row=row_a, home_dir=str(home), mode="discover") is False
|
|
# the same id is fine for the workspace the cache names
|
|
row_b = _row("ws-b-creator-agy", "agy", ws_b, epoch)
|
|
assert verify_session_uuid(str(ws_b), "agy", UB, row=row_b, home_dir=str(home), mode="discover") is True
|
|
|
|
|
|
def _mock_agent(state_file, name, cwd, pid, epoch):
|
|
state = json.loads(Path(state_file).read_text())
|
|
state.setdefault("agents", {})[name] = {"pid": pid, "cwd": str(cwd), "command": "agy", "agent": "agy",
|
|
"created": epoch, "buffer": f"Antigravity {cwd}\n>"}
|
|
Path(state_file).write_text(json.dumps(state))
|
|
|
|
|
|
def _reconcile(sandbox):
|
|
script = sandbox / ".agents" / "skills" / "multi-agent-mux-monitor" / "scripts" / "reconcile.sh"
|
|
res = subprocess.run(["bash", str(script), "--once", "--emit-diff"], capture_output=True, text=True,
|
|
cwd=str(sandbox), env={**os.environ, "WORKSPACE_ROOT": str(sandbox)}, timeout=120)
|
|
assert res.returncode == 0, res.stderr
|
|
return res
|
|
|
|
|
|
def test_reconcile_does_not_pin_another_workspaces_agy_conversation(mam_sandbox, mock_herdr):
|
|
"""The issue's example: proj-A's agy session is unpinned, only proj-B's
|
|
conversation changed after A started. reconcile used to glob every *.db
|
|
and record B's id on A."""
|
|
ws_a, ws_b = mam_sandbox / "proj-a", mam_sandbox / "proj-b"
|
|
ws_a.mkdir()
|
|
ws_b.mkdir()
|
|
epoch = int(time.time()) - 100
|
|
_agy_conv(mam_sandbox, UB)
|
|
_agy_cache(mam_sandbox, {ws_b: UB})
|
|
name = "proj-a-creator-agy"
|
|
yaml_path = mam_sandbox / ".mam" / "agent-sessions.yaml"
|
|
yaml_path.write_text(yaml.safe_dump({"herdr_sessions": [_row(name, "agy", ws_a, epoch, pid=os.getpid())]}))
|
|
_mock_agent(mock_herdr, name, ws_a, os.getpid(), epoch)
|
|
|
|
_reconcile(mam_sandbox)
|
|
row = _rows(yaml_path)[name]
|
|
assert row["status"] == "running"
|
|
assert row["agy_conversation_id_own"] is None
|
|
|
|
|
|
def test_autopin_does_not_pin_another_workspaces_agy_conversation(tmp_path, owners):
|
|
home, ws_a, ws_b = tmp_path / "home", tmp_path / "ws_a", tmp_path / "ws_b"
|
|
ws_a.mkdir()
|
|
ws_b.mkdir()
|
|
epoch = int(time.time()) - 100
|
|
_agy_conv(home, UB)
|
|
_agy_cache(home, {ws_b: UB})
|
|
held, _ = owners
|
|
held[UB] = [4242] # even a holder that looks like A's pane does not make B's id A's
|
|
yaml_path, jobs = _control_plane(tmp_path / "ctl", [_row("a-agy", "agy", ws_a, epoch, pid=4242)])
|
|
_job(jobs, "j1", "herdr:a-agy")
|
|
res = _handle(jobs, "j1", yaml_path, home)
|
|
assert res["status"] == "no-candidate"
|
|
row = _rows(yaml_path)["a-agy"]
|
|
assert row["agy_conversation_id_own"] is None
|
|
assert "autopin no-candidate (job j1)" in row["last_visible_status"]
|
|
|
|
|
|
# --------------------------------------------------------------------------- 2) first completed job pins
|
|
|
|
def test_completed_onboarding_pins_agy_without_daemon(tmp_path, owners, monkeypatch):
|
|
"""A's onboarding completed; B's conversation was updated meanwhile (and
|
|
later than A's). A gets A's id, owned by A's pane."""
|
|
monkeypatch.setenv("MAM_NO_SUBSCRIBER_DAEMON", "1")
|
|
home, ws_a, ws_b = tmp_path / "home", tmp_path / "ws_a", tmp_path / "ws_b"
|
|
ws_a.mkdir()
|
|
ws_b.mkdir()
|
|
now = time.time()
|
|
epoch = int(now) - 100
|
|
_agy_conv(home, UA, mtime=now - 10)
|
|
_agy_conv(home, UB, mtime=now)
|
|
_agy_cache(home, {ws_a: UA, ws_b: UB})
|
|
held, _ = owners
|
|
held[UA], held[UB] = [5001], [6001]
|
|
yaml_path, jobs = _control_plane(tmp_path / "ctl", [_row("a-agy", "agy", ws_a, epoch, pid=5001)])
|
|
_job(jobs, "onb1", "herdr:a-agy")
|
|
|
|
res = _handle(jobs, "onb1", yaml_path, home)
|
|
assert res["status"] == "pinned", res
|
|
row = _rows(yaml_path)["a-agy"]
|
|
assert row["agy_conversation_id_own"] == UA
|
|
assert row["session_id_source"] == "discovered"
|
|
assert row["session_id_verified"] is True
|
|
assert row["last_visible_status"] == "resume_verified"
|
|
|
|
|
|
@pytest.mark.parametrize("agent,sid,other", [("opencode", "ses_own_a", "ses_other_b"),
|
|
("codex", UA, UB), ("hermes", "h-own-a", "h-other-b")])
|
|
def test_completed_first_job_pins_opencode_codex_hermes(tmp_path, monkeypatch, fake_opencode, agent, sid, other):
|
|
home, ws, ws_other = tmp_path / "home", tmp_path / "ws", tmp_path / "ws_other"
|
|
ws.mkdir()
|
|
ws_other.mkdir()
|
|
monkeypatch.setenv("HOME", str(home))
|
|
epoch = int(time.time()) - 100
|
|
_seed_class_a(agent, home, ws, ws_other, sid, other, started=epoch + 10)
|
|
yaml_path, jobs = _control_plane(tmp_path / "ctl", [_row(f"w-creator-{agent}", agent, ws, epoch)])
|
|
_job(jobs, "first", f"herdr:w-creator-{agent}", agent=agent)
|
|
|
|
res = _handle(jobs, "first", yaml_path, home)
|
|
assert res["status"] == "pinned", res
|
|
row = _rows(yaml_path)[f"w-creator-{agent}"]
|
|
from lib_py.agents.registry import own_key
|
|
assert row[own_key(agent)] == sid
|
|
assert (row["session_id_source"], row["session_id_verified"]) == ("discovered", True)
|
|
|
|
|
|
def _fake_lsof(bindir, uuid, pid):
|
|
exe = Path(bindir) / "lsof"
|
|
exe.write_text(f'#!/bin/sh\nfor a in "$@"; do case "$a" in *{uuid}*) echo {pid}; exit 0;; esac; done\nexit 1\n')
|
|
exe.chmod(0o755)
|
|
|
|
|
|
def test_reconcile_pins_agy_owned_by_the_pane(mam_sandbox, mock_herdr):
|
|
"""drift C uses the same capture: A's cached id, held by A's pane, is pinned;
|
|
B's newer conversation in another workspace is not considered."""
|
|
ws_a, ws_b = mam_sandbox / "proj-a", mam_sandbox / "proj-b"
|
|
ws_a.mkdir()
|
|
ws_b.mkdir()
|
|
now = time.time()
|
|
epoch = int(now) - 100
|
|
_agy_conv(mam_sandbox, UA, mtime=now - 10)
|
|
_agy_conv(mam_sandbox, UB, mtime=now)
|
|
_agy_cache(mam_sandbox, {ws_a: UA, ws_b: UB})
|
|
_fake_lsof(mam_sandbox / "bin", UA, os.getpid())
|
|
name = "proj-a-creator-agy"
|
|
yaml_path = mam_sandbox / ".mam" / "agent-sessions.yaml"
|
|
yaml_path.write_text(yaml.safe_dump({"herdr_sessions": [_row(name, "agy", ws_a, epoch, pid=os.getpid())]}))
|
|
_mock_agent(mock_herdr, name, ws_a, os.getpid(), epoch)
|
|
|
|
_reconcile(mam_sandbox)
|
|
row = _rows(yaml_path)[name]
|
|
assert row["agy_conversation_id_own"] == UA
|
|
assert (row["session_id_source"], row["session_id_verified"]) == ("discovered", True)
|
|
|
|
|
|
def test_lib_sh_helper_pins_through_the_cli(mam_sandbox, mock_herdr, fake_opencode):
|
|
"""The wrapper's and create's entry point: lib.sh delegate_autopin_completed
|
|
runs `python -m lib_py.agents.autopin` against the sandbox control plane."""
|
|
ws, ws_other = mam_sandbox / "proj", mam_sandbox / "other"
|
|
ws.mkdir()
|
|
ws_other.mkdir()
|
|
epoch = int(time.time()) - 100
|
|
_seed_class_a("opencode", mam_sandbox, ws, ws_other, "ses_own", "ses_other", started=epoch + 10)
|
|
yaml_path = mam_sandbox / ".mam" / "agent-sessions.yaml"
|
|
name = "proj-creator-opencode"
|
|
yaml_path.write_text(yaml.safe_dump({"herdr_sessions": [_row(name, "opencode", ws, epoch)]}))
|
|
jobs = mam_sandbox / ".mam" / "jobs"
|
|
jobs.mkdir(parents=True, exist_ok=True)
|
|
_job(jobs, "jcli", f"herdr:{name}", agent="opencode")
|
|
|
|
lib_sh = mam_sandbox / ".agents" / "skills" / "lib.sh"
|
|
res = subprocess.run(["bash", "-c", f'source "{lib_sh}" && delegate_autopin_completed jcli tok-1 "{jobs}"'],
|
|
capture_output=True, text=True, env=dict(os.environ), timeout=120)
|
|
assert res.returncode == 0, res.stderr
|
|
row = _rows(yaml_path)[name]
|
|
assert row["opencode_session_id_own"] == "ses_own", res.stderr
|
|
marker = json.loads((jobs / "jcli" / "autopin.json").read_text())
|
|
assert marker["key"] == "jcli:tok-1:1" and marker["result"]["status"] == "pinned"
|
|
# a stale dispatch token is not this attempt's completion
|
|
_job(jobs, "jstale", f"herdr:{name}", agent="opencode", token="tok-new")
|
|
subprocess.run(["bash", "-c", f'source "{lib_sh}" && delegate_autopin_completed jstale tok-old "{jobs}"'],
|
|
capture_output=True, text=True, env=dict(os.environ), timeout=120)
|
|
assert not (jobs / "jstale" / "autopin.json").exists()
|
|
|
|
|
|
def test_create_onboard_pins_without_daemon(mam_sandbox, mock_herdr, mock_agents):
|
|
"""create --onboard with MAM_NO_SUBSCRIBER_DAEMON=1 (mam_sandbox): the
|
|
mock agent completes the onboarding and create runs the handler itself."""
|
|
create = mam_sandbox / ".agents" / "skills" / "multi-agent-mux-create" / "scripts" / "create_session.sh"
|
|
res = subprocess.run(["bash", str(create), "--workspace", str(mam_sandbox), "--agent", "opencode",
|
|
"--role", "creator", "--onboard"],
|
|
capture_output=True, text=True, cwd=str(mam_sandbox), timeout=300)
|
|
assert res.returncode == 0, res.stderr
|
|
yaml_path = mam_sandbox / ".mam" / "agent-sessions.yaml"
|
|
(row,) = _rows(yaml_path).values()
|
|
job = row["delegate_job_id"]
|
|
marker = json.loads((mam_sandbox / ".mam" / "jobs" / job / "autopin.json").read_text())
|
|
# create's async reconcile may get there first; either way the handler ran
|
|
assert marker["result"]["status"] in ("pinned", "already-set"), marker
|
|
assert row["opencode_session_id_own"]
|
|
assert (row["session_id_source"], row["session_id_verified"]) == ("discovered", True)
|
|
|
|
|
|
# --------------------------------------------------------------------------- agy: the cwd cache proves nothing
|
|
|
|
def test_agy_same_cwd_sibling_never_cross_pins(tmp_path, owners):
|
|
"""Two agy sessions in one cwd; the cache names U, whose presence lock A's
|
|
pane holds. Only A may get U - before and after A is pinned."""
|
|
from lib_py.verify_session import verify_session_uuid
|
|
home, ws = tmp_path / "home", tmp_path / "ws"
|
|
ws.mkdir()
|
|
epoch = int(time.time()) - 100
|
|
_agy_conv(home, UA)
|
|
_agy_cache(home, {ws: UA})
|
|
held, parents = owners
|
|
held[UA] = [5001]
|
|
parents.update({5001: 1, 6001: 1})
|
|
yaml_path, jobs = _control_plane(tmp_path / "ctl", [_row("a-agy", "agy", ws, epoch, pid=5001),
|
|
_row("b-agy", "agy", ws, epoch, pid=6001)])
|
|
_job(jobs, "jb1", "herdr:b-agy")
|
|
assert _handle(jobs, "jb1", yaml_path, home)["status"] == "owner-mismatch"
|
|
_job(jobs, "ja1", "herdr:a-agy")
|
|
assert _handle(jobs, "ja1", yaml_path, home)["status"] == "pinned"
|
|
_job(jobs, "jb2", "herdr:b-agy")
|
|
assert _handle(jobs, "jb2", yaml_path, home)["status"] == "sibling-claimed"
|
|
rows = _rows(yaml_path)
|
|
assert rows["a-agy"]["agy_conversation_id_own"] == UA
|
|
assert rows["b-agy"]["agy_conversation_id_own"] is None
|
|
# a cache hit no longer bypasses the sibling exclusion
|
|
row_b = dict(rows["b-agy"], _sibling_claimed_uuids=[UA])
|
|
assert verify_session_uuid(str(ws), "agy", UA, row=row_b, home_dir=str(home), mode="discover") is False
|
|
|
|
|
|
@pytest.mark.parametrize("holders,parents,expected", [
|
|
(None, {}, "unverifiable-owner"), # lsof missing or failed
|
|
([], {}, "unverifiable-owner"), # nobody holds the lock
|
|
([7001], {7001: 1, 5001: 1}, "owner-mismatch"), # held outside the pane's tree
|
|
([7001], {7001: 7000, 7000: 5001, 5001: 1}, "captured"), # a descendant of the pane
|
|
])
|
|
def test_agy_capture_requires_presence_owned_by_pane(tmp_path, owners, holders, parents, expected):
|
|
from lib_py.agents.autopin import capture
|
|
home, ws = tmp_path / "home", tmp_path / "ws"
|
|
ws.mkdir()
|
|
epoch = int(time.time()) - 100
|
|
_agy_conv(home, UA)
|
|
_agy_cache(home, {ws: UA})
|
|
held, ppid = owners
|
|
if holders is None:
|
|
held["__lsof__"] = None
|
|
else:
|
|
held[UA] = holders
|
|
ppid.update(parents)
|
|
row = _row("a-agy", "agy", ws, epoch, pid=5001)
|
|
res = capture(row, "agy", {"herdr_sessions": [row]}, home_dir=str(home))
|
|
assert res["status"] == expected, res
|
|
assert (res["session_id"] == UA) == (expected == "captured")
|
|
|
|
|
|
def test_agy_holder_pids_reads_lsof(tmp_path, monkeypatch):
|
|
"""The real helper: lsof's pids, [] when it finds no holder, None without lsof."""
|
|
from lib_py.agents.adapters import agy
|
|
f = tmp_path / "x.lock"
|
|
f.write_text("")
|
|
bindir = tmp_path / "bin"
|
|
bindir.mkdir()
|
|
monkeypatch.setenv("PATH", str(bindir))
|
|
assert agy._holder_pids([str(f)]) is None
|
|
_fake_lsof(bindir, "x.lock", 4321)
|
|
assert agy._holder_pids([str(f)]) == [4321]
|
|
assert agy._holder_pids([str(tmp_path / "missing.lock")]) == []
|
|
(bindir / "lsof").write_text("#!/bin/sh\nexit 1\n")
|
|
assert agy._holder_pids([str(f)]) == []
|
|
|
|
|
|
# --------------------------------------------------------------------------- 3) once per job, ambiguity
|
|
|
|
def _count_captures(monkeypatch):
|
|
from lib_py.agents import autopin
|
|
calls = []
|
|
real = autopin.capture
|
|
|
|
def counting(*a, **kw):
|
|
calls.append(a[0].get("name"))
|
|
return real(*a, **kw)
|
|
|
|
monkeypatch.setattr(autopin, "capture", counting)
|
|
return calls
|
|
|
|
|
|
def _loop_runner(mam_dir, workspace_root=REPO):
|
|
from lib_py.loop.runner import LoopRunner
|
|
r = LoopRunner.__new__(LoopRunner)
|
|
r.workspace_root = Path(workspace_root)
|
|
r.mam_dir, r.jobs_dir = Path(mam_dir), Path(mam_dir) / "jobs"
|
|
r.failed_submit_jobs, r._stall_seen = set(), {}
|
|
r.verbose, r.interrupted = False, False
|
|
return r
|
|
|
|
|
|
def test_wrapper_and_loop_recovery_pin_once(tmp_path, monkeypatch, fake_opencode):
|
|
home, ws, ws_other = tmp_path / "home", tmp_path / "ws", tmp_path / "ws_other"
|
|
ws.mkdir()
|
|
ws_other.mkdir()
|
|
epoch = int(time.time()) - 100
|
|
_seed_class_a("opencode", home, ws, ws_other, "ses_own", "ses_other", started=epoch + 10)
|
|
yaml_path, jobs = _control_plane(tmp_path / "ctl", [_row("w-opencode", "opencode", ws, epoch)])
|
|
monkeypatch.setenv("AGENT_SESSIONS_YAML", str(yaml_path))
|
|
monkeypatch.setenv("HOME_DIR", str(home))
|
|
_job(jobs, "dup", "herdr:w-opencode", agent="opencode")
|
|
calls = _count_captures(monkeypatch)
|
|
|
|
first = _handle(jobs, "dup", yaml_path, home) # the wrapper's subscriber terminal
|
|
assert first["status"] == "pinned"
|
|
written = yaml_path.stat().st_mtime_ns
|
|
marker = (jobs / "dup" / "autopin.json").read_text()
|
|
# loop recovery (a subprocess of the same handler): deduplicated, nothing rewritten
|
|
assert _loop_runner(tmp_path / "ctl" / ".mam").wait_for_job("dup", max_wait=5) is True
|
|
assert (jobs / "dup" / "autopin.json").read_text() == marker
|
|
again = _handle(jobs, "dup", yaml_path, home)
|
|
assert again.get("deduplicated") is True and again["status"] == "pinned"
|
|
assert calls == ["w-opencode"]
|
|
assert yaml_path.stat().st_mtime_ns == written
|
|
assert _rows(yaml_path)["w-opencode"]["opencode_session_id_own"] == "ses_own"
|
|
|
|
|
|
def test_loop_recovery_pins_when_the_wrapper_did_not(mam_sandbox, mock_herdr, fake_opencode):
|
|
"""wait_for_job's recovery call alone pins the id (the wrapper's call was missed)."""
|
|
ws, ws_other = mam_sandbox / "proj", mam_sandbox / "other"
|
|
ws.mkdir()
|
|
ws_other.mkdir()
|
|
epoch = int(time.time()) - 100
|
|
_seed_class_a("opencode", mam_sandbox, ws, ws_other, "ses_own", "ses_other", started=epoch + 10)
|
|
yaml_path = mam_sandbox / ".mam" / "agent-sessions.yaml"
|
|
name = "proj-creator-opencode"
|
|
yaml_path.write_text(yaml.safe_dump({"herdr_sessions": [_row(name, "opencode", ws, epoch)]}))
|
|
jobs = mam_sandbox / ".mam" / "jobs"
|
|
jobs.mkdir(parents=True, exist_ok=True)
|
|
_job(jobs, "jloop", f"herdr:{name}", agent="opencode")
|
|
|
|
assert _loop_runner(mam_sandbox / ".mam", workspace_root=mam_sandbox).wait_for_job("jloop", max_wait=5) is True
|
|
row = _rows(yaml_path)[name]
|
|
assert row["opencode_session_id_own"] == "ses_own"
|
|
assert json.loads((jobs / "jloop" / "autopin.json").read_text())["result"]["status"] == "pinned"
|
|
|
|
|
|
def test_ambiguous_candidates_leave_reason_without_pin(tmp_path, monkeypatch, fake_opencode):
|
|
home, ws = tmp_path / "home", tmp_path / "ws"
|
|
ws.mkdir()
|
|
epoch = int(time.time()) - 100
|
|
_seed_class_a("opencode", home, ws, ws, "ses_one", "ses_two", started=epoch + 10) # both in ws
|
|
yaml_path, jobs = _control_plane(tmp_path / "ctl", [_row("w-opencode", "opencode", ws, epoch)])
|
|
_job(jobs, "amb", "herdr:w-opencode", agent="opencode")
|
|
calls = _count_captures(monkeypatch)
|
|
|
|
res = _handle(jobs, "amb", yaml_path, home)
|
|
assert res["status"] == "ambiguous" and sorted(res["candidates"]) == ["ses_one", "ses_two"]
|
|
row = _rows(yaml_path)["w-opencode"]
|
|
assert row["opencode_session_id_own"] is None
|
|
assert row["last_visible_status"].startswith("autopin ambiguous (job amb): 2 candidates")
|
|
assert _handle(jobs, "amb", yaml_path, home).get("deduplicated") is True
|
|
assert calls == ["w-opencode"]
|
|
|
|
|
|
# --------------------------------------------------------------------------- 4) untouched rows, skipped jobs
|
|
|
|
def test_preassigned_and_existing_ids_are_untouched(tmp_path, owners, fake_opencode):
|
|
home, ws = tmp_path / "home", tmp_path / "ws"
|
|
ws.mkdir()
|
|
epoch = int(time.time()) - 100
|
|
_agy_conv(home, UB)
|
|
_agy_cache(home, {ws: UB})
|
|
held, _ = owners
|
|
held[UB] = [5003]
|
|
_seed_class_a("opencode", home, ws, ws, "ses_x", "ses_y", started=epoch + 10)
|
|
rows = [
|
|
_row("c-claude", "claude", ws, epoch, claude_session_id_own=UA, session_id_source="assigned"),
|
|
_row("g-grok", "grok", ws, epoch, grok_session_id_own=UC, session_id_source="assigned"),
|
|
_row("p-agy", "agy", ws, epoch, pid=5003, agy_conversation_id_own="agy-pinned-id",
|
|
session_id_source="discovered", session_id_verified=True, last_visible_status="resume_verified"),
|
|
_row("o-opencode", "opencode", ws, epoch, opencode_session_id_own="ses_kept"),
|
|
]
|
|
yaml_path, jobs = _control_plane(tmp_path / "ctl", rows)
|
|
before = _rows(yaml_path)
|
|
for i, r in enumerate(rows):
|
|
_job(jobs, f"k{i}", f"herdr:{r['name']}", agent=r["agent"])
|
|
assert _handle(jobs, f"k{i}", yaml_path, home)["status"] in ("pre-assigned", "already-set")
|
|
assert _rows(yaml_path) == before
|
|
|
|
|
|
@pytest.mark.parametrize("change,reason", [
|
|
({"transport": "remote", "session": "remote:box"}, "not a local agent session"),
|
|
({"agent": "human", "session": "human"}, "not a local agent session"),
|
|
({"status": "error"}, "job j is error"),
|
|
({"status": "running"}, "job j is running"),
|
|
])
|
|
def test_ineligible_jobs_pin_nothing(tmp_path, owners, change, reason):
|
|
home, ws = tmp_path / "home", tmp_path / "ws"
|
|
ws.mkdir()
|
|
epoch = int(time.time()) - 100
|
|
_agy_conv(home, UA)
|
|
_agy_cache(home, {ws: UA})
|
|
owners[0][UA] = [5001]
|
|
yaml_path, jobs = _control_plane(tmp_path / "ctl", [_row("a-agy", "agy", ws, epoch, pid=5001)])
|
|
_job(jobs, "j", change.get("session", "herdr:a-agy"), status=change.get("status", "completed"),
|
|
transport=change.get("transport", "herdr"), agent=change.get("agent", "agy"))
|
|
res = _handle(jobs, "j", yaml_path, home)
|
|
assert (res["status"], res["reason"]) == ("skipped", reason)
|
|
assert _rows(yaml_path)["a-agy"]["agy_conversation_id_own"] is None
|
|
|
|
|
|
def test_stale_dispatch_token_pins_nothing(tmp_path, owners):
|
|
home, ws = tmp_path / "home", tmp_path / "ws"
|
|
ws.mkdir()
|
|
_agy_conv(home, UA)
|
|
_agy_cache(home, {ws: UA})
|
|
owners[0][UA] = [5001]
|
|
yaml_path, jobs = _control_plane(tmp_path / "ctl", [_row("a-agy", "agy", ws, int(time.time()) - 100, pid=5001)])
|
|
_job(jobs, "j", "herdr:a-agy", token="tok-2")
|
|
res = _handle(jobs, "j", yaml_path, home, expect_token="tok-1")
|
|
assert (res["status"], res["reason"]) == ("skipped", "stale dispatch token")
|
|
assert not (jobs / "j" / "autopin.json").exists()
|
|
|
|
|
|
def test_call_sites_and_publisher_boundary():
|
|
"""The wrapper's direct path and create --onboard call the handler on
|
|
`completed`; the agent-run publisher never touches the session registry."""
|
|
wrapper = (SKILLS / "multi-agent-mux-delegate-job" / "multi-agent-mux-delegate-job").read_text()
|
|
assert 'delegate_autopin_completed "$JOB_ID" "$_disp_tok" "$REGISTRY_DIR"' in wrapper
|
|
create = (SKILLS / "multi-agent-mux-create" / "scripts" / "create_session.sh").read_text()
|
|
assert 'delegate_autopin_completed "$DELEGATE_JOB_ID" "$DELEGATE_JOB_TOKEN"' in create
|
|
for f in ("publish_event.py", "publisher.py"):
|
|
src = (SKILLS / "lib_py" / "delegation" / f).read_text()
|
|
assert "atomic_dump_yaml" not in src and "agent-sessions" not in src and "autopin" not in src
|