test(issue-13): give each purge test fresh conversation ids

Round-2 review of 7610258 (planner-reviewer-claude-01, reviewer-grok-01,
N2): the purge tests used two fixed UUIDs, while conversation_in_use scans
every process on the host. A review job's delegation wrapper carries the
whole brief - including this file's diff and so both constants - on its
command line, so five tests read "in use" for as long as that wrapper
lived. Reproduced here: this job's own wrapper named both constants and
the same five tests failed.

- tests: `conv` / `other` fixtures hand every test a fresh uuid4 and
  assert that no live process names it yet; the helpers already took the
  id as an argument. With a bystander process naming the old constants,
  the file passes 21/21 and the brief's required set 248/248
- orc-onboard SKILL.md: any live process whose command line contains the
  UUID blocks the purge, not only a resumed session (review R-2), and how
  to find it

No product code changes.

Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01HCGpJXhSZasJLU38ksDghe

tests/test_issue10_13_purge.py

@@ -11,6 +11,10 @@ purgeable through orc_onboard.sh.
      adapter from what is on disk (orchestrator_uuids stores bare UUIDs), purges
      through it, and refuses while a live process still names the UUID -
      including the orchestrator's own session running the command.
+
+Every test gets fresh conversation ids (the `conv` / `other` fixtures):
+conversation_in_use scans every process on the host, so a fixed id in this
+file is named by anything that carries the file's text on its command line.
 """
 from __future__ import annotations
 
@@ -20,6 +24,7 @@ import sqlite3
 import subprocess
 import sys
 import time
+import uuid as uuidlib
 from pathlib import Path
 
 import pytest
@@ -32,15 +37,37 @@ sys.path.insert(0, str(SKILLS))
 from lib_py.agents.registry import get_adapter  # noqa: E402
 from lib_py.agents.base import DiscoveryContext, conversation_in_use  # noqa: E402
 
-UUID = "7d0b6f12-3a4c-4e5f-8a9b-0c1d2e3f4a5b"
-OTHER = "11111111-2222-4333-8444-555555555555"
-
 # The session-id variables an agent CLI exports to its tool shells. Scrubbed
 # from every subprocess env below so the caller's own session never leaks in.
 SESSION_ENV = ("CLAUDE_CODE_SESSION_ID", "ANTIGRAVITY_CONVERSATION_ID",
                "HERMES_SESSION_ID", "GROK_SESSION_ID", "OPENCODE_SESSION_ID")
 
 
+def _processes_naming(ident: str) -> list:
+    out = subprocess.run(["ps", "-axo", "pid=,command="], capture_output=True, text=True, timeout=10).stdout
+    return [line.strip()[:120] for line in out.splitlines() if ident in line]
+
+
+def _fresh_id() -> str:
+    """A conversation id no other process can already be naming. With fixed
+    ids, the delegation wrapper of a review job - whose --prompt carried this
+    file's diff - named them, and five tests read "in use" during review."""
+    ident = str(uuidlib.uuid4())
+    named = _processes_naming(ident)
+    assert not named, f"fresh id {ident} is already named by a live process: {named}"
+    return ident
+
+
+@pytest.fixture
+def conv() -> str:
+    return _fresh_id()
+
+
+@pytest.fixture
+def other() -> str:
+    return _fresh_id()
+
+
 def _clean_env(extra=None) -> dict:
     env = dict(os.environ)
     for k in SESSION_ENV + ("MAM_ORCHESTRATOR_UUIDS",):
@@ -130,26 +157,26 @@ def _registered(sandbox: Path, uuid: str) -> bool:
 # ---------------------------------------------------------------------------
 # #10 — agy adapter
 # ---------------------------------------------------------------------------
-def test_agy_purge_removes_sidecars_annotation_and_lock(tmp_path):
-    paths = _agy_tree(tmp_path, UUID)
+def test_agy_purge_removes_sidecars_annotation_and_lock(tmp_path, conv):
+    paths = _agy_tree(tmp_path, conv)
     adapter = get_adapter("agy")
     ctx = DiscoveryContext(workspace=str(tmp_path), agent_name="agy", home_dir=str(tmp_path))
 
