hermes-agent/tools/bot_mode_dm_coordination.py

"""``message_agent`` coordination: a fan-out to several teammates and cancelling one's own request.

Fan-out: each target is resolved and admitted exactly like a single ``message_agent`` call, but its
reply runner is not started on its own. One combined runner (``bot_mode_dm.py --fan-out <spec>``)
runs those per-target runners side by side and prints ONE result once every one has settled or hit
its bound, so the sender wakes once with a typed outcome per teammate instead of N separate turns.

Cancel: the sender withdraws its own queued request, or stops the turn that request started, on the
teammate's owner through the same ``prompt.cancel`` a viewer's Stop uses. Only the request's author
may: the receipt records who sent it.

The runner half (``run_fan_out``) is stdlib only: it runs as a background process whose stdout is
the sender's completion notification, and each sub-runner boots Hermes itself.
"""

from __future__ import annotations

import contextlib
import json
import logging
import os
import re
import shlex
import subprocess
import sys
import tempfile
import threading
from pathlib import Path
from typing import Any

# Every sub-runner bounds its own wait (the live lane at 30 min, the relay waiter by its budget);
# this only reaps one that outlives that, reported as pending rather than lost.
FAN_OUT_WAIT_SECONDS = 32 * 60
_HEX_ID_RE = re.compile(r"[0-9a-f]{32,64}")
logger = logging.getLogger(__name__)


def split_targets(target: str, sender: Any) -> list[str]:
    """``target`` as the recipients it names: one name, or a comma-separated list deduplicated by
    the destination each name reaches (``hermes`` and ``default`` are one bot; a registered peer
    and a local bot whose friendly name matches it are two). A friendly name that itself contains
    a comma ('QA, Release') still resolves whole."""
    from tools.bot_mode_dm import _resolve_local_name

    raw = str(target or "").strip().lstrip("@")
    if "," not in raw or _resolve_local_name(raw, list(sender.roster_homes), sender.root) is not None:
        return [raw] if raw else []
    named: dict[tuple[str, str], str] = {}
    for part in raw.split(","):
        name = part.strip().lstrip("@")
        if name:
            named.setdefault(_destination(name, sender), name)
    return list(named.values())


def _destination(name: str, sender: Any) -> tuple[str, str]:
    """``(transport, recipient)`` for ``name``, resolved in ``_dispatch``'s order: a peer target
    first, then a connection-qualified relay name, then a local bot (folder id or friendly name)."""
    from tools.bot_mode_dm import _PEER_TARGET_RE, _resolve_local_name

    peer = _PEER_TARGET_RE.match(name)
    if peer or name.lower() in sender.peers:
        peer_name, peer_profile = peer.groups() if peer else (name.lower(), None)
        return "peer", f"{peer_name}/{peer_profile or ''}".lower()
    if "@" in name:
        return "relay", name.lower()
    local = _resolve_local_name(name, list(sender.roster_homes), sender.root)
    return ("local", local) if local is not None else ("unresolved", name.lower())


def fan_out(targets: list[str], content: str, sender: Any, *, expects_reply: bool) -> str:
    """Dispatch ``content`` to every target; with a reply expected, start the one combined runner."""
    from tools.bot_mode_dm import _dispatch

    collected: list[dict] = []
    acks = []
    for target in targets:
        try:
            ack = json.loads(_dispatch(target, content, sender, wake=collected if expects_reply else "none"))
        except Exception as exc:
            # One target's setup failure stays that target's: requests already admitted still get
            # their combined waiter below.
            logger.debug("message_agent fan-out to %s failed", target, exc_info=True)
            ack = {"error": f"NOT sent: {exc}"}
        acks.append({"to": f"@{target}", **ack})
    sent = [ack for ack in acks if not ack.get("error")]
    if not collected:
        return json.dumps({"status": "queued" if sent else "failed", "fan_out": acks,
                           "detail": ("Nothing to wait for. " if sent else "Nothing was sent. ")
                           + "Do not resend to a teammate whose entry has no error."})
    started = _start_combined(collected, sender)
    if started.get("error"):
        # The local admissions stand (their owner runs them); only the joint wake is lost. A peer
        # entry was never sent: its runner owned the send.
        for ack in acks:
            if any(entry["delivery_id"] == ack.get("delivery_id") and entry["dm_file"] for entry in collected):
                ack.update(status="failed", error="NOT sent: the fan-out runner did not start")
        return json.dumps({"status": "queued", "fan_out": acks, "notification_error": started["error"],
                           "detail": "The replies will NOT wake you. Do not resend to entries without an error."})
    detail = (f"One combined completion carries every reply (or why not) once all {len(collected)} settle "
              "or hit their bound — do NOT wait, poll or resend; finish your turn.")
    if started.get("reply_delivery") == "poll":
        detail = started["detail"]
    return json.dumps({"status": "queued", "fan_out": acks, "process_id": started.get("process_id"),
                       "reply_delivery": started.get("reply_delivery"), "detail": detail})


