From feccfd5e0356c8acf0438bc68bce7ea36ce2588e Mon Sep 17 00:00:00 2001 From: Godopu Date: Tue, 20 Oct 2026 23:40:03 +0900 Subject: [PATCH] fix(agents): pin post-discovered session ids on a verified completed; agy never takes another workspace's or session's conversation (Issue #40) - 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 --- .agents/skills/lib.sh | 12 + .agents/skills/lib_py/agents/adapters/agy.py | 129 +++- .../skills/lib_py/agents/adapters/claude.py | 7 +- .agents/skills/lib_py/agents/adapters/grok.py | 7 +- .agents/skills/lib_py/agents/autopin.py | 261 ++++++++ .agents/skills/lib_py/agents/base.py | 41 ++ .agents/skills/lib_py/loop/runner.py | 18 + .../skills/multi-agent-mux-create/SKILL.md | 1 + .../scripts/create_session.sh | 3 + .../multi-agent-mux-delegate-job | 5 + .../skills/multi-agent-mux-monitor/SKILL.md | 2 + .../scripts/reconcile.sh | 328 +--------- docs/NEW_AGENT_INTEGRATION_GUIDE.md | 14 +- tests/test_issue40_auto_pin.py | 599 ++++++++++++++++++ tests/test_issue60_linux_fixes.py | 2 +- tests/test_issue60_opencode_v2.py | 3 +- tests/test_issue64_ro_connect.py | 21 +- tests/test_tier1_unit.py | 3 +- tests/test_tier2_component.py | 10 +- 19 files changed, 1128 insertions(+), 338 deletions(-) create mode 100644 .agents/skills/lib_py/agents/autopin.py create mode 100644 tests/test_issue40_auto_pin.py diff --git a/.agents/skills/lib.sh b/.agents/skills/lib.sh index c8ab38b..91d8617 100644 --- a/.agents/skills/lib.sh +++ b/.agents/skills/lib.sh @@ -2467,6 +2467,18 @@ delegate_wait_terminal() { --job "$job_id" --registry-dir "$reg_dir" --timeout "$timeout_s" --watch-pid "$$" } +# delegate_autopin_completed [dispatch_token] [registry_dir] +# Issue #40: once this side has seen the job `completed`, pin the session's +# post-discovered id (lib_py.agents.autopin decides and writes; remote, human +# and pre-assigned sessions are left alone, one job/token/attempt is handled +# once). Reports on stderr; never changes the job's result or this exit code. +delegate_autopin_completed() { + local job_id="$1" token="${2:-}" reg_dir="${3:-${MAM_JOBS_DIR:-${WORKSPACE_ROOT:-$PWD}/.mam/jobs}}" + PYTHONPATH="$SKILL_DIR:${PYTHONPATH:-}" HOME_DIR="${HOME_DIR:-$HOME}" LIB_SH="$SKILL_DIR/lib.sh" \ + "$(_delegate_py_bin)" -m lib_py.agents.autopin --job "$job_id" --registry-dir "$reg_dir" \ + --yaml "$AGENT_SESSIONS_YAML" ${token:+--expect-dispatch-token "$token"} >&2 || true +} + # delegate_onboard_exit # Issue #50: create's exit code for an onboarding status (0, 44 or 45). delegate_onboard_exit() { diff --git a/.agents/skills/lib_py/agents/adapters/agy.py b/.agents/skills/lib_py/agents/adapters/agy.py index 5ce7790..febdef9 100644 --- a/.agents/skills/lib_py/agents/adapters/agy.py +++ b/.agents/skills/lib_py/agents/adapters/agy.py @@ -1,6 +1,43 @@ -import os, sys, json, shutil -from typing import Optional, Any -from lib_py.agents.base import BaseAgentAdapter, DiscoveryContext, _ro_connect, conversation_in_use +import os, sys, json, shutil, subprocess +from typing import Optional, Any, Dict, List +from lib_py.agents.base import BaseAgentAdapter, DiscoveryContext, _ro_connect, capture_result, conversation_in_use +from lib_py.verify_session import verify_session_uuid, workspace_key + + +def _holder_pids(paths: List[str]) -> Optional[List[int]]: + """PIDs holding any of `paths` open (lsof -t), [] when none does, None + when lsof cannot answer (missing, failed, timed out).""" + existing = [p for p in paths if p and os.path.exists(p)] + if not existing: + return [] + lsof = shutil.which("lsof") + if not lsof: + return None + try: + r = subprocess.run([lsof, "-t"] + existing, capture_output=True, text=True, timeout=10) + except Exception: + return None + pids = sorted({int(t) for t in r.stdout.split() if t.isdigit()}) + # lsof exits 1 both when nothing holds the files and on errors. + if r.returncode not in (0, 1) or (r.returncode == 1 and not pids and r.stderr.strip()): + return None + return pids + + +def _process_parents() -> Optional[Dict[int, int]]: + """pid -> ppid of every running process (ps), None when ps fails.""" + try: + out = subprocess.run(["ps", "-eo", "pid=,ppid="], capture_output=True, text=True, timeout=10) + except Exception: + return None + if out.returncode != 0: + return None + parents: Dict[int, int] = {} + for line in out.stdout.splitlines(): + parts = line.split() + if len(parts) == 2 and parts[0].isdigit() and parts[1].isdigit(): + parents[int(parts[0])] = int(parts[1]) + return parents or None class AgyAgentAdapter(BaseAgentAdapter): @property @@ -58,18 +95,15 @@ class AgyAgentAdapter(BaseAgentAdapter): if ctx.epoch and os.path.getmtime(path) < ctx.epoch: return False if ctx.mode == "discover": - lc = f"{ctx.home_dir}/.gemini/antigravity-cli/cache/last_conversations.json" - cache_match = False - if os.path.exists(lc): - try: - with open(lc) as f: - lc_data = json.load(f) - cache_match = (lc_data.get(ctx.cwd) == uuid) - except Exception: - cache_match = False - if not cache_match: - if uuid in (ctx.row.get("_sibling_claimed_uuids") or []): - return False + # Issue #40: the cache maps each workspace to the conversation last + # opened there, so an id it maps to another workspace is that + # workspace's; and an id another row records is never this row's, + # whether or not the cache names it for this cwd. + here = workspace_key(ctx.cwd) + if any(v == uuid and workspace_key(k) != here for k, v in self._last_conversations(ctx).items()): + return False + if uuid in (ctx.row.get("_sibling_claimed_uuids") or []): + return False try: conn = _ro_connect(path) r = conn.execute("SELECT count(*) FROM steps").fetchone() @@ -83,6 +117,71 @@ class AgyAgentAdapter(BaseAgentAdapter): def _cli_root(self, ctx: DiscoveryContext) -> str: return f"{ctx.home_dir}/.gemini/antigravity-cli" + def _last_conversations(self, ctx: DiscoveryContext) -> dict: + """cache/last_conversations.json: {cwd: id of the conversation last opened there}.""" + try: + with open(f"{self._cli_root(ctx)}/cache/last_conversations.json") as f: + data = json.load(f) + return data if isinstance(data, dict) else {} + except Exception: + return {} + + def capture_active_session_id(self, ctx: DiscoveryContext) -> dict: + """Issue #40: the cache only proposes an id - it holds one id per cwd, + rewritten whenever any agy opens a conversation there, so it cannot tell + two sessions in one cwd apart. The id is this session's only when its + presence lock (or DB) is held open by the row's pane process or one of + its descendants. Anything unprovable is not pinned (no cache-only fallback).""" + here = workspace_key(ctx.cwd) + proposed = sorted({v for k, v in self._last_conversations(ctx).items() + if v and workspace_key(k) == here}) + if not proposed: + return capture_result("no-candidate", reason="last_conversations.json names no conversation for this cwd") + if len(proposed) > 1: + return capture_result("ambiguous", reason="last_conversations.json maps this cwd to several conversations", + candidates=proposed) + uuid = proposed[0] + if uuid in (ctx.row.get("_sibling_claimed_uuids") or []): + return capture_result("sibling-claimed", reason=f"{uuid} is recorded by another session", candidates=[uuid]) + if not verify_session_uuid(ctx.cwd, self.name, uuid, row=ctx.row, home_dir=ctx.home_dir, mode="discover"): + return capture_result("rejected", reason=f"{uuid} failed verification (missing, older than the session, or empty)", + candidates=[uuid]) + status, reason = self._owner_check(uuid, ctx) + if status != "ok": + return capture_result(status, reason=reason, candidates=[uuid]) + return capture_result("captured", session_id=uuid, candidates=[uuid]) + + def _owner_check(self, uuid: str, ctx: DiscoveryContext): + """('ok', '') when a holder of the conversation's presence lock or DB is + the row's pane process or its descendant; otherwise + ('unverifiable-owner' | 'owner-mismatch', reason).""" + try: + pane_pid = int((ctx.row.get("pane") or {}).get("pid") or 0) + except (TypeError, ValueError): + pane_pid = 0 + if pane_pid <= 1: + return "unverifiable-owner", "the row records no pane pid" + db = self.artifact_path(uuid, ctx) + paths = [f"{self._cli_root(ctx)}/presence/{uuid}.lock", db, f"{db}-wal", f"{db}-shm"] + holders = _holder_pids(paths) + if holders is None: + return "unverifiable-owner", "lsof is unavailable or failed" + if not holders: + return "unverifiable-owner", f"no process holds {uuid}'s presence lock or DB" + if pane_pid in holders: + return "ok", "" + parents = _process_parents() + if parents is None or pane_pid not in parents: + return "unverifiable-owner", f"pane process {pane_pid} cannot be found" + for pid in holders: + seen = set() + while pid > 1 and pid not in seen: + if pid == pane_pid: + return "ok", "" + seen.add(pid) + pid = parents.get(pid, 0) + return "owner-mismatch", f"{uuid} is held by pid(s) {holders}, none in pane process {pane_pid}'s tree" + def _artifact_paths(self, uuid: str, ctx: DiscoveryContext) -> list: """Every on-disk piece of one conversation, DB first: the SQLite sidecars, the brain dir, the title annotation and the presence lock.""" diff --git a/.agents/skills/lib_py/agents/adapters/claude.py b/.agents/skills/lib_py/agents/adapters/claude.py index 2510761..a9240a1 100644 --- a/.agents/skills/lib_py/agents/adapters/claude.py +++ b/.agents/skills/lib_py/agents/adapters/claude.py @@ -1,6 +1,6 @@ import os, re, json, glob, math, stat, time, shutil, calendar, subprocess from typing import Optional, Any -from lib_py.agents.base import BaseAgentAdapter, DiscoveryContext, conversation_in_use, scan_names +from lib_py.agents.base import BaseAgentAdapter, DiscoveryContext, PRE_ASSIGNED, conversation_in_use, scan_names from lib_py.verify_session import workspace_key @@ -184,6 +184,11 @@ class ClaudeAgentAdapter(BaseAgentAdapter): def own_key(self) -> str: return 'claude_session_id_own' + @property + def id_lifecycle(self) -> str: + # create hands the CLI its id (--session-id); nothing to capture (#40) + return PRE_ASSIGNED + @property def ready_tokens(self) -> str: return 'Anthropic|Assistant|Chat|Welcome|Claude Code|Opus|Sonnet|Haiku' diff --git a/.agents/skills/lib_py/agents/adapters/grok.py b/.agents/skills/lib_py/agents/adapters/grok.py index 72f0270..fa8a804 100644 --- a/.agents/skills/lib_py/agents/adapters/grok.py +++ b/.agents/skills/lib_py/agents/adapters/grok.py @@ -1,7 +1,7 @@ import os, json, glob, shutil from typing import Optional, Any from urllib.parse import quote -from lib_py.agents.base import BaseAgentAdapter, DiscoveryContext, scan_names +from lib_py.agents.base import BaseAgentAdapter, DiscoveryContext, PRE_ASSIGNED, scan_names from lib_py.verify_session import workspace_key @@ -14,6 +14,11 @@ class GrokAgentAdapter(BaseAgentAdapter): def own_key(self) -> str: return 'grok_session_id_own' + @property + def id_lifecycle(self) -> str: + # create hands the CLI its id (--session-id); nothing to capture (#40) + return PRE_ASSIGNED + @property def ready_tokens(self) -> str: return 'Grok|xAI|Assistant|❯|>>>' diff --git a/.agents/skills/lib_py/agents/autopin.py b/.agents/skills/lib_py/agents/autopin.py new file mode 100644 index 0000000..d366d26 --- /dev/null +++ b/.agents/skills/lib_py/agents/autopin.py @@ -0,0 +1,261 @@ +"""Auto-pin of a post-discovered session id on a verified local `completed` (Issue #40). + +Class A agents (agy, opencode, codex, hermes) write their conversation id to +disk only after their first turn, so create records the row without one. The +side that verified a job's `completed` - the delegate-job wrapper, create +--onboard, and the loop's wait_for_job as a recovery path - calls +handle_completed(). It captures the id through the adapter +(capture_active_session_id) and pins it with pin_row() under atomic_dump_yaml. +reconcile's drift C calls the same capture() and pin_row(), so the two paths +reach one decision. + +One (job_id, dispatch_token, attempt_id) is handled once: a job-scoped flock +and a marker in the job's directory make a second call (wrapper + loop +recovery) a no-op. Remote and human jobs have no local disk or pane and are +skipped; the agent's own publish_event.py never writes the control plane +(#15/#20/#21). A failed pin never changes the job's result: the reason is +written on the row and returned. + +CLI: python -m lib_py.agents.autopin --registry-dir DIR --job ID + [--yaml AGENT_SESSIONS_YAML] [--expect-dispatch-token TOKEN] +""" + +import argparse +import fcntl +import json +import os +import subprocess +import sys +import tempfile +import time +from typing import Any, Callable, Dict, List, Optional + +from lib_py.agents.base import DiscoveryContext, POST_DISCOVERED, capture_result +from lib_py.agents.registry import _ADAPTERS, agent_of_row, get_adapter + +OWN_KEYS = tuple(a.own_key for a in _ADAPTERS.values()) +MARKER = "autopin.json" + + +def claimed_elsewhere(row: Dict[str, Any], state: Dict[str, Any]) -> List[str]: + """Every id another row records: an id already some session's is never this one's.""" + ids: List[str] = [] + for other in state.get("herdr_sessions") or []: + if other is row or other.get("name") == row.get("name"): + continue + for key in OWN_KEYS: + v = other.get(key) + if v and v not in ids: + ids.append(v) + return ids + + +def capture(row: Dict[str, Any], agent: str, state: Dict[str, Any], home_dir: Optional[str] = None) -> Dict[str, Any]: + """The adapter's decision for this row (read-only). Besides the capture + statuses: 'pre-assigned' and 'already-set' (nothing to capture), 'skipped'.""" + adapter = get_adapter(agent) + if adapter is None: + return capture_result("skipped", reason=f"unknown agent {agent!r}") + if adapter.id_lifecycle != POST_DISCOVERED: + return capture_result("pre-assigned") + if row.get(adapter.own_key): + return capture_result("already-set", session_id=row[adapter.own_key]) + cwd = (row.get("pane") or {}).get("cwd") or "" + if not cwd: + return capture_result("no-candidate", reason="the row has no pane cwd") + row_eval = dict(row) + row_eval["_sibling_claimed_uuids"] = claimed_elsewhere(row, state) + ctx = DiscoveryContext(workspace=cwd, agent_name=agent, home_dir=home_dir, + epoch=row.get("herdr_session_epoch") or 0, row=row_eval, mode="discover") + return adapter.capture_active_session_id(ctx) + + +def pin_row(row: Dict[str, Any], agent: str, uuid: str) -> None: + """Record a verified id: own key, source and verified flag, in one change.""" + row[get_adapter(agent).own_key] = uuid + row["session_id_source"] = "discovered" + row["session_id_verified"] = True + row["last_visible_status"] = "pinned" + + +def _skills_dir() -> str: + return os.path.dirname(os.path.dirname(os.path.dirname(os.path.abspath(__file__)))) + + +def resume_dry_run(skills_dir: str, cwd: str, agent: str, name: str, env: Optional[Dict[str, str]] = None) -> str: + """last_visible_status after a pin: 'resume_verified' or why the dry-run failed.""" + cmd = ["bash", os.path.join(skills_dir, "multi-agent-mux-resume", "scripts", "resume_session.sh"), + "--workspace", cwd, "--agent", agent, "--session", name, "--dry-run"] + res = subprocess.run(cmd, capture_output=True, text=True, env=env) + if res.returncode == 0: + return "resume_verified" + return f"resume dry-run failed: {res.stderr.strip() or res.stdout.strip()}" + + +def tui_viewport(lib_sh: str, name: str, agent: str, cwd: str, env: Optional[Dict[str, str]] = None) -> int: + """lib.sh verify_tui_viewport: 0 match, 1 mismatch, other = no evidence.""" + script = 'source "$1" && verify_tui_viewport "$2" "$3" "$4"' + return subprocess.run(["bash", "-c", script, "_", lib_sh, name, agent, cwd], env=env).returncode + + +def _load_record(registry_dir: str, job_id: str) -> Optional[Dict[str, Any]]: + try: + with open(os.path.join(registry_dir, f"{job_id}.json"), encoding="utf-8") as f: + rec = json.load(f) + return rec if isinstance(rec, dict) else None + except (OSError, ValueError): + return None + + +_MUTATION: Optional[Callable[[Dict[str, Any]], bool]] = None + + +def _atomic_update(yaml_path: str, mutate: Callable[[Dict[str, Any]], bool]) -> None: + """Run `mutate(d)` as an atomic_dump_yaml mutation in this process; it + returns False to leave the registry untouched.""" + from lib_py import atomic_yaml + global _MUTATION + saved = {k: os.environ.get(k) for k in ("YAML_PATH", "AGENT_SESSIONS_MUTATION")} + _MUTATION = mutate + os.environ["YAML_PATH"] = yaml_path + # this very module (it is __main__ under `python -m`), not a second copy + os.environ["AGENT_SESSIONS_MUTATION"] = ( + "import sys as _sys\n" + f"if not _sys.modules[{__name__!r}]._MUTATION(d):\n" + " raise SystemExit(0)\n") + try: + atomic_yaml.atomic_dump_yaml_main() + except SystemExit as exc: + if exc.code not in (0, None): + raise RuntimeError(str(exc.code)) + finally: + _MUTATION = None + for k, v in saved.items(): + if v is None: + os.environ.pop(k, None) + else: + os.environ[k] = v + + + +def _subprocess_env(yaml_path: str, registry_dir: str) -> Dict[str, str]: + """The control plane the lib.sh checks run against: the registry's own.""" + env = dict(os.environ) + env.setdefault("AGENT_SESSIONS_YAML", yaml_path) + env.setdefault("MAM_JOBS_DIR", registry_dir) + mam = os.path.dirname(yaml_path) + if os.path.basename(mam) == ".mam": + env.setdefault("WORKSPACE_ROOT", os.path.dirname(mam)) + return env + + +def _pin_session(name: str, job_id: str, yaml_path: str, home_dir: Optional[str], + viewport: Callable[[str, str, str], int], resume: Callable[[str, str, str], str]) -> Dict[str, Any]: + out: Dict[str, Any] = {} + + def mutate(d: Dict[str, Any]) -> bool: + row = next((s for s in d.get("herdr_sessions") or [] if s.get("name") == name), None) + if row is None: + out.update(capture_result("skipped", reason=f"no session row {name!r}")) + return False + if row.get("status") != "running": + out.update(capture_result("skipped", reason=f"session {name!r} is {row.get('status')}")) + return False + agent = agent_of_row(row) + res = capture(row, agent or "", d, home_dir=home_dir) + cwd = (row.get("pane") or {}).get("cwd") or "" + if res["status"] == "captured": + rc = viewport(name, agent, cwd) + if rc == 1: + res = capture_result("viewport-mismatch", reason="the pane does not show this workspace", + candidates=res["candidates"]) + else: + pin_row(row, agent, res["session_id"]) + row["last_visible_status"] = resume(name, agent, cwd) + res = dict(res, status="pinned", degraded=(rc != 0)) + out.update(res) + return True + out.update(res) + if res["status"] in ("pre-assigned", "already-set", "skipped"): + return False + row["last_visible_status"] = f"autopin {res['status']} (job {job_id}): {res['reason']}" + return True + + _atomic_update(yaml_path, mutate) + return out + + +def handle_completed(registry_dir: str, job_id: str, yaml_path: Optional[str] = None, + expect_token: Optional[str] = None, home_dir: Optional[str] = None, + viewport: Optional[Callable[[str, str, str], int]] = None, + resume: Optional[Callable[[str, str, str], str]] = None) -> Dict[str, Any]: + """Pin the job's session id if the job completed locally and the row still + lacks a post-discovered id. Returns the result (also kept in the marker).""" + rec = _load_record(registry_dir, job_id) + if rec is None: + return capture_result("skipped", reason=f"no job record {job_id}") + if rec.get("status") != "completed": + return capture_result("skipped", reason=f"job {job_id} is {rec.get('status')}") + token = rec.get("dispatch_token") or "" + if expect_token and token != expect_token: + return capture_result("skipped", reason="stale dispatch token") + session = str(rec.get("agent_session") or "") + if (rec.get("transport") == "remote" or session.startswith("remote:") + or rec.get("agent") == "human" or session in ("", "human", "herdr:human")): + return capture_result("skipped", reason="not a local agent session") + name = session[len("herdr:"):] if session.startswith("herdr:") else session + yaml_path = yaml_path or os.environ.get("AGENT_SESSIONS_YAML") or os.path.join( + os.path.dirname(os.path.abspath(registry_dir)), "agent-sessions.yaml") + if not (os.path.exists(yaml_path) or os.path.exists(os.path.splitext(yaml_path)[0] + ".db")): + return capture_result("skipped", reason=f"no session registry at {yaml_path}") + + key = f"{job_id}:{token}:{rec.get('attempt_id')}" + job_dir = os.path.join(registry_dir, job_id) + os.makedirs(job_dir, exist_ok=True) + env = _subprocess_env(yaml_path, registry_dir) + if viewport is None: + lib_sh = os.environ.get("LIB_SH") or os.path.join(_skills_dir(), "lib.sh") + viewport = lambda n, a, c: tui_viewport(lib_sh, n, a, c, env=env) # noqa: E731 + if resume is None: + resume = lambda n, a, c: resume_dry_run(_skills_dir(), c, a, n, env=env) # noqa: E731 + + with open(os.path.join(job_dir, "autopin.lock"), "w") as lock: + fcntl.flock(lock, fcntl.LOCK_EX) + marker = os.path.join(job_dir, MARKER) + try: + with open(marker, encoding="utf-8") as f: + done = json.load(f) + if done.get("key") == key: + return dict(done.get("result") or {}, deduplicated=True) + except (OSError, ValueError): + pass + try: + result = _pin_session(name, job_id, yaml_path, home_dir, viewport, resume) + except Exception as exc: + result = capture_result("error", reason=f"{type(exc).__name__}: {exc}") + result["session"] = name + fd, tmp = tempfile.mkstemp(dir=job_dir, prefix=".autopin.", suffix=".tmp") + with os.fdopen(fd, "w", encoding="utf-8") as f: + json.dump({"key": key, "at": int(time.time()), "result": result}, f) + os.replace(tmp, marker) + return result + + +def main(argv: Optional[List[str]] = None) -> int: + p = argparse.ArgumentParser(prog="python -m lib_py.agents.autopin") + p.add_argument("--registry-dir", required=True) + p.add_argument("--job", required=True) + p.add_argument("--yaml", default=None, help="agent-sessions.yaml (default: beside the registry)") + p.add_argument("--expect-dispatch-token", default=None) + a = p.parse_args(argv) + res = handle_completed(a.registry_dir, a.job, yaml_path=a.yaml, expect_token=a.expect_dispatch_token) + if res.get("status") == "pinned": + print(f"autopin: {res['session']} pinned {res['session_id']} (job {a.job})", file=sys.stderr) + elif res.get("status") not in ("skipped", "pre-assigned", "already-set") and not res.get("deduplicated"): + print(f"autopin: {res.get('session')} not pinned - {res['status']}: {res.get('reason')} (job {a.job})", + file=sys.stderr) + return 0 + + +if __name__ == "__main__": + sys.exit(main()) diff --git a/.agents/skills/lib_py/agents/base.py b/.agents/skills/lib_py/agents/base.py index 51cc6d9..35423e6 100644 --- a/.agents/skills/lib_py/agents/base.py +++ b/.agents/skills/lib_py/agents/base.py @@ -169,6 +169,22 @@ def db_rows_naming(db: str, uuid: str, columns=(), id_tables=(), connect=None) - conn.close() return found +# Session-id lifecycles (Issue #40): the id is handed to the CLI before it +# starts, or the CLI writes it to disk after its first turn. +PRE_ASSIGNED = 'pre_assigned' +POST_DISCOVERED = 'post_discovered' + + +def capture_result(status: str, session_id: Optional[str] = None, reason: str = "", + candidates: Optional[List[str]] = None) -> Dict[str, Any]: + """What capture_active_session_id found (Issue #40). `status` is + 'captured' (session_id set) or why nothing may be pinned: 'no-candidate', + 'ambiguous', 'sibling-claimed', 'rejected', 'unverifiable-owner', + 'owner-mismatch'.""" + return {"status": status, "session_id": session_id, "reason": reason, + "candidates": list(candidates or [])} + + class SpawnSpec: def __init__(self, agent_name: str, session_name: str, command: str, env: Optional[Dict[str, str]] = None): self.agent_name = agent_name @@ -358,3 +374,28 @@ class BaseAgentAdapter: def discover(self, ctx: DiscoveryContext) -> list: raise NotImplementedError + + @property + def id_lifecycle(self) -> str: + """PRE_ASSIGNED: the id is given to the CLI at spawn (nothing to capture). + POST_DISCOVERED: the CLI writes it after its first turn (Issue #40).""" + return POST_DISCOVERED + + def capture_active_session_id(self, ctx: DiscoveryContext) -> Dict[str, Any]: + """This session's id, decided only when it is unambiguous (Issue #40): + exactly one discover() candidate that verify_session_uuid accepts and + no other row records (ctx.row['_sibling_claimed_uuids']). Read-only.""" + claimed = set(ctx.row.get("_sibling_claimed_uuids") or []) + found: List[str] = [] + for uuid in self.discover(ctx): + if not uuid or uuid in claimed or uuid in found: + continue + if verify_session_uuid(ctx.cwd, self.name, uuid, row=ctx.row, home_dir=ctx.home_dir, + claude_dir=ctx.claude_dir, mode="discover"): + found.append(uuid) + if not found: + return capture_result("no-candidate", reason="no verified candidate newer than the session") + if len(found) > 1: + return capture_result("ambiguous", reason=f"{len(found)} candidates newer than the session", + candidates=found) + return capture_result("captured", session_id=found[0], candidates=found) diff --git a/.agents/skills/lib_py/loop/runner.py b/.agents/skills/lib_py/loop/runner.py index 1ad23df..c30eef2 100644 --- a/.agents/skills/lib_py/loop/runner.py +++ b/.agents/skills/lib_py/loop/runner.py @@ -676,6 +676,7 @@ class LoopRunner: if status == "completed": if self.verbose: log_success(f"Job '{job_id}' completed successfully.") + self._autopin(job_id) return True elif status in ("error", "cancelled"): log_error(f"Job '{job_id}' finished with status '{status}'.") @@ -686,6 +687,23 @@ class LoopRunner: log_error(f"Job '{job_id}' timed out after {max_wait}s.") return False + def _autopin(self, job_id: str) -> None: + """Issue #40 recovery path: the handler the wrapper already ran for this + job (one job/token/attempt is handled once), run like the wrapper from + the control root's skills - lib_py.loop imports no sibling but + delegation (G9). Never fails the wait; the handler reports on stderr.""" + skills = self.workspace_root / ".agents" / "skills" + if not (skills / "lib_py" / "agents" / "autopin.py").exists(): + skills = Path(__file__).resolve().parents[2] + yaml_path = os.environ.get("AGENT_SESSIONS_YAML") or str(self.mam_dir / "agent-sessions.yaml") + env = dict(os.environ) + env["PYTHONPATH"] = str(skills) + (os.pathsep + env["PYTHONPATH"] if env.get("PYTHONPATH") else "") + try: + subprocess.run([sys.executable, "-m", "lib_py.agents.autopin", "--registry-dir", str(self.jobs_dir), + "--job", job_id, "--yaml", yaml_path], env=env, timeout=300) + except Exception as exc: + log_warn(f"autopin for job '{job_id}' failed: {exc}") + def _relay_while_submitting(self, label: str, since: str, stop: threading.Event) -> None: """Issue #30 (review R2): the wrapper runs until its job ends and its output is captured, so a blocked agent would only reach the loop log diff --git a/.agents/skills/multi-agent-mux-create/SKILL.md b/.agents/skills/multi-agent-mux-create/SKILL.md index 4d7912c..3d2a0b3 100644 --- a/.agents/skills/multi-agent-mux-create/SKILL.md +++ b/.agents/skills/multi-agent-mux-create/SKILL.md @@ -220,6 +220,7 @@ bash .agents/skills/multi-agent-mux-create/scripts/worktree_prepare.sh \ - **Don't use `nohup`/`disown`/`setsid` for the agent itself** — those background the agent outside herdr. The whole point of this skill is *the herdr session is the supervisor*. `nohup` is OK only for *launching the wrapper* (which itself creates the herdr session via `herdr new-session -d`). - **Claude Session ID Auto-Assignment**: Fresh `claude` sessions automatically generate a new UUID (`mam_gen_uuid`) passed via `claude --session-id `. `claude_session_id_own` is stored in `.mam/agent-sessions.yaml` with `session_id_source: assigned` and `session_id_verified: false`. - **First Message Materialization**: The transcript file `.jsonl` is created on disk when the first user prompt is sent. The monitor loop (`reconcile.sh`) verifies the transcript and promotes `session_id_verified: true` and `last_visible_status: pinned`. +- **Post-discovered ids (agy, hermes, opencode, codex)**: these rows start with no id (`session_id_source: pending-discovery`). With `--onboard`, create pins the id itself once the onboarding job completes (Issue #40, `lib_py.agents.autopin`), monitor or not; later, every completed delegate job does the same until the id is set. - **Wrapper script MUST NOT be created via `hermes profile alias`** — that command writes a `hermes -p ` wrapper that destroys the herdr behavior. Create wrappers manually (see `lab-landing-page-creator-claude` template). - **Always use the workspace-relative path** in herdr `cwd` — relative paths break when herdr respawns in a different shell context. diff --git a/.agents/skills/multi-agent-mux-create/scripts/create_session.sh b/.agents/skills/multi-agent-mux-create/scripts/create_session.sh index 7e9c131..210e2ad 100755 --- a/.agents/skills/multi-agent-mux-create/scripts/create_session.sh +++ b/.agents/skills/multi-agent-mux-create/scripts/create_session.sh @@ -751,6 +751,9 @@ If you delegate any part of it with multi-agent-mux-delegate-job submit, add --p case "$ONBOARD_STATUS" in completed) echo "onboarding completed: job $DELEGATE_JOB_ID ($(( $(date +%s) - ONBOARD_T0 ))s)" + # Issue #40: the first turn is done, so a post-discovered id is on disk - + # pin it now instead of waiting for a monitor (worktree creators run none). + delegate_autopin_completed "$DELEGATE_JOB_ID" "$DELEGATE_JOB_TOKEN" "$task_reg_dir" ;; *) # The session is kept, as for a missing handshake above: the row says diff --git a/.agents/skills/multi-agent-mux-delegate-job/multi-agent-mux-delegate-job b/.agents/skills/multi-agent-mux-delegate-job/multi-agent-mux-delegate-job index d658619..1127876 100755 --- a/.agents/skills/multi-agent-mux-delegate-job/multi-agent-mux-delegate-job +++ b/.agents/skills/multi-agent-mux-delegate-job/multi-agent-mux-delegate-job @@ -451,6 +451,11 @@ EOF finalize_unfinished "$JOB_ID" "$job_status" "$sub_rc" "direct" job_status="${FINAL_JOB_STATUS:-$job_status}" + # Issue #40: this subscriber verified the `completed` (or the disk did) - + # pin the session's post-discovered id now, daemon or not. + if [ "$job_status" = "completed" ]; then + delegate_autopin_completed "$JOB_ID" "$_disp_tok" "$REGISTRY_DIR" + fi # Last stdout line: the persistent audit-log dir for this job (see SKILL.md # "Audit Logs"). Callers can scrape `tail -n1` to find it. diff --git a/.agents/skills/multi-agent-mux-monitor/SKILL.md b/.agents/skills/multi-agent-mux-monitor/SKILL.md index af628e2..bd2d21b 100644 --- a/.agents/skills/multi-agent-mux-monitor/SKILL.md +++ b/.agents/skills/multi-agent-mux-monitor/SKILL.md @@ -161,6 +161,8 @@ YAML: no such session → report: "lab-landing-page-creator-claude: session id materialized b3a7...c2f" ``` - **C-ambiguous. Multiple candidates detected**: If multiple candidate transcripts match an unassigned session, the monitor avoids random pinning, reports `C-ambiguous`, and sets `last_visible_status: "ambiguous: N candidates"`. +- **Post-discovered ids (agy, hermes, opencode, codex; Issue #40)**: drift C asks the adapter's `capture_active_session_id`, the decision the completed handler (`lib_py.agents.autopin`) also uses. A candidate must be newer than the session, pass verification and be recorded by no other row. For agy, `last_conversations.json[cwd]` only proposes the id: the conversation's presence lock (or DB) must be held by the row's pane process or a descendant, and an id the cache maps to another workspace is rejected. The conversations folder is no longer scanned. A pin records `session_id_source: discovered` and `session_id_verified: true`. +- **Pinned without a monitor**: the side that verified a job's `completed` (the delegate-job wrapper, `create --onboard`, the loop's job wait) pins the session's post-discovered id at once, so worktree creators (`MAM_NO_SUBSCRIBER_DAEMON=1`) need no reconcile. One job/token/attempt is handled once; a failed or ambiguous capture leaves `last_visible_status: "autopin (job ): "` and the job's result unchanged. ## Pitfalls diff --git a/.agents/skills/multi-agent-mux-monitor/scripts/reconcile.sh b/.agents/skills/multi-agent-mux-monitor/scripts/reconcile.sh index 7711f66..d052e95 100755 --- a/.agents/skills/multi-agent-mux-monitor/scripts/reconcile.sh +++ b/.agents/skills/multi-agent-mux-monitor/scripts/reconcile.sh @@ -420,7 +420,7 @@ import yaml from lib_py.verify_session import verify_session_uuid, workspace_key from lib_py.agents.registry import get_adapter -from lib_py.agents.base import DiscoveryContext, _ro_connect +from lib_py.agents import autopin def _slug(path): if not path: @@ -554,17 +554,9 @@ def pane_meta(session, srv): from lib_py.agents.registry import get_adapter, own_key as _get_own_key, agent_of_row def _pin_and_verify_resume(s, agent, cwd, uuid, degraded=False): - own_key_field = _get_own_key(agent) - if own_key_field: - s[own_key_field] = uuid - s['last_visible_status'] = 'pinned' - resume_cmd = ['bash', os.path.join(skills_dir, 'multi-agent-mux-resume', 'scripts', 'resume_session.sh'), - '--workspace', cwd, '--agent', agent, '--session', s['name'], '--dry-run'] - res = subprocess.run(resume_cmd, capture_output=True, text=True) - if res.returncode == 0: - s['last_visible_status'] = 'resume_verified' - else: - s['last_visible_status'] = f"resume dry-run failed: {res.stderr.strip() or res.stdout.strip()}" + # the completed handler's pin (Issue #40): own id, source and verified together + autopin.pin_row(s, agent, uuid) + s['last_visible_status'] = autopin.resume_dry_run(skills_dir, cwd, agent, s['name']) id_name = 'session' if agent == 'claude' else 'conversation' if not degraded: @@ -783,314 +775,40 @@ for s in d.get('herdr_sessions', []): else: _pin_and_verify_resume(s, 'claude', cwd, uuid, degraded=True) -# === drift C (agy): agy 새 session id materialize (per-row own id) === +# === drift C (agy), drift C (hermes), drift C (opencode), drift C (codex) === +# Post-discovered ids (Issue #40): the adapter's capture_active_session_id +# decides - the same decision the completed handler (lib_py.agents.autopin) +# makes - so the two paths cannot drift apart. Exactly one verified candidate +# newer than the session that no other row records; agy additionally needs its +# presence lock held by this row's pane (its cwd cache alone proves nothing). for s in d.get('herdr_sessions', []): - if row_agent(s) != 'agy': + agent = row_agent(s) + if agent not in ('agy', 'hermes', 'opencode', 'codex'): continue if s.get('status') != 'running': continue - if s.get('agy_conversation_id_own'): + if s.get(OWN_KEY_BY_AGENT[agent]): continue cwd = (s.get('pane') or {}).get('cwd', '') if not cwd: continue - conv_dir = f"{home}/.gemini/antigravity-cli/conversations" - if not os.path.isdir(conv_dir): - continue - dbs = sorted(glob.glob(f"{conv_dir}/*.db"), key=os.path.getmtime, reverse=True) - - sibling_claimed = [ - other.get('agy_conversation_id_own') - for other in d.get('herdr_sessions', []) - if other is not s - and (other.get('pane') or {}).get('cwd') == cwd - and other.get('status') not in ('stopped', 'terminated') - and other.get('agy_conversation_id_own') - ] - s_eval = dict(s) - s_eval['_sibling_claimed_uuids'] = sibling_claimed - - valid_candidates = [] - for db in dbs: - uuid = os.path.basename(db)[:-3] - if verify_session_uuid(cwd, 'agy', uuid, s_eval, mode="discover"): - valid_candidates.append(uuid) - - if len(valid_candidates) > 1: + res = autopin.capture(s, agent, d, home_dir=home) + if res['status'] == 'ambiguous': + n = len(res['candidates']) drifts.append({'class': 'C-ambiguous', 'name': s['name'], - 'msg': f"{s['name']}: {len(valid_candidates)} candidate transcripts newer than session epoch; not pinning"}) - s['last_visible_status'] = f"ambiguous: {len(valid_candidates)} candidates" + 'msg': f"{s['name']}: {n} candidate transcripts newer than session epoch; not pinning"}) + s['last_visible_status'] = f"ambiguous: {n} candidates" actions.append(f"ambiguous candidates: {s['name']}") - if len(valid_candidates) == 1: - uuid = valid_candidates[0] - cmd = ['bash', '-c', f'source "{lib_sh}" && verify_tui_viewport "{s["name"]}" "agy" "{cwd}"'] + elif res['status'] == 'captured': + uuid = res['session_id'] + cmd = ['bash', '-c', f'source "{lib_sh}" && verify_tui_viewport "{s["name"]}" "{agent}" "{cwd}"'] rc = subprocess.run(cmd).returncode if rc == 0: - _pin_and_verify_resume(s, 'agy', cwd, uuid, degraded=False) + _pin_and_verify_resume(s, agent, cwd, uuid, degraded=False) elif rc == 1: drifts.append({'class': 'C-warn', 'name': s['name'], 'msg': f"{s['name']}: TUI viewport mismatch for candidate {uuid} — not pinned, will retry next cycle"}) else: - _pin_and_verify_resume(s, 'agy', cwd, uuid, degraded=True) - -# === drift C (hermes): hermes 새 session id materialize (per-row own id) === -for s in d.get('herdr_sessions', []): - if row_agent(s) != 'hermes': - continue - if s.get('status') != 'running': - continue - if s.get('hermes_conversation_id_own'): - continue - cwd = (s.get('pane') or {}).get('cwd', '') - if not cwd: - continue - hdb = f"{home}/.hermes/state.db" - if not os.path.exists(hdb): - continue - - epoch_threshold = s.get('herdr_session_epoch', 0) - - sibling_claimed = [ - other.get('hermes_conversation_id_own') - for other in d.get('herdr_sessions', []) - if other is not s - and (other.get('pane') or {}).get('cwd') == cwd - and other.get('status') not in ('stopped', 'terminated') - and other.get('hermes_conversation_id_own') - ] - s_eval = dict(s) - s_eval['_sibling_claimed_uuids'] = sibling_claimed - - valid_candidates = [] - try: - conn = _ro_connect(hdb) - rows = conn.execute( - "SELECT id FROM sessions WHERE cwd=? AND started_at >= ? ORDER BY started_at DESC LIMIT 20", - (cwd, epoch_threshold) - ).fetchall() - conn.close() - for (uuid,) in rows: - if uuid in (sibling_claimed or []): - continue - if verify_session_uuid(cwd, 'hermes', uuid, s_eval, mode="discover"): - valid_candidates.append(uuid) - except Exception: - pass - - if len(valid_candidates) > 1: - drifts.append({'class': 'C-ambiguous', 'name': s['name'], - 'msg': f"{s['name']}: {len(valid_candidates)} candidate transcripts newer than session epoch; not pinning"}) - s['last_visible_status'] = f"ambiguous: {len(valid_candidates)} candidates" - actions.append(f"ambiguous candidates: {s['name']}") - if len(valid_candidates) == 1: - uuid = valid_candidates[0] - cmd = ['bash', '-c', f'source "{lib_sh}" && verify_tui_viewport "{s["name"]}" "hermes" "{cwd}"'] - rc = subprocess.run(cmd).returncode - if rc == 0: - _pin_and_verify_resume(s, 'hermes', cwd, uuid, degraded=False) - elif rc == 1: - drifts.append({'class': 'C-warn', 'name': s['name'], 'msg': f"{s['name']}: TUI viewport mismatch for candidate {uuid} — not pinned, will retry next cycle"}) - else: - _pin_and_verify_resume(s, 'hermes', cwd, uuid, degraded=True) - -# === drift C (opencode): opencode 새 session id materialize (per-row own id) === -for s in d.get('herdr_sessions', []): - if row_agent(s) != 'opencode': - continue - if s.get('status') != 'running': - continue - if s.get('opencode_session_id_own'): - continue - cwd = (s.get('pane') or {}).get('cwd', '') - if not cwd: - continue - - # Resolve DB path via shared OpenCodeAgentAdapter helper with fallbacks - odb = None - try: - op_adapter = get_adapter('opencode') - if op_adapter: - odb = op_adapter._resolve_db(DiscoveryContext(workspace=cwd, agent_name='opencode', home_dir=home)) - except Exception: - pass - - if not odb: - opencode_base = f"{home}/.local/share/opencode" - candidates = [ - f"{opencode_base}/opencode.db", - f"{opencode_base}/storage.db", - ] - for cand in candidates: - if os.path.exists(cand): - odb = cand - break - if not odb and os.path.isdir(f"{opencode_base}/project"): - for root, _, files in os.walk(f"{opencode_base}/project"): - for f in files: - if f.endswith(".db"): - odb = os.path.join(root, f) - break - if odb and os.path.exists(odb): - break - - if not odb or not os.path.exists(odb): - continue - - epoch_threshold = s.get('herdr_session_epoch', 0) - - sibling_claimed = [ - other.get('opencode_session_id_own') - for other in d.get('herdr_sessions', []) - if other is not s - and (other.get('pane') or {}).get('cwd') == cwd - and other.get('status') not in ('stopped', 'terminated') - and other.get('opencode_session_id_own') - ] - s_eval = dict(s) - s_eval['_sibling_claimed_uuids'] = sibling_claimed - - valid_candidates = [] - try: - # read-only: nothing is created or removed beside the agent's db (Issue #60) - conn = _ro_connect(odb) - cursor = conn.cursor() - # opencode 2.x keeps sessions in session_v2 (a migrated 1.x `session` - # table is no longer written). - cursor.execute("PRAGMA table_info(session_v2)") - cols = [col[1] for col in cursor.fetchall()] - if cols: - cwd_col = "directory" if "directory" in cols else ("cwd" if "cwd" in cols else None) - time_col = "time_created" if "time_created" in cols else ("created_at" if "created_at" in cols else ("started_at" if "started_at" in cols else None)) - - query = "SELECT id FROM session_v2" - params = [] - where_clauses = [] - if "parent_id" in cols: - # a subagent's session shares the directory and is newer - where_clauses.append("parent_id IS NULL") - if cwd_col and cwd: - where_clauses.append(f"{cwd_col} = ?") - params.append(cwd) - if time_col and epoch_threshold: - if time_col == "time_created": - where_clauses.append(f"{time_col} >= ?") - params.append(int(float(epoch_threshold) * 1000)) - else: - where_clauses.append(f"{time_col} >= ?") - params.append(epoch_threshold) - if where_clauses: - query += " WHERE " + " AND ".join(where_clauses) - if time_col: - query += f" ORDER BY {time_col} DESC" - query += " LIMIT 20" - rows = cursor.execute(query, params).fetchall() - conn.close() - for (uuid,) in rows: - if uuid in (sibling_claimed or []): - continue - if verify_session_uuid(cwd, 'opencode', uuid, s_eval, mode="discover"): - valid_candidates.append(uuid) - else: - conn.close() - except Exception: - pass - - if len(valid_candidates) > 1: - drifts.append({'class': 'C-ambiguous', 'name': s['name'], - 'msg': f"{s['name']}: {len(valid_candidates)} candidate transcripts newer than session epoch; not pinning"}) - s['last_visible_status'] = f"ambiguous: {len(valid_candidates)} candidates" - actions.append(f"ambiguous candidates: {s['name']}") - if len(valid_candidates) == 1: - uuid = valid_candidates[0] - cmd = ['bash', '-c', f'source "{lib_sh}" && verify_tui_viewport "{s["name"]}" "opencode" "{cwd}"'] - rc = subprocess.run(cmd).returncode - if rc == 0: - _pin_and_verify_resume(s, 'opencode', cwd, uuid, degraded=False) - elif rc == 1: - drifts.append({'class': 'C-warn', 'name': s['name'], 'msg': f"{s['name']}: TUI viewport mismatch for candidate {uuid} — not pinned, will retry next cycle"}) - else: - _pin_and_verify_resume(s, 'opencode', cwd, uuid, degraded=True) - -# === drift C (codex): codex 새 session id materialize (per-row own id) === -for s in d.get('herdr_sessions', []): - if row_agent(s) != 'codex': - continue - if s.get('status') != 'running': - continue - if s.get('codex_session_id_own'): - continue - cwd = (s.get('pane') or {}).get('cwd', '') - if not cwd: - continue - - codex_home = os.environ.get('CODEX_HOME') or f"{home}/.codex" - epoch_threshold = s.get('herdr_session_epoch', 0) - - sibling_claimed = [ - other.get('codex_session_id_own') - for other in d.get('herdr_sessions', []) - if other is not s - and (other.get('pane') or {}).get('cwd') == cwd - and other.get('status') not in ('stopped', 'terminated') - and other.get('codex_session_id_own') - ] - s_eval = dict(s) - s_eval['_sibling_claimed_uuids'] = sibling_claimed - - valid_candidates = [] - try: - # Rev.2: state DB first (indexed, single query) — dynamic filename, - # no hardcoded suffix (see §1.5 / §2 _state_db()). - state_matches = glob.glob(f"{codex_home}/state_*.sqlite") - state_db = max(state_matches, key=os.path.getmtime) if state_matches else None - rows = [] - if state_db and os.path.exists(state_db): - try: - conn = _ro_connect(state_db) - cols = [c[1] for c in conn.execute("PRAGMA table_info(threads)").fetchall()] - order_col = "updated_at" if "updated_at" in cols else ("created_at" if "created_at" in cols else None) - if order_col: - rows = conn.execute( - f"SELECT id, {order_col} FROM threads WHERE cwd=? ORDER BY {order_col} DESC LIMIT 20", - (cwd,) - ).fetchall() - else: - rows = [(r[0], int(time.time())) for r in conn.execute("SELECT id FROM threads WHERE cwd=? LIMIT 20", (cwd,)).fetchall()] - conn.close() - except Exception: - rows = [] - if not rows: - # Fallback: bounded glob over the date-nested tree - paths = sorted( - glob.glob(f"{codex_home}/sessions/*/*/*/rollout-*.jsonl"), - key=os.path.getmtime, reverse=True - )[:50] - rows = [(os.path.basename(p)[:-6][-36:], os.path.getmtime(p)) for p in paths] - - for uuid, ts in rows: - if epoch_threshold and ts and ts < epoch_threshold: - continue - if uuid in (sibling_claimed or []): - continue - if verify_session_uuid(cwd, 'codex', uuid, s_eval, mode="discover"): - valid_candidates.append(uuid) - except Exception: - pass - - if len(valid_candidates) > 1: - drifts.append({'class': 'C-ambiguous', 'name': s['name'], - 'msg': f"{s['name']}: {len(valid_candidates)} candidate transcripts newer than session epoch; not pinning"}) - s['last_visible_status'] = f"ambiguous: {len(valid_candidates)} candidates" - actions.append(f"ambiguous candidates: {s['name']}") - elif len(valid_candidates) == 1: - uuid = valid_candidates[0] - cmd = ['bash', '-c', f'source "{lib_sh}" && verify_tui_viewport "{s["name"]}" "codex" "{cwd}"'] - rc = subprocess.run(cmd).returncode - if rc == 0: - _pin_and_verify_resume(s, 'codex', cwd, uuid, degraded=False) - elif rc == 1: - drifts.append({'class': 'C-warn', 'name': s['name'], 'msg': f"{s['name']}: TUI viewport mismatch for candidate {uuid} — not pinned, will retry next cycle"}) - else: - _pin_and_verify_resume(s, 'codex', cwd, uuid, degraded=True) + _pin_and_verify_resume(s, agent, cwd, uuid, degraded=True) result = { 'timestamp': now_iso, diff --git a/docs/NEW_AGENT_INTEGRATION_GUIDE.md b/docs/NEW_AGENT_INTEGRATION_GUIDE.md index 964a0af..f652df6 100644 --- a/docs/NEW_AGENT_INTEGRATION_GUIDE.md +++ b/docs/NEW_AGENT_INTEGRATION_GUIDE.md @@ -34,6 +34,7 @@ MAM은 에이전트별 동작 특성(TUI 프롬프트 패턴, 세션 복원 인 │ │ • input_prompt, input_placeholder, input_rule_pattern │ │ │ │ • artifact_path(), verify_artifact(), purge_artifacts() │ │ │ │ • spawn_spec(), resume_spec(), auth_ok(), discover() │ │ +│ │ • id_lifecycle, capture_active_session_id() (Issue #40) │ │ │ └────────────────────────────────────────────────────────────────┬─────────────────────────────────────────────────────────────────┘ │ │ │ │ │ ┌────────────────────┬───────────────────────────────┬───────────────────────────────┬─────────────────────┐ │ @@ -70,7 +71,7 @@ MAM은 에이전트별 동작 특성(TUI 프롬프트 패턴, 세션 복원 인 import os from typing import Any, List, Optional -from ..base import BaseAgentAdapter, DiscoveryContext +from ..base import BaseAgentAdapter, DiscoveryContext, PRE_ASSIGNED class GrokAgentAdapter(BaseAgentAdapter): @@ -169,8 +170,19 @@ class GrokAgentAdapter(BaseAgentAdapter): def discover(self, ctx: DiscoveryContext) -> List[str]: """디스크의 세션 저장소에서 워크스페이스와 일치하는 세션 UUID 목록 탐색""" return [] + + # ========================================================================= + # 세션 ID 라이프사이클 (Issue #40) + # ========================================================================= + @property + def id_lifecycle(self) -> str: + """PRE_ASSIGNED: 기동 시 --session-id 로 ID 를 넘긴다 (claude, grok). + POST_DISCOVERED(기본값): 첫 턴 뒤 디스크에 생긴 ID 를 찾는다 (agy, opencode, codex, hermes).""" + return PRE_ASSIGNED ``` +**세션 ID 자동 확정 (Issue #40)**: `POST_DISCOVERED` 어댑터는 `capture_active_session_id(ctx)` 를 제공합니다. 기본 구현은 `discover()` 후보 중 `verify_session_uuid` 를 통과하고 다른 행이 기록하지 않은 것이 **정확히 하나**일 때만 `capture_result('captured', session_id=...)` 를 돌려주고, 아니면 `no-candidate` / `ambiguous` 등 이유를 돌려줍니다(읽기 전용). 후보 출처가 하나의 값만 주는 경우(agy 의 cwd 캐시처럼)에는 이 유일성이 아무것도 증명하지 못하므로, 그 어댑터는 이 메서드를 재정의해 소유를 따로 증명해야 합니다(agy: presence lock 을 쥔 프로세스가 그 행의 pane 프로세스 또는 자손). 실제 기록은 `completed` 를 검증한 수신 측(`lib_py.agents.autopin`: 위임 래퍼, `create --onboard`, 루프의 잡 대기)과 reconcile drift C 가 같은 판정으로 하며, 어댑터는 YAML/SQLite 에 쓰지 않습니다. + --- ### Step 2: 어댑터 레지스트리 등록 (`.agents/skills/lib_py/agents/registry.py`) diff --git a/tests/test_issue40_auto_pin.py b/tests/test_issue40_auto_pin.py new file mode 100644 index 0000000..528a66d --- /dev/null +++ b/tests/test_issue40_auto_pin.py @@ -0,0 +1,599 @@ +"""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 /.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 diff --git a/tests/test_issue60_linux_fixes.py b/tests/test_issue60_linux_fixes.py index d476a58..534585b 100644 --- a/tests/test_issue60_linux_fixes.py +++ b/tests/test_issue60_linux_fixes.py @@ -193,7 +193,7 @@ def test_a_query_reads_right_and_leaves_the_sidecars_alone(wal_home, state, list @pytest.mark.parametrize("script,count", [ ("multi-agent-mux-status/scripts/status.sh", 2), - ("multi-agent-mux-monitor/scripts/reconcile.sh", 1), + ("multi-agent-mux-monitor/scripts/reconcile.sh", 0), # Issue #40: through the adapter only ]) def test_status_and_reconcile_open_the_database_read_only(script, count): text = (SKILLS / script).read_text() diff --git a/tests/test_issue60_opencode_v2.py b/tests/test_issue60_opencode_v2.py index 7380fbc..5f207e1 100644 --- a/tests/test_issue60_opencode_v2.py +++ b/tests/test_issue60_opencode_v2.py @@ -281,7 +281,8 @@ def test_status_and_reconcile_read_session_v2(): for rel in ("multi-agent-mux-status/scripts/status.sh", "multi-agent-mux-monitor/scripts/reconcile.sh"): text = (SKILLS / rel).read_text() assert not re.search(r"FROM session\b(?!_)|table_info\(session\)", text), rel - assert "session_v2" in text, rel + # Issue #40: reconcile's drift C reads session_v2 through the adapter + assert ("session_v2" in text) if "status" in rel else ("autopin.capture(" in text), rel # --------------------------------------------------------------------------- diff --git a/tests/test_issue64_ro_connect.py b/tests/test_issue64_ro_connect.py index 63c9e9c..30c2a22 100644 --- a/tests/test_issue64_ro_connect.py +++ b/tests/test_issue64_ro_connect.py @@ -230,10 +230,11 @@ def test_status_and_reconcile_static_ro_connect_usage(): assert status_text.count("_ro_connect(db)") == 2 assert "sqlite3.connect" not in status_text + # Issue #40: drift C reads the agents' databases only through the adapters + # (lib_py.agents.autopin.capture), which use _ro_connect. reconcile_text = (SKILLS / "multi-agent-mux-monitor" / "scripts" / "reconcile.sh").read_text() - assert reconcile_text.count("_ro_connect(hdb)") == 1 - assert reconcile_text.count("_ro_connect(odb)") == 1 - assert reconcile_text.count("_ro_connect(state_db)") == 1 + assert "autopin.capture(s, agent, d, home_dir=home)" in reconcile_text + assert "_ro_connect" not in reconcile_text assert "sqlite3.connect" not in reconcile_text @@ -370,12 +371,11 @@ d['herdr_sessions'] = [ # 4. Cross-workspace agy DB sidecar regression test (Task 1 root-cause) # --------------------------------------------------------------------------- def test_unpinned_agy_reconcile_leaves_other_workspace_clean_db_without_sidecars(tmp_path): - """When workspace A has an unpinned running agy session, reconcile.sh's - drift-C scan iterates every `*.db` in `~/.gemini/antigravity-cli/conversations/` - and calls `verify_session_uuid(ws_a, 'agy', uuid, ..., mode='discover')`. - Even when `last_conversations.json` maps `uuid_b` to another workspace `ws_b`, - `verify_artifact` opens `uuid_b.db` to query `SELECT count(*) FROM steps`; - with `_ro_connect`, no `-wal` or `-shm` sidecar is created beside `uuid_b.db`.""" + """`verify_session_uuid(ws_a, 'agy', uuid_b, ..., mode='discover')` for an + unpinned running agy session in workspace A leaves no `-wal` or `-shm` + sidecar beside another workspace's `uuid_b.db` (reconcile.sh's drift C + used to call it for every `*.db` in the conversations folder). Since Issue + #40 an id `last_conversations.json` maps to another workspace is rejected.""" home = tmp_path / "home" ws_a = tmp_path / "ws_a" ws_b = tmp_path / "ws_b" @@ -404,5 +404,6 @@ def test_unpinned_agy_reconcile_leaves_other_workspace_clean_db_without_sidecars "pane": {"cwd": str(ws_a), "cmd": "agy"}, "_sibling_claimed_uuids": [], } - assert verify_session_uuid(str(ws_a), "agy", uuid_b, row=row_a, home_dir=str(home), mode="discover") is True + # Issue #40: an id the cache maps to another workspace is rejected + assert verify_session_uuid(str(ws_a), "agy", uuid_b, row=row_a, home_dir=str(home), mode="discover") is False assert sorted(os.listdir(db_b.parent)) == [f"{uuid_b}.db"] diff --git a/tests/test_tier1_unit.py b/tests/test_tier1_unit.py index 4b1e6a3..a18b75b 100644 --- a/tests/test_tier1_unit.py +++ b/tests/test_tier1_unit.py @@ -1054,7 +1054,8 @@ def test_opencode_shell_and_scripts_integration(mam_sandbox): recon_path = mam_sandbox / "skills" / "multi-agent-mux-monitor" / "scripts" / "reconcile.sh" recon_content = recon_path.read_text() assert 'import os, json, glob, subprocess, time, sqlite3, re, shutil' in recon_content - assert 'op_adapter = get_adapter(\'opencode\')' in recon_content + # Issue #40: drift C (opencode) goes through the adapter's capture + assert 'autopin.capture(s, agent, d, home_dir=home)' in recon_content def test_codex_shell_and_scripts_integration(mam_sandbox): diff --git a/tests/test_tier2_component.py b/tests/test_tier2_component.py index 0228ff6..213c4e6 100644 --- a/tests/test_tier2_component.py +++ b/tests/test_tier2_component.py @@ -1233,7 +1233,10 @@ def test_comp_opencode_create_and_reconcile_and_stop(mam_sandbox, mock_herdr, mo "bash", str(create_script), "--workspace", str(mam_sandbox), "--agent", "opencode", - "--role", "creator" + "--role", "creator", + # the completed onboarding would pin the id itself (Issue #40); this + # test is about reconcile's drift C + "--no-onboard" ], capture_output=True, text=True, cwd=str(mam_sandbox)) assert res.returncode == 0, res.stderr @@ -1296,7 +1299,10 @@ def test_comp_codex_create_and_reconcile_and_stop(mam_sandbox, mock_herdr, mock_ "bash", str(create_script), "--workspace", str(mam_sandbox), "--agent", "codex", - "--role", "creator" + "--role", "creator", + # the completed onboarding would pin the id itself (Issue #40); this + # test is about reconcile's drift C + "--no-onboard" ], capture_output=True, text=True, cwd=str(mam_sandbox)) assert res.returncode == 0, res.stderr