-    purged = adapter.purge_artifacts(UUID, ctx)
+    purged = adapter.purge_artifacts(conv, ctx)
 
     assert len(purged) == 6, purged
     for key, p in paths.items():
         assert not p.exists(), f"{key} survived the purge: {p}"
-    assert adapter.residual_artifacts(UUID, ctx) == []
+    assert adapter.residual_artifacts(conv, ctx) == []
 
 
-def test_agy_purge_keeps_presence_lock_while_in_use(tmp_path):
-    paths = _agy_tree(tmp_path, UUID)
+def test_agy_purge_keeps_presence_lock_while_in_use(tmp_path, conv):
+    paths = _agy_tree(tmp_path, conv)
     adapter = get_adapter("agy")
     ctx = DiscoveryContext(workspace=str(tmp_path), agent_name="agy", home_dir=str(tmp_path))
-    holder = _holder(UUID)
+    holder = _holder(conv)
     try:
-        purged = adapter.purge_artifacts(UUID, ctx)
+        purged = adapter.purge_artifacts(conv, ctx)
     finally:
         _stop(holder)
 
@@ -157,34 +184,34 @@ def test_agy_purge_keeps_presence_lock_while_in_use(tmp_path):
     assert str(paths["lock"]) not in purged
     for key in ("db", "wal", "shm", "brain", "annotation"):
         assert not paths[key].exists(), key
-    assert adapter.residual_artifacts(UUID, ctx) == [str(paths["lock"])]
+    assert adapter.residual_artifacts(conv, ctx) == [str(paths["lock"])]
 
 
-def test_agy_purge_reports_only_what_existed(tmp_path):
+def test_agy_purge_reports_only_what_existed(tmp_path, conv):
     """The historical 2-item contract (db + brain) still holds when no sidecar
     exists (test_a4_adapter_contract pins len == 2)."""
     root = tmp_path / ".gemini" / "antigravity-cli"
     (root / "conversations").mkdir(parents=True)
-    (root / "conversations" / f"{UUID}.db").write_text("db")
-    (root / "brain" / UUID).mkdir(parents=True)
+    (root / "conversations" / f"{conv}.db").write_text("db")
+    (root / "brain" / conv).mkdir(parents=True)
     adapter = get_adapter("agy")
     ctx = DiscoveryContext(workspace=str(tmp_path), agent_name="agy", home_dir=str(tmp_path))
-    assert len(adapter.purge_artifacts(UUID, ctx)) == 2
+    assert len(adapter.purge_artifacts(conv, ctx)) == 2
 
 
-def test_conversation_in_use_ignores_own_lineage():
+def test_conversation_in_use_ignores_own_lineage(other):
     """The purge command names the uuid on its own command line (and its
     parent shell does); only *other* processes count as holders."""
-    assert conversation_in_use(OTHER) is False
-    holder = _holder(OTHER)
+    assert conversation_in_use(other) is False
+    holder = _holder(other)
     try:
-        assert conversation_in_use(OTHER) is True
+        assert conversation_in_use(other) is True
     finally:
         _stop(holder)
