# __main__.py — CLI bridge for lib_py.agents facts and resolution (N4) import os, sys, json, shlex, time from lib_py.agents.registry import get_adapter, own_key, agent_of_row # `purge auto` probes adapters in this order; opencode last because its # artifact path is one shared DB that exists for every uuid. PURGE_PROBE_ORDER = ('agy', 'claude', 'hermes', 'grok', 'codex', 'opencode') def _resolve_purge_target(agent_hint: str, uuid: str, workspace: str): """(agent, adapter, ctx) whose on-disk artifact carries `uuid`, else None. An explicit agent is trusted as long as its artifact exists. `auto` tries each adapter's own verification first (revalidate mode: an orchestrator conversation is not in any workspace cache), then the adapter's own `has_artifacts` - which, for agy, also accepts orphaned sidecars, the annotation or the presence lock left behind after the DB is gone. """ from lib_py.agents.base import DiscoveryContext names = [agent_hint] if agent_hint and agent_hint != 'auto' else list(PURGE_PROBE_ORDER) probes = [] for name in names: adapter = get_adapter(name) if not adapter: continue probes.append((name, adapter, DiscoveryContext(workspace, name, mode="revalidate"))) for name, adapter, ctx in probes: try: if adapter.verify_artifact(uuid, ctx): return name, adapter, ctx except Exception: pass for name, adapter, ctx in probes: try: if adapter.has_artifacts(uuid, ctx): return name, adapter, ctx except Exception: continue return None def main(): if len(sys.argv) < 2: print("Usage: python -m lib_py.agents [args...]", file=sys.stderr) sys.exit(1) cmd = sys.argv[1] if cmd == 'facts': agent_name = sys.argv[2] if len(sys.argv) > 2 else '' adapter = get_adapter(agent_name) if not adapter: print(f"ERROR: Unknown agent {agent_name!r}", file=sys.stderr) sys.exit(1) # Emit shell-eval friendly facts q = shlex.quote print(f"MAM_AGENT_NAME={q(adapter.name)}") print(f"MAM_OWN_KEY={q(adapter.own_key)}") print(f"MAM_INPUT_PROMPT={q(adapter.input_prompt or '')}") print(f"MAM_INPUT_PLACEHOLDER={q(adapter.input_placeholder or '')}") print(f"MAM_INPUT_RULE_PATTERN={q(adapter.input_rule_pattern or '')}") print(f"MAM_INPUT_FRAME_CHARS={q(adapter.input_frame_chars)}") print(f"MAM_INPUT_BLOCK_PREFIX={q(adapter.input_block_prefix or '')}") print(f"MAM_INPUT_STOP_PATTERN={q(adapter.input_stop_pattern or '')}") print(f"MAM_READY_TOKENS={q(adapter.ready_tokens)}") print(f"MAM_STRONG_READY_TOKENS={q(adapter.strong_ready_tokens)}") print(f"MAM_WEAK_READY_TOKENS={q(adapter.weak_ready_tokens)}") print(f"MAM_MODAL_TOKENS={q(adapter.modal_tokens or '')}") print(f"MAM_STARTUP_DIALOGS={q(json.dumps(adapter.startup_dialogs))}") print(f"MAM_RESIDENT_CHILDREN={q(json.dumps(list(adapter.resident_children)))}") print(f"MAM_AGENT_STATUS_RELIABLE={1 if adapter.agent_status_reliable else 0}") print(f"MAM_EXIT_KEY={q(adapter.exit_key)}") print(f"MAM_DELEGATE_AGENT_KEY={q(adapter.delegate_agent_key)}") elif cmd == 'resolve': name = sys.argv[2] if len(sys.argv) > 2 else '' resolved = agent_of_row({}, session_name=name) if resolved: print(resolved) sys.exit(0) sys.exit(1) elif cmd == 'input-region': agent_name = sys.argv[2] if len(sys.argv) > 2 else '' pane_text = sys.stdin.read() from lib_py.agents.input_region import extract_input_region try: region = extract_input_region(pane_text, agent_name) print(region) sys.exit(0) except Exception as ex: print(f"ERROR: {ex}", file=sys.stderr) sys.exit(5) elif cmd == 'spawn-spec': agent_name = sys.argv[2] if len(sys.argv) > 2 else '' binary = sys.argv[3] if len(sys.argv) > 3 else agent_name uuid = sys.argv[4] if len(sys.argv) > 4 else '' use_wrapper = (sys.argv[5].lower() in ('1', 'true', 'yes')) if len(sys.argv) > 5 else False adapter = get_adapter(agent_name) if not adapter: sys.exit(1) print(adapter.spawn_spec(binary, uuid, use_wrapper=use_wrapper)) sys.exit(0) elif cmd == 'resume-spec': agent_name = sys.argv[2] if len(sys.argv) > 2 else '' binary = sys.argv[3] if len(sys.argv) > 3 else agent_name uuid = sys.argv[4] if len(sys.argv) > 4 else '' workspace = sys.argv[5] if len(sys.argv) > 5 else '' adapter = get_adapter(agent_name) if not adapter: sys.exit(1) from lib_py.agents.base import DiscoveryContext ctx = DiscoveryContext(workspace, agent_name) if workspace else None mat = adapter.verify_artifact(uuid, ctx) if (ctx and uuid) else False print(adapter.resume_spec(binary, uuid, materialized=mat)) sys.exit(0) elif cmd == 'exit-key': agent_name = sys.argv[2] if len(sys.argv) > 2 else '' adapter = get_adapter(agent_name) if adapter: print(adapter.exit_key) sys.exit(0) print('/exit') sys.exit(0) elif cmd == 'purge': # purge [workspace] (Issue #13) # Deletes a conversation's on-disk artifacts through its adapter and # prints one JSON object: {"agent", "purged", "residual", "in_use"}. # Exit 0 purged; 4 refused, conversation still in use (nothing # removed); 5 no adapter has an artifact for this uuid. agent_hint = sys.argv[2] if len(sys.argv) > 2 else 'auto' uuid = sys.argv[3] if len(sys.argv) > 3 else '' workspace = sys.argv[4] if len(sys.argv) > 4 and sys.argv[4] else os.getcwd() if not uuid: print("ERROR: purge requires ", file=sys.stderr) sys.exit(1) target = _resolve_purge_target(agent_hint, uuid, workspace) if not target: print(json.dumps({"agent": None, "purged": [], "residual": [], "in_use": False})) sys.exit(5) name, adapter, ctx = target if adapter.conversation_in_use(uuid, ctx): print(json.dumps({"agent": name, "purged": [], "residual": [], "in_use": True})) sys.exit(4) purged = adapter.purge_artifacts(uuid, ctx) residual = adapter.residual_artifacts(uuid, ctx) print(json.dumps({"agent": name, "purged": purged, "residual": residual, "in_use": False})) sys.exit(0) elif cmd == 'in-use': # in-use [workspace] [--wait ] (Issue #31) # The purge CLI's own in-use verdict, for stop_session.sh to ask # before it purges. Re-checks every 0.25 s for up to --wait seconds. # Prints {"agent", "in_use"}. Exit 0 free; 4 still in use; 1 bad args. args = sys.argv[2:] wait = 0.0 if '--wait' in args: i = args.index('--wait') try: wait = float(args[i + 1]) except (IndexError, ValueError): print("ERROR: --wait requires seconds", file=sys.stderr) sys.exit(1) del args[i:i + 2] agent_name = args[0] if args else '' uuid = args[1] if len(args) > 1 else '' workspace = args[2] if len(args) > 2 and args[2] else os.getcwd() adapter = get_adapter(agent_name) if not adapter or not uuid: print("ERROR: in-use requires ", file=sys.stderr) sys.exit(1) from lib_py.agents.base import DiscoveryContext ctx = DiscoveryContext(workspace, agent_name) deadline = time.monotonic() + wait in_use = adapter.conversation_in_use(uuid, ctx) while in_use and time.monotonic() < deadline: time.sleep(0.25) in_use = adapter.conversation_in_use(uuid, ctx) print(json.dumps({"agent": agent_name, "in_use": in_use})) sys.exit(4 if in_use else 0) else: print(f"ERROR: Unknown CLI command {cmd!r}", file=sys.stderr) sys.exit(1) if __name__ == '__main__': main()