multi-agent-mux/.agents/skills/lib_py/agents/__main__.py

# __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 <facts|resolve|purge> [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 <agent|auto> <uuid> [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 <agent|auto> <uuid>", 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 <agent> <uuid> [workspace] [--wait <sec>]  (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"}, and while in use where to look on
        # stderr. 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 <agent> <uuid>", 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}))
        if in_use:
            print("  What may hold it:", file=sys.stderr)
            for hint in adapter.in_use_hints(uuid, ctx):
                print(f"    {hint}", file=sys.stderr)
        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()