def _start_combined(collected: list[dict], sender: Any) -> dict:
    from tools.bot_mode_dm import REPLY_COMPLETION_CHARS, _dm_dir, _spawn_delivery, _unlink_dm_file

    fd, spec = tempfile.mkstemp(prefix="fanout-", suffix=".json", dir=_dm_dir(), text=True)
    with os.fdopen(fd, "w", encoding="utf-8") as stream:
        json.dump([{**{k: entry[k] for k in ("command", "label", "delivery_id")},
                    "output": _runner_output(entry["command"])} for entry in collected], stream)
    argv = [sys.executable, str(Path(__file__).resolve().parent / "bot_mode_dm.py"), "--fan-out", spec]
    if sys.platform == "win32":
        argv = [part.replace("\\", "/") for part in argv]  # Git Bash runs the tracked command (see _delivery_command)
    started = json.loads(_spawn_delivery(shlex.join(argv), f"{len(collected)} teammates", dm_file=spec,
                                         task_id=sender.task_id, agent=sender.agent,
                                         completion_chars=REPLY_COMPLETION_CHARS * len(collected)))
    if started.get("error"):
        for entry in collected:
            if entry["dm_file"]:
                _unlink_dm_file(entry["dm_file"])
    return started


def _runner_output(command: str) -> str:
    """How a sub-runner reports: the live lane (``--run-delivery query-file``) prints one JSON
    payload; the relay waiter and the ``hermes peer dm`` transport print the reply as text."""
    from tools.bot_mode_dm import _runner_argv

    argv = shlex.split(command)
    marker = argv.index("--run-delivery") if "--run-delivery" in argv else len(argv)
    parsed = _runner_argv(argv[marker:])
    return "json" if parsed is not None and parsed[1] == "query-file" else "text"


def run_fan_out(spec_path: str) -> int:
    """Runner entry: run each target's runner concurrently, print one JSON outcome list."""
    with open(spec_path, encoding="utf-8-sig") as stream:
        entries = json.load(stream)
    with contextlib.suppress(OSError):
        os.unlink(spec_path)
    results: list[dict] = [{} for _ in entries]

    def run(index: int, entry: dict) -> None:
        base = {"to": entry["label"], "delivery_id": entry["delivery_id"]}
        try:
            done = subprocess.run(shlex.split(entry["command"]), stdin=subprocess.DEVNULL, capture_output=True,
                                  text=True, encoding="utf-8", errors="replace", timeout=FAN_OUT_WAIT_SECONDS,
                                  check=False)
        except subprocess.TimeoutExpired:
            results[index] = {**base, "status": "pending", "detail": "No outcome within the bound. Do not resend."}
            return
        except (OSError, ValueError) as exc:
            results[index] = {**base, "status": "failed", "error": f"Its delivery runner could not start: {exc}"}
            return
        results[index] = {**base, **_outcome(entry.get("output", "text"), done.returncode, done.stdout, done.stderr)}

    threads = [threading.Thread(target=run, args=item, daemon=True) for item in enumerate(entries)]
    for thread in threads:
        thread.start()
    for thread in threads:
        thread.join()
    settled = sum(result.get("status") == "settled" for result in results)
    from tools.bot_mode_dm_outcome import combined_summary

    for result in results:
        result.pop("outcome", None)  # the summary below says it once, by name
    print(json.dumps({"outcome": combined_summary(results), "replies": results, "settled": settled,
                      "of": len(results)}, ensure_ascii=False))
    return 0 if settled == len(results) else 1