-    assert conversation_in_use(OTHER) is False
+    assert conversation_in_use(other) is False
 
 
-def test_conversation_in_use_excludes_only_the_non_agent_helper_chain(tmp_path):
+def test_conversation_in_use_excludes_only_the_non_agent_helper_chain(tmp_path, other):
     """The purge's own helper chain names the uuid too - the python argv and
     the script that launched it - and must not count. The exclusion stops at
     the nearest agent CLI ancestor, though: that process is the session the
@@ -196,13 +223,13 @@ def test_conversation_in_use_excludes_only_the_non_agent_helper_chain(tmp_path):
     helper.write_text('#!/bin/sh\n"$@"\nexit $?\n')
     helper.chmod(0o755)
 
-    alone = subprocess.run([str(helper), sys.executable, "-c", probe, OTHER],
+    alone = subprocess.run([str(helper), sys.executable, "-c", probe, other],
                            capture_output=True, text=True, env=_clean_env(), timeout=60)
     assert alone.stdout.strip() == "False", alone.stdout + alone.stderr
 
     cli = _fake_agent_cli(tmp_path, "agy")
-    under_agent = subprocess.run([str(cli), "--conversation", OTHER, "--",
-                                  str(helper), sys.executable, "-c", probe, OTHER],
+    under_agent = subprocess.run([str(cli), "--conversation", other, "--",
+                                  str(helper), sys.executable, "-c", probe, other],
                                  capture_output=True, text=True, env=_clean_env(), timeout=60)
     assert under_agent.stdout.strip() == "True", under_agent.stdout + under_agent.stderr
 
@@ -217,9 +244,9 @@ def _purge_cli(agent: str, uuid: str, home: Path, extra_env=None, prefix=()) ->
     return subprocess.run(cmd, capture_output=True, text=True, env=env, cwd=str(home), timeout=60)
 
 
-def test_purge_cli_auto_resolves_agy_from_disk(tmp_path):
-    paths = _agy_tree(tmp_path, UUID)
-    res = _purge_cli("auto", UUID, tmp_path)
+def test_purge_cli_auto_resolves_agy_from_disk(tmp_path, conv):
+    paths = _agy_tree(tmp_path, conv)
+    res = _purge_cli("auto", conv, tmp_path)
     assert res.returncode == 0, res.stderr
     out = json.loads(res.stdout)
     assert out["agent"] == "agy"
@@ -227,11 +254,11 @@ def test_purge_cli_auto_resolves_agy_from_disk(tmp_path):
     assert not any(p.exists() for p in paths.values())
 
 
-def test_purge_cli_refuses_while_in_use(tmp_path):
-    paths = _agy_tree(tmp_path, UUID)
-    holder = _holder(UUID)
+def test_purge_cli_refuses_while_in_use(tmp_path, conv):
+    paths = _agy_tree(tmp_path, conv)
+    holder = _holder(conv)
     try:
-        res = _purge_cli("agy", UUID, tmp_path)
+        res = _purge_cli("agy", conv, tmp_path)
     finally:
         _stop(holder)
     assert res.returncode == 4, res.stdout + res.stderr
@@ -239,33 +266,33 @@ def test_purge_cli_refuses_while_in_use(tmp_path):
     assert all(p.exists() for p in paths.values()), "a refused purge must remove nothing"
 
 
-def test_purge_cli_refuses_inside_the_agent_session_that_owns_the_uuid(tmp_path):
+def test_purge_cli_refuses_inside_the_agent_session_that_owns_the_uuid(tmp_path, conv):
     """An orchestrator purging its own conversation from its own session: the
     purge runs as a descendant of the agent CLI whose command line names the
     uuid. That ancestor is the conversation's live process, not part of the
     purge's helper chain, so the purge must be refused and delete nothing."""
-    paths = _agy_tree(tmp_path, UUID)
+    paths = _agy_tree(tmp_path, conv)
     cli = _fake_agent_cli(tmp_path, "claude")
-    res = _purge_cli("agy", UUID, tmp_path, prefix=(str(cli), "--session-id", UUID, "--"))
+    res = _purge_cli("agy", conv, tmp_path, prefix=(str(cli), "--session-id", conv, "--"))
     assert res.returncode == 4, res.stdout + res.stderr
     assert json.loads(res.stdout)["in_use"] is True
     assert all(p.exists() for p in paths.values()), "a refused purge must remove nothing"
 
 
 @pytest.mark.parametrize("var", SESSION_ENV)
-def test_purge_cli_refuses_from_the_sessions_own_shell(tmp_path, var):
+def test_purge_cli_refuses_from_the_sessions_own_shell(tmp_path, var, conv):
     """The session a tool shell belongs to is in its environment even when no
     process carries the uuid in argv and nothing holds the artifact open
     (claude keeps no transcript open): a purge run from there is refused."""
