box/bin/muse_hybrid.py

#!/usr/bin/env python3
"""
muse_hybrid.py: Hybrid integration layer combining fast headless gateway actions
(via muse-cli-node with per-node Cloudflare WARP egress & cookies) with Chromebox
CDP DOM interactions.

Provides:
- muse_threads(agent) -> list of threads
- muse_history(agent, thread_id=None, limit=10) -> list of messages
- muse_send(agent, text, thread_id=None, wait=0) -> dict response
- muse_session_start(agent, title=None) -> dict response
"""

import subprocess
import json
import os
import sys
from pathlib import Path

BIN_DIR = Path(__file__).resolve().parent
MUSE_CLI_NODE = BIN_DIR / "muse-cli-node"

def is_node_configured(node):
    conf_dir = Path.home() / ".config" / "muse-cli" / node
    return (conf_dir / "cookies.txt").exists()

def run_muse_cli(node, args, timeout=60):
    cmd = [str(MUSE_CLI_NODE), node] + args
    res = subprocess.run(cmd, capture_output=True, text=True, timeout=timeout)
    return res.returncode, res.stdout, res.stderr

def get_threads(node):
    """Retrieve thread list via fast gateway. Returns (threads_list, error_str)."""
    rc, stdout, stderr = run_muse_cli(node, ["threads"])
    if rc != 0:
        return None, stderr or stdout
    try:
        data = json.loads(stdout)
        return data, None
    except Exception as e:
        return None, f"JSON parse error: {e}"

def start_session(node, title=None):
    """Start a new session/subagent via fast gateway. Returns (dict, error_str)."""
    args = ["session-start"]
    if title:
        args.extend(["--title", title])
    rc, stdout, stderr = run_muse_cli(node, args)
    if rc != 0:
        return None, stderr or stdout
    try:
        data = json.loads(stdout)
        return data, None
    except Exception as e:
        return None, f"JSON parse error: {e}"

def get_history(node, thread_id=None, limit=15):
    """Retrieve message history via fast gateway. Returns (history_list, error_str)."""
    args = ["history", "--limit", str(limit)]
    if thread_id and thread_id != "main":
        args.extend(["--thread", str(thread_id)])
    rc, stdout, stderr = run_muse_cli(node, args)
    if rc != 0:
        return None, stderr or stdout
    try:
        data = json.loads(stdout)
        return data, None
    except Exception as e:
        return None, f"JSON parse error: {e}"

def send_message(node, text, thread_id=None, wait=0):
    """Send message via fast gateway. Returns (result_dict, error_str)."""
    args = ["send"]
    if thread_id and thread_id != "main":
        args.extend(["--thread", str(thread_id)])
    args.extend(["--wait", str(wait)])
    args.append(text)
    rc, stdout, stderr = run_muse_cli(node, args, timeout=max(60, wait + 30))
    if rc != 0:
        return None, stderr or stdout
    try:
        data = json.loads(stdout)
        return data, None
    except Exception as e:
        return {"raw": stdout}, None

if __name__ == "__main__":
    if len(sys.argv) < 3:
        print("Usage: muse_hybrid.py <node> <threads|history|send> [args...]")
        sys.exit(1)
    node = sys.argv[1]
    action = sys.argv[2]
    if action == "threads":
        threads, err = get_threads(node)
        if err:
            print("Error:", err, file=sys.stderr)
            sys.exit(1)
        print(json.dumps(threads, indent=2))
    elif action == "history":
        tid = sys.argv[3] if len(sys.argv) > 3 else None
        msgs, err = get_history(node, tid)
        if err:
            print("Error:", err, file=sys.stderr)
            sys.exit(1)
        print(json.dumps(msgs, indent=2))
    elif action == "send":
        txt = sys.argv[3]
        tid = sys.argv[4] if len(sys.argv) > 4 else None
        res, err = send_message(node, txt, tid)
        if err:
            print("Error:", err, file=sys.stderr)
            sys.exit(1)
        print(json.dumps(res, indent=2))