def _outcome(output: str, returncode: int, stdout: str, stderr: str = "") -> dict:
    """A sub-runner's result as a typed outcome. Only a ``json`` runner's stdout is a delivery
    record; a ``text`` runner's successful stdout is the teammate's answer verbatim (even one that
    looks like JSON). A failing text runner may still have been accepted: every runner says "do not
    resend" exactly when the request is still in flight, which is pending, never failed."""
    text = stdout.strip()
    if output == "json" or returncode != 0:
        try:
            parsed = json.loads(text.splitlines()[-1]) if text else None
        except ValueError:
            parsed = None
        if isinstance(parsed, dict):
            outcome = {k: v for k, v in parsed.items() if k != "delivery_id"}
            outcome.setdefault("status", "failed" if parsed.get("error") else "settled")
            return outcome
    if returncode == 0:
        return {"status": "settled", "reply": text or "(empty reply)"}
    said = "\n".join(part for part in (text, stderr.strip()) if part)
    if "do not resend" in said.lower():
        return {"status": "pending", "detail": said}
    return {"status": "failed", "error": said or f"delivery runner exited {returncode}"}


def cancel_own_delivery(delivery_id: str, raw_target: str, sender: Any) -> str:
    """Withdraw the caller's own queued request, or stop the turn it started, on a local teammate."""
    from tools.bot_live_delivery import _TERMINAL, _authority_rpc, read_delivery_result
    from tools.bot_mode_dm import _err, _resolve_local_name

    if not _HEX_ID_RE.fullmatch(delivery_id):
        return _err("cancel takes the delivery_id from your own message_agent ack. Do not retry.")
    named = raw_target.strip().lstrip("@")
    resolved = _resolve_local_name(named, list(sender.roster_homes), sender.root) if named else None
    homes = ([sender.roster_homes[resolved]] if resolved else
             [home for name, home in sender.roster_homes.items() if name != sender.me])
    found = next(((home, record) for home in homes if (record := read_delivery_result(home, delivery_id))), None)
    if found is None:
        return _err("No request with that delivery_id to a teammate on this install (requests to another "
                    "machine cannot be cancelled). Do not retry.")
    home, record = found
    if (record.get("author") or {}).get("id") != f"bot:{sender.me}":
        return _err("Only the agent that sent a request can cancel it. Do not retry.")
    if record["status"] in _TERMINAL:
        return json.dumps({"status": record["status"], "delivery_id": delivery_id,
                           "detail": "That request already finished; nothing to cancel."})
    if not record.get("admission_id"):
        return _err("That request has no admission to cancel (its outcome is unknown). Do not retry.")
    try:
        # The owner decides which admission is the request now, under the lock that admits its one
        # automatic retry: a retry not admitted yet is withdrawn instead of racing this cancel.
        answer = _authority_rpc(home, "bot_relay.cancel", {
            "id": delivery_id, "author_id": f"bot:{sender.me}",
            "profile": Path(home).name if Path(home).parent.name == "profiles" else "default"})
    except Exception as exc:
        # The owner may be down or refuse: the sender gets that as a tool error, never a crash.
        logger.debug("message_agent cancel of %s failed", delivery_id, exc_info=True)
        return _err(f"Cancel failed: {exc}")
    detail = {"cancelled": "Withdrawn before it ran (or before its automatic retry); it will not run.",
              "stopping": "Its turn was running and was told to stop; its delivery process reports how it ended.",
              }.get(answer.get("status"), "That request already finished; nothing to cancel.")
    status = answer.get("status") if answer.get("status") in {"cancelled", "stopping"} else answer.get("receipt_status")
    return json.dumps({"status": status or "terminal", "delivery_id": delivery_id, "detail": detail})