-    paths = _agy_tree(tmp_path, UUID)
-    res = _purge_cli("agy", UUID, tmp_path, extra_env={var: UUID})
+    paths = _agy_tree(tmp_path, conv)
+    res = _purge_cli("agy", conv, tmp_path, extra_env={var: conv})
     assert res.returncode == 4, res.stdout + res.stderr
     assert json.loads(res.stdout)["in_use"] is True
     assert all(p.exists() for p in paths.values()), "a refused purge must remove nothing"
 
 
-def test_purge_cli_unresolved_uuid(tmp_path):
-    res = _purge_cli("auto", OTHER, tmp_path)
+def test_purge_cli_unresolved_uuid(tmp_path, other):
+    res = _purge_cli("auto", other, tmp_path)
     assert res.returncode == 5, res.stdout + res.stderr
     assert json.loads(res.stdout)["agent"] is None
 
@@ -273,82 +300,82 @@ def test_purge_cli_unresolved_uuid(tmp_path):
 # ---------------------------------------------------------------------------
 # #13 — orc_onboard.sh --remove --purge-conversation
 # ---------------------------------------------------------------------------
-def test_orc_remove_purge_requires_yes(mam_sandbox):
-    paths = _agy_tree(mam_sandbox, UUID)
-    assert _run_onboard(["--uuid", UUID], mam_sandbox).returncode == 0
+def test_orc_remove_purge_requires_yes(mam_sandbox, conv):
+    paths = _agy_tree(mam_sandbox, conv)
+    assert _run_onboard(["--uuid", conv], mam_sandbox).returncode == 0
 
-    res = _run_onboard(["--remove", UUID, "--purge-conversation"], mam_sandbox)
+    res = _run_onboard(["--remove", conv, "--purge-conversation"], mam_sandbox)
 
     assert res.returncode == 4, res.stdout + res.stderr
     assert "Re-run with --yes" in res.stderr
     assert all(p.exists() for p in paths.values())
-    assert _registered(mam_sandbox, UUID), "registration must be untouched without --yes"
+    assert _registered(mam_sandbox, conv), "registration must be untouched without --yes"
 
 
-def test_orc_remove_purge_deletes_and_unregisters(mam_sandbox):
-    paths = _agy_tree(mam_sandbox, UUID)
-    assert _run_onboard(["--uuid", UUID], mam_sandbox).returncode == 0
+def test_orc_remove_purge_deletes_and_unregisters(mam_sandbox, conv):
+    paths = _agy_tree(mam_sandbox, conv)
+    assert _run_onboard(["--uuid", conv], mam_sandbox).returncode == 0
 
-    res = _run_onboard(["--remove", UUID, "--purge-conversation", "--yes"], mam_sandbox)
+    res = _run_onboard(["--remove", conv, "--purge-conversation", "--yes"], mam_sandbox)
 
     assert res.returncode == 0, res.stdout + res.stderr
     assert "Successfully removed orchestrator UUID" in res.stdout
     assert res.stdout.count("purged: ") == 6, res.stdout
     assert "residual: none (agent: agy)" in res.stdout
     assert not any(p.exists() for p in paths.values())
-    assert not _registered(mam_sandbox, UUID)
+    assert not _registered(mam_sandbox, conv)
 
 
-def test_orc_remove_purge_refused_while_in_use(mam_sandbox):
-    paths = _agy_tree(mam_sandbox, UUID)
-    assert _run_onboard(["--uuid", UUID], mam_sandbox).returncode == 0
-    holder = _holder(UUID)
+def test_orc_remove_purge_refused_while_in_use(mam_sandbox, conv):
+    paths = _agy_tree(mam_sandbox, conv)
+    assert _run_onboard(["--uuid", conv], mam_sandbox).returncode == 0
+    holder = _holder(conv)
     try:
-        res = _run_onboard(["--remove", UUID, "--purge-conversation", "--yes"], mam_sandbox)
+        res = _run_onboard(["--remove", conv, "--purge-conversation", "--yes"], mam_sandbox)
     finally:
         _stop(holder)
 
     assert res.returncode == 5, res.stdout + res.stderr
     assert "still in use" in res.stderr
     assert all(p.exists() for p in paths.values())
-    assert _registered(mam_sandbox, UUID), "a refused purge must leave the registration in place"
+    assert _registered(mam_sandbox, conv), "a refused purge must leave the registration in place"
 
 
-def test_orc_remove_purge_refused_from_the_orchestrators_own_session(mam_sandbox):
+def test_orc_remove_purge_refused_from_the_orchestrators_own_session(mam_sandbox, conv):
     """The case the skill documents: the orchestrator removes and purges its
     own UUID from inside its own session (its tool shell carries the session
     id). The purge must be refused (exit 5) and the registration kept."""
-    paths = _agy_tree(mam_sandbox, UUID)
-    assert _run_onboard(["--uuid", UUID], mam_sandbox).returncode == 0
+    paths = _agy_tree(mam_sandbox, conv)
+    assert _run_onboard(["--uuid", conv], mam_sandbox).returncode == 0
 
-    res = _run_onboard(["--remove", UUID, "--purge-conversation", "--yes"], mam_sandbox,
-                       extra_env={"CLAUDE_CODE_SESSION_ID": UUID})
+    res = _run_onboard(["--remove", conv, "--purge-conversation", "--yes"], mam_sandbox,
+                       extra_env={"CLAUDE_CODE_SESSION_ID": conv})
 
     assert res.returncode == 5, res.stdout + res.stderr
     assert "still in use" in res.stderr
     assert all(p.exists() for p in paths.values())
-    assert _registered(mam_sandbox, UUID), "a refused purge must leave the registration in place"
+    assert _registered(mam_sandbox, conv), "a refused purge must leave the registration in place"
 
 
-def test_orc_remove_purge_without_artifacts_still_unregisters(mam_sandbox):
-    assert _run_onboard(["--uuid", UUID], mam_sandbox).returncode == 0
-    res = _run_onboard(["--remove", UUID, "--purge-conversation", "--yes"], mam_sandbox)
+def test_orc_remove_purge_without_artifacts_still_unregisters(mam_sandbox, conv):
+    assert _run_onboard(["--uuid", conv], mam_sandbox).returncode == 0
+    res = _run_onboard(["--remove", conv, "--purge-conversation", "--yes"], mam_sandbox)
     assert res.returncode == 0, res.stdout + res.stderr
     assert "nothing to purge" in res.stderr
-    assert not _registered(mam_sandbox, UUID)
+    assert not _registered(mam_sandbox, conv)
 
 
-def test_orc_purge_flag_needs_remove(mam_sandbox):
-    res = _run_onboard(["--uuid", UUID, "--purge-conversation", "--yes"], mam_sandbox)
+def test_orc_purge_flag_needs_remove(mam_sandbox, conv):
+    res = _run_onboard(["--uuid", conv, "--purge-conversation", "--yes"], mam_sandbox)
     assert res.returncode == 2
-    assert not _registered(mam_sandbox, UUID)
+    assert not _registered(mam_sandbox, conv)
 
 
-def test_orc_dry_run_purge_changes_nothing(mam_sandbox):
-    paths = _agy_tree(mam_sandbox, UUID)
-    assert _run_onboard(["--uuid", UUID], mam_sandbox).returncode == 0
-    res = _run_onboard(["--remove", UUID, "--purge-conversation", "--yes", "--dry-run"], mam_sandbox)
+def test_orc_dry_run_purge_changes_nothing(mam_sandbox, conv):
+    paths = _agy_tree(mam_sandbox, conv)
+    assert _run_onboard(["--uuid", conv], mam_sandbox).returncode == 0
+    res = _run_onboard(["--remove", conv, "--purge-conversation", "--yes", "--dry-run"], mam_sandbox)
     assert res.returncode == 0
     assert "purge its conversation artifacts" in res.stdout
     assert all(p.exists() for p in paths.values())
-    assert _registered(mam_sandbox, UUID)
+    assert _registered(mam_sandbox, conv)