diff --git a/skills/ai-peer/SKILL.md b/skills/ai-peer/SKILL.md index 63919f6..f89e459 100644 --- a/skills/ai-peer/SKILL.md +++ b/skills/ai-peer/SKILL.md @@ -10,7 +10,7 @@ Peer-to-peer conversation rooms for AI CLI tools (Claude Code, Codex, OpenCode) - **Auth**: Local daemon: none. Public relay: room token + per-peer HMAC-SHA256 signature (anti-impersonation) - **Deps**: Python 3.10+ (stdlib only, zero pip deps) - **Relay**: `https://relay.ai-peer.chat` (Durable Objects, strong consistency) -- **Operations**: 16 (daemon 3, room 4, chat 2, invite 1, quick 1, discuss 1, peer 3, export 1) +- **Operations**: 16 sub-commands across 12 top-level commands (daemon: start/stop/status, room: create/join/list/delete, chat: send/read, invite, quick, discuss, discover, list, register, history, export, identity) - **Storage**: Local SQLite at `~/.ai-peers/peers.db`, daemon config at `~/.ai-peers/daemon.json` - **Tests**: 69 (unit + integration + crypto + auth) diff --git a/skills/ai-peer/references/PRD.md b/skills/ai-peer/references/PRD.md index 8bc4292..b124f0c 100644 --- a/skills/ai-peer/references/PRD.md +++ b/skills/ai-peer/references/PRD.md @@ -72,7 +72,7 @@ The closest attempt (claude-peers-mcp) is single-machine, Claude-only, and requi | Component | Tech | Purpose | |-----------|------|---------| | **Daemon** | Python stdlib `http.server` + SQLite | Local message store + HTTP API for CLI commands | -| **CLI** | Python `argparse` | 16 commands: daemon, room, chat, invite, quick, discuss, discover, etc. | +| **CLI** | Python manual dispatch | 12 top-level commands (daemon, room, chat, invite, quick, discuss, discover, list, register, history, export, identity) with sub-commands | | **Relay** | Cloudflare Worker + Durable Objects | Cross-internet message forwarding, strong consistency | | **Crypto** | PBKDF2 + Fernet (`cryptography` package) | Optional E2E encryption, relay stores only ciphertext | | **Spawner** | `subprocess.run` | Invoke AI CLI tools with conversation context as prompt | @@ -186,7 +186,7 @@ The core differentiator — bring any AI CLI tool into a conversation. ### P1: Quality & Reliability - SQLite WAL mode + threading locks for concurrent access - Cross-platform signal handling (SIGTERM/SIGBREAK) -- 40 automated tests (unit + integration + crypto) +- 69 automated tests (unit + integration + crypto + auth) - Grade A on skill-optimizer (40/40) - 3-AI deep review (Claude+Codex+OpenCode), 10 HIGH issues fixed diff --git a/skills/ai-peer/scripts/ai_peer/__init__.py b/skills/ai-peer/scripts/ai_peer/__init__.py index 2a5ec92..f01cab4 100644 --- a/skills/ai-peer/scripts/ai_peer/__init__.py +++ b/skills/ai-peer/scripts/ai_peer/__init__.py @@ -1,17 +1,29 @@ """AI Peer — decentralized AI-to-AI and human-AI communication.""" import json import sys +import os + +# Force UTF-8 output on Windows to handle Unicode characters in AI responses +if sys.platform == "win32": + os.environ.setdefault("PYTHONIOENCODING", "utf-8") + for stream_name in ("stdout", "stderr"): + stream = getattr(sys, stream_name, None) + if stream and hasattr(stream, "reconfigure"): + try: + stream.reconfigure(encoding="utf-8") + except Exception: + pass def output(ok, data=None, error=None): - """Standard output envelope.""" + """Standard output envelope. Returns exit code.""" result = {"ok": ok, "schema_version": "1"} if ok: result["data"] = data else: result["error"] = str(error) print(json.dumps(result, ensure_ascii=False, indent=2)) - sys.exit(0 if ok else 1) + return 0 if ok else 1 def main(): @@ -38,11 +50,11 @@ def main(): try: from .ops import run_command result = run_command(args) - output(True, result) + sys.exit(output(True, result)) except KeyboardInterrupt: sys.exit(130) except Exception as e: - output(False, error=e) + sys.exit(output(False, error=e)) if __name__ == "__main__": diff --git a/skills/ai-peer/scripts/ai_peer/client.py b/skills/ai-peer/scripts/ai_peer/client.py index 5030806..a6e5d17 100644 --- a/skills/ai-peer/scripts/ai_peer/client.py +++ b/skills/ai-peer/scripts/ai_peer/client.py @@ -29,11 +29,11 @@ def _req(self, method, path, data=None, timeout=30): except Exception: return {"error": f"HTTP {e.code}"} except (urllib.error.URLError, OSError): - return None # Daemon not running or connection failed + return {"error": "daemon not running or connection failed"} def is_alive(self): r = self._req("GET", "/health", timeout=3) - return r is not None and r.get("ok") + return isinstance(r, dict) and r.get("ok") # === Rooms === @@ -142,7 +142,12 @@ def ensure_daemon(host=None, port=None): # Need to restart with new config sys.stderr.write(f"⟳ Restarting daemon ({cfg_host}:{cfg_port} → {host}:{port})...\n") _stop_daemon() - time.sleep(0.5) + # Wait for old daemon to fully stop before starting new one + for _ in range(20): + time.sleep(0.2) + check_client = PeerClient(check_host, port) + if not check_client.is_alive(): + break else: return client @@ -187,6 +192,13 @@ def _start_daemon(host, port): PEERS_HOME.mkdir(parents=True, exist_ok=True) + # Rotate log if too large (>1MB) + if LOG_FILE.exists() and LOG_FILE.stat().st_size > 1_000_000: + rotated = LOG_FILE.with_suffix(".log.1") + if rotated.exists(): + rotated.unlink() + LOG_FILE.rename(rotated) + scripts_dir = str(Path(__file__).resolve().parent.parent) existing_pypath = os.environ.get("PYTHONPATH", "") pypath = scripts_dir + os.pathsep + existing_pypath if existing_pypath else scripts_dir diff --git a/skills/ai-peer/scripts/ai_peer/constants.py b/skills/ai-peer/scripts/ai_peer/constants.py index daa4fa8..ccaef6d 100644 --- a/skills/ai-peer/scripts/ai_peer/constants.py +++ b/skills/ai-peer/scripts/ai_peer/constants.py @@ -33,12 +33,19 @@ "env_unset": ["CLAUDECODE"], }, "opencode": { - "cmd": ["opencode", "-p", "{prompt}"], + "cmd": ["opencode", "run", "{prompt}"], "env_unset": [], }, } def get_machine_id(): - """Stable machine identifier.""" - return platform.node() + """Stable machine identifier with fallback chain.""" + name = platform.node() + if not name: + try: + import socket + name = socket.gethostname() + except Exception: + pass + return name or "unknown" diff --git a/skills/ai-peer/scripts/ai_peer/crypto.py b/skills/ai-peer/scripts/ai_peer/crypto.py index 015ac21..c14365a 100644 --- a/skills/ai-peer/scripts/ai_peer/crypto.py +++ b/skills/ai-peer/scripts/ai_peer/crypto.py @@ -50,10 +50,21 @@ def decrypt(ciphertext: str, password: str, room_id: str, strict: bool = False) f = Fernet(key) try: return f.decrypt(ciphertext.encode()).decode() - except Exception: + except InvalidToken: return "[encrypted — wrong password or corrupted]" def is_encrypted(text: str) -> bool: - """Check if text looks like a Fernet token.""" - return text.startswith("gAAA") and len(text) > 50 + """Check if text looks like a Fernet token by attempting to decode it.""" + if not text or not isinstance(text, str): + return False + if not HAS_CRYPTO: + return text.startswith("gAAA") and len(text) > 50 + try: + # Fernet tokens are URL-safe base64, decode to check structure + import base64 + decoded = base64.urlsafe_b64decode(text + "==") + # Fernet tokens start with version byte (0x80) + 8-byte timestamp + return len(decoded) > 9 and decoded[0] == 0x80 + except Exception: + return False diff --git a/skills/ai-peer/scripts/ai_peer/db.py b/skills/ai-peer/scripts/ai_peer/db.py index cb87fac..7846eaf 100644 --- a/skills/ai-peer/scripts/ai_peer/db.py +++ b/skills/ai-peer/scripts/ai_peer/db.py @@ -81,6 +81,8 @@ def __init__(self, db_path: Path): self.conn.executescript(SCHEMA) self._run_migrations() self._lock = threading.Lock() + # WAL mode allows concurrent reads without locks — only writes need the lock. + # Reads (get_room, get_messages, list_rooms, room_peers) are safe without _lock. def _run_migrations(self): for sql in MIGRATIONS: @@ -115,7 +117,7 @@ def create_room(self, name, mode="local", relay_url=None, password=None, room_id (rid, name, mode, relay_url, password, token, _now()), ) self.conn.commit() - return _row_to_dict(self.conn.execute("SELECT * FROM rooms WHERE id=?", (rid,)).fetchone()) + return _row_to_dict(self.conn.execute("SELECT * FROM rooms WHERE id=?", (rid,)).fetchone()) def get_room(self, room_id): return _row_to_dict(self.conn.execute("SELECT * FROM rooms WHERE id=?", (room_id,)).fetchone()) diff --git a/skills/ai-peer/scripts/ai_peer/helpers.py b/skills/ai-peer/scripts/ai_peer/helpers.py index 739c95e..7215748 100644 --- a/skills/ai-peer/scripts/ai_peer/helpers.py +++ b/skills/ai-peer/scripts/ai_peer/helpers.py @@ -143,9 +143,9 @@ def _merge_messages(local_resp, relay_resp): seen_sys.add(sys_key) elif not msg_id: # Non-system without ID: use content hash as fallback - fallback = hashlib.md5( + fallback = hashlib.sha256( f"{msg.get('peer_id', '')}:{msg.get('content', '')}:{msg.get('created_at', '')}".encode() - ).hexdigest() + ).hexdigest()[:16] if fallback in seen_ids: continue seen_ids.add(fallback) @@ -186,7 +186,13 @@ def _peer_signature(peer_secret, room_id): def get_or_create_identity(): """Get or create local human identity. Auto-migrates old PID-based IDs.""" - username = os.environ.get("USER", os.environ.get("USERNAME", "user")) + username = os.environ.get("USER", os.environ.get("USERNAME")) + if not username: + try: + import getpass + username = getpass.getuser() + except Exception: + username = "user" expected_id = f"{get_machine_id()}-human-{username}" if IDENTITY_FILE.exists(): diff --git a/skills/ai-peer/scripts/ai_peer/ops_ai.py b/skills/ai-peer/scripts/ai_peer/ops_ai.py index 95ee632..d9dc0fd 100644 --- a/skills/ai-peer/scripts/ai_peer/ops_ai.py +++ b/skills/ai-peer/scripts/ai_peer/ops_ai.py @@ -5,7 +5,7 @@ import secrets import sys -from .constants import get_machine_id +from .constants import get_machine_id, TOOL_COMMANDS from .client import ensure_daemon from .relay_client import RelayClient from .spawn import build_conversation_prompt, spawn_ai @@ -16,6 +16,14 @@ ) +def _safe_int(value, name): + """Parse int with clear error message.""" + try: + return int(value) + except (ValueError, TypeError): + raise ValueError(f"Invalid {name}: {value!r} (must be a number)") + + def cmd_invite(args): if not args: raise ValueError( @@ -34,7 +42,7 @@ def cmd_invite(args): elif args[i] == "--context" and i + 1 < len(args): context = args[i + 1]; i += 2 elif args[i] == "--timeout" and i + 1 < len(args): - timeout = int(args[i + 1]); i += 2 + timeout = _safe_int(args[i + 1], "timeout"); i += 2 elif not tool: tool = args[i]; i += 1 else: @@ -49,7 +57,10 @@ def cmd_invite(args): identity = get_or_create_identity() human_secret = identity.get("peer_secret", "") # Derive per-AI secret from human secret + tool name (true per-peer auth) - ai_secret = hmac.new(human_secret.encode(), tool.encode(), hashlib.sha256).hexdigest() if human_secret else "" + if human_secret and len(human_secret) >= 16: + ai_secret = hmac.new(human_secret.encode(), tool.encode(), hashlib.sha256).hexdigest() + else: + ai_secret = "" ai_peer_id = make_ai_peer_id(tool) client.register_peer(ai_peer_id, tool, "ai", tool, get_machine_id()) client.join_room(room_id, ai_peer_id) @@ -93,7 +104,9 @@ def cmd_invite(args): peer_name=tool, peer_tool=tool, peer_signature=sig, ) - mentions = re.findall(r"@(claude-code|codex|opencode)\b", response, re.IGNORECASE) + # Build dynamic @mention regex from known tools + tool_names = "|".join(re.escape(t) for t in TOOL_COMMANDS) + mentions = re.findall(rf"@({tool_names})\b", response, re.IGNORECASE) mentions = list(dict.fromkeys(m.lower() for m in mentions if m.lower() != tool.lower())) result = {"tool": tool, "response": response, "room": room_id} @@ -126,7 +139,7 @@ def cmd_quick(args): elif args[i] == "--name" and i + 1 < len(args): room_name = args[i + 1]; i += 2 elif args[i] == "--timeout" and i + 1 < len(args): - timeout = int(args[i + 1]); i += 2 + timeout = _safe_int(args[i + 1], "timeout"); i += 2 elif not question and not args[i].startswith("-"): question = args[i]; i += 1 else: @@ -168,8 +181,8 @@ def cmd_quick(args): for tool in tools: sys.stderr.write(f"⏳ Inviting {tool}...\n") try: - result = cmd_invite(["--tool", tool, "--room", room_id, - "--timeout", str(timeout)]) + invite_args = ["--tool", tool, "--room", room_id, "--timeout", str(timeout)] + result = cmd_invite(invite_args) responses[tool] = result.get("response", "") sys.stderr.write(f"✓ {tool} responded ({len(responses[tool])} chars)\n") except Exception as e: @@ -190,11 +203,11 @@ def cmd_discuss(args): """Run a multi-round discussion between AIs.""" if not args: raise ValueError( - "Usage: peer discuss --tools --rounds N [--room ] [--context \"topic\"]\n" + "Usage: peer discuss --tools --rounds N [--room ] [--name ] [--context \"topic\"]\n" " Without --room, creates a new room automatically." ) - room_id = tools_csv = context = None + room_id = tools_csv = context = room_name = None rounds = 2 timeout = 120 i = 0 @@ -204,11 +217,13 @@ def cmd_discuss(args): elif args[i] == "--tools" and i + 1 < len(args): tools_csv = args[i + 1]; i += 2 elif args[i] == "--rounds" and i + 1 < len(args): - rounds = int(args[i + 1]); i += 2 + rounds = _safe_int(args[i + 1], "rounds"); i += 2 elif args[i] == "--context" and i + 1 < len(args): context = args[i + 1]; i += 2 elif args[i] == "--timeout" and i + 1 < len(args): - timeout = int(args[i + 1]); i += 2 + timeout = _safe_int(args[i + 1], "timeout"); i += 2 + elif args[i] == "--name" and i + 1 < len(args): + room_name = args[i + 1]; i += 2 else: i += 1 @@ -222,7 +237,7 @@ def cmd_discuss(args): client = ensure_daemon() if not room_id: - room = client.create_room("discussion", "local") + room = client.create_room(room_name or "discussion", "local") if not room or room.get("error"): raise RuntimeError(f"Failed to create room: {(room or {}).get('error', 'daemon unreachable')}") room_id = room["id"] diff --git a/skills/ai-peer/scripts/ai_peer/ops_chat.py b/skills/ai-peer/scripts/ai_peer/ops_chat.py index 4952b9e..9a0593a 100644 --- a/skills/ai-peer/scripts/ai_peer/ops_chat.py +++ b/skills/ai-peer/scripts/ai_peer/ops_chat.py @@ -37,7 +37,11 @@ def cmd_chat(args): elif rest[i] in ("--interactive", "-i"): interactive = True; i += 1 elif rest[i] == "-n" and i + 1 < len(rest): - limit = int(rest[i + 1]); i += 2 + try: + limit = int(rest[i + 1]) + except ValueError: + raise ValueError(f"Invalid number for -n: {rest[i + 1]}") + i += 2 else: message_parts.append(rest[i]); i += 1 @@ -108,7 +112,7 @@ def _display_msg(msg): def _on_ws_message(data): if data.get("type") == "message" and data.get("data"): - msg = data["data"] + msg = dict(data["data"]) # Copy to avoid mutating shared data # WS delivers raw ciphertext — decrypt here (poll path decrypts in _read_messages) if password and msg.get("type") != "system": msg["content"] = decrypt(msg["content"], password, room_id, strict=True) @@ -136,8 +140,8 @@ def _on_ws_close(): ) ws_client.on_close = _on_ws_close ws_connected[0] = True - except Exception: - pass + except Exception as e: + sys.stderr.write(f" WebSocket unavailable: {e}\n") # Always fetch initial context regardless of WS status try: @@ -241,5 +245,6 @@ def poll(): running[0] = False if ws_client: ws_client.close() + poller.join(timeout=2) sys.stderr.write("\nLeft the room.\n") return {"status": "exited"} diff --git a/skills/ai-peer/scripts/ai_peer/ops_peer.py b/skills/ai-peer/scripts/ai_peer/ops_peer.py index 2bc22f9..b6957fa 100644 --- a/skills/ai-peer/scripts/ai_peer/ops_peer.py +++ b/skills/ai-peer/scripts/ai_peer/ops_peer.py @@ -32,15 +32,22 @@ def cmd_register(args): name, ptype, tool = args[0], args[1], args[2] host = port = None - for i, a in enumerate(args[3:], 3): - if a == "--host" and i + 1 < len(args): - host = args[i + 1] - if a == "--port" and i + 1 < len(args): - port = int(args[i + 1]) + i = 3 + while i < len(args): + if args[i] == "--host" and i + 1 < len(args): + host = args[i + 1]; i += 2 + elif args[i] == "--port" and i + 1 < len(args): + try: + port = int(args[i + 1]) + except ValueError: + raise ValueError(f"Invalid port: {args[i + 1]} (must be a number)") + i += 2 + else: + i += 1 client = ensure_daemon() - pid = f"{host or get_machine_id()}-{tool}" - return client.register_peer(pid, name, ptype, tool, host or get_machine_id(), host, port) + pid = make_ai_peer_id(tool) + return client.register_peer(pid, name, ptype, tool, get_machine_id(), host, port) def cmd_history(args): @@ -48,7 +55,10 @@ def cmd_history(args): limit = 50 for i, a in enumerate(args or []): if a == "-n" and i + 1 < len(args): - limit = int(args[i + 1]) + try: + limit = int(args[i + 1]) + except ValueError: + raise ValueError(f"Invalid number for -n: {args[i + 1]}") client = ensure_daemon() diff --git a/skills/ai-peer/scripts/ai_peer/ops_room.py b/skills/ai-peer/scripts/ai_peer/ops_room.py index 1b5dfef..4d95fe5 100644 --- a/skills/ai-peer/scripts/ai_peer/ops_room.py +++ b/skills/ai-peer/scripts/ai_peer/ops_room.py @@ -43,7 +43,10 @@ def cmd_daemon(args): port = DEFAULT_PORT for i, a in enumerate(args): if a == "--port" and i + 1 < len(args): - port = int(args[i + 1]) + try: + port = int(args[i + 1]) + except ValueError: + raise ValueError(f"Invalid port: {args[i + 1]} (must be a number)") if action == "start": host = LAN_HOST if lan else DEFAULT_HOST @@ -165,7 +168,7 @@ def cmd_room(args): probe_relay = RelayClient(join_relay_url) info = probe_relay.room_info(join_room_id) if not info or info.get("error"): - sys.stderr.write(f"⚠ Room '{join_room_id}' not yet active on relay. Joining anyway.\n") + sys.stderr.write(f" Note: Room '{join_room_id}' not yet active on relay. Joining anyway.\n") # Check if room already exists locally — update metadata, preserve messages existing = client.get_room(join_room_id) diff --git a/skills/ai-peer/scripts/ai_peer/relay_client.py b/skills/ai-peer/scripts/ai_peer/relay_client.py index 44956e9..e95e4cd 100644 --- a/skills/ai-peer/scripts/ai_peer/relay_client.py +++ b/skills/ai-peer/scripts/ai_peer/relay_client.py @@ -52,7 +52,7 @@ def join_room(self, room_id, peer_id, name, peer_type="ai", tool="", machine="", def send_message(self, room_id, peer_id, content, peer_name="", peer_tool="", msg_type="message", msg_id=None, peer_signature=""): data = { - "peer_id": peer_id, "peer_name": peer_name, "peer_tool": peer_tool, + "peer_id": peer_id, "peer_name": peer_name or "", "peer_tool": peer_tool or "", "content": content, "type": msg_type, } if msg_id: diff --git a/skills/ai-peer/scripts/ai_peer/server.py b/skills/ai-peer/scripts/ai_peer/server.py index a8095e8..64331fe 100644 --- a/skills/ai-peer/scripts/ai_peer/server.py +++ b/skills/ai-peer/scripts/ai_peer/server.py @@ -108,6 +108,10 @@ def handle_health(self): def handle_shutdown(self): """Graceful shutdown via HTTP — cross-platform alternative to SIGTERM.""" + # Localhost-only check to prevent remote DoS + client_ip = self.client_address[0] + if client_ip not in ("127.0.0.1", "::1", ""): + return self._json({"error": "shutdown only allowed from localhost"}, 403) self._json({"ok": True, "message": "shutting down"}) if self._server_ref: import threading @@ -162,6 +166,8 @@ def handle_send_message(self, room_id): return if not b.get("peer_id") or not b.get("content") or not isinstance(b.get("content"), str): return self._json({"error": "missing peer_id or content"}, 400) + if len(b["content"]) > 100 * 1024: # 100KB per-message limit + return self._json({"error": "message content too large (max 100KB)"}, 413) msg = self.db.add_message( room_id=room_id, peer_id=b["peer_id"], content=b["content"], msg_type=b.get("type", "message"), metadata=b.get("metadata"), @@ -200,8 +206,10 @@ def handle_list_peers(self): def run_server(host=DEFAULT_HOST, port=DEFAULT_PORT): """Run the daemon (blocking). Called as subprocess.""" PEERS_HOME.mkdir(parents=True, exist_ok=True) + log_level = os.environ.get("AI_PEER_LOG_LEVEL", "INFO").upper() logging.basicConfig( - level=logging.INFO, format="%(asctime)s %(levelname)s %(message)s", + level=getattr(logging, log_level, logging.INFO), + format="%(asctime)s %(levelname)s %(message)s", handlers=[logging.FileHandler(LOG_FILE), logging.StreamHandler(sys.stderr)], ) db = PeerDB(DB_PATH) @@ -224,8 +232,9 @@ def shutdown(sig, frame): if hasattr(signal, "SIGBREAK"): signal.signal(signal.SIGBREAK, shutdown) signal.signal(signal.SIGINT, shutdown) - PID_FILE.write_text(str(os.getpid())) logger.info(f"AI Peer daemon on {host}:{port}") + # Write PID after server is bound to avoid race on restart + PID_FILE.write_text(str(os.getpid()), encoding="utf-8") try: server.serve_forever() finally: diff --git a/skills/ai-peer/scripts/ai_peer/spawn.py b/skills/ai-peer/scripts/ai_peer/spawn.py index 30cb5a3..b64b6ab 100644 --- a/skills/ai-peer/scripts/ai_peer/spawn.py +++ b/skills/ai-peer/scripts/ai_peer/spawn.py @@ -1,10 +1,26 @@ """AI CLI spawning — invoke local AI tools and capture responses.""" +import atexit import os import shutil import subprocess import sys +import tempfile from .constants import TOOL_COMMANDS, get_machine_id +# Track temp files for cleanup on exit +_tmp_files = [] + + +def _cleanup_tmp_files(): + for f in _tmp_files: + try: + os.unlink(f) + except OSError: + pass + + +atexit.register(_cleanup_tmp_files) + def discover_tools(): """Discover locally installed AI CLI tools.""" @@ -65,6 +81,18 @@ def spawn_ai(tool_name, prompt, timeout=120): return None, f"'{binary}' not found in PATH. Install {tool_name} first." resolved = shutil.which(binary) or binary + # On Windows, npm .ps1/.CMD wrappers cause command-line length issues. + # Resolve to the actual .exe binary inside node_modules if possible. + if sys.platform == "win32" and resolved and resolved.endswith((".ps1", ".CMD", ".cmd")): + npm_dir = os.path.dirname(resolved) + node_modules = os.path.join(npm_dir, "node_modules") + if os.path.isdir(node_modules): + for pkg_dir in os.listdir(node_modules): + bin_dir = os.path.join(node_modules, pkg_dir, "bin") + exe = os.path.join(bin_dir, f"{binary}.exe") + if os.path.isfile(exe): + resolved = exe + break use_stdin = config.get("stdin_prompt", False) if use_stdin: @@ -72,15 +100,42 @@ def spawn_ai(tool_name, prompt, timeout=120): # e.g. ["claude", "-p", "{prompt}", "--output-format", "text"] # → ["claude", "-p", "--output-format", "text"] # claude -p (no arg) reads prompt from stdin - cmd = [resolved if part == binary else part for part in config["cmd"] if part != "{prompt}"] + cmd = [resolved if i == 0 else part for i, part in enumerate(config["cmd"]) if part != "{prompt}"] else: - # Fallback: pass prompt as argument (truncate to prevent ARG_MAX) + # Pass prompt as argument. On Windows, write to temp file if too long + # for command-line limit (~32K). Uses --file flag for opencode. max_arg = 32000 if sys.platform == "win32" else 131072 - safe_prompt = prompt[:max_arg] if len(prompt) > max_arg else prompt - cmd = [safe_prompt if part == "{prompt}" else (resolved if part == binary else part) for part in config["cmd"]] + _tmp_prompt_file = None + if len(prompt) > max_arg and sys.platform == "win32": + # Write prompt to temp file and use --file flag + tmp = tempfile.NamedTemporaryFile(mode="w", suffix=".md", delete=False, + encoding="utf-8") + tmp.write(prompt) + tmp.close() + _tmp_prompt_file = tmp.name + _tmp_files.append(_tmp_prompt_file) + cmd = [] + for part in config["cmd"]: + if part == "{prompt}": + cmd.extend(["--file", _tmp_prompt_file]) + elif part == binary: + cmd.append(resolved) + else: + cmd.append(part) + else: + safe_prompt = prompt[:max_arg] if len(prompt) > max_arg else prompt + cmd = [] + for part in config["cmd"]: + if part == "{prompt}": + cmd.append(safe_prompt) + elif part == binary: + cmd.append(resolved) + else: + cmd.append(part) # Clean environment env = os.environ.copy() + env["PYTHONIOENCODING"] = "utf-8" for var in config.get("env_unset", []): env.pop(var, None) diff --git a/skills/ai-peer/scripts/ai_peer/ws_client.py b/skills/ai-peer/scripts/ai_peer/ws_client.py index f771d91..f6a7b8f 100644 --- a/skills/ai-peer/scripts/ai_peer/ws_client.py +++ b/skills/ai-peer/scripts/ai_peer/ws_client.py @@ -15,7 +15,7 @@ import threading from urllib.parse import urlparse -from ai_peer.constants import VERSION +from .constants import VERSION class WebSocketClient: @@ -77,8 +77,8 @@ def connect(self): if "101" not in status_line: raise ConnectionError(f"WebSocket handshake failed: {status_line}") - # Remove handshake timeout — recv should block indefinitely for idle chat rooms - self._sock.settimeout(None) + # Use a moderate timeout so recv doesn't block forever if server dies + self._sock.settimeout(30.0) self._running = True self._thread = threading.Thread(target=self._recv_loop, daemon=True) @@ -96,7 +96,12 @@ def _recv_loop(self): """Read WebSocket frames in a loop.""" try: while self._running: - msg = self._read_frame() + try: + msg = self._read_frame() + except socket.timeout: + # Idle timeout — send ping to keep connection alive + self._send_ping(b"") + continue if msg is None: break try: @@ -200,6 +205,16 @@ def _send_pong(self, payload): except Exception: pass + def _send_ping(self, payload): + """Send a masked ping frame (RFC 6455: client frames MUST be masked).""" + mask = os.urandom(4) + masked = bytes(b ^ mask[i % 4] for i, b in enumerate(payload)) + frame = bytes([0x89, 0x80 | len(payload)]) + mask + masked + try: + self._sock.sendall(frame) + except Exception: + pass + def _recv_exact(self, n): """Read exactly n bytes, consuming from handshake buffer first.""" buf = b"" diff --git a/skills/ai-peer/tests/conftest.py b/skills/ai-peer/tests/conftest.py index 30e5ff8..1d8fc75 100644 --- a/skills/ai-peer/tests/conftest.py +++ b/skills/ai-peer/tests/conftest.py @@ -22,21 +22,36 @@ def daemon(): client = ensure_daemon() yield client - # Cleanup — use subprocess-agnostic termination (works on Windows too) + # Cleanup — try HTTP shutdown first (graceful), fall back to force kill import os import subprocess + import urllib.request + from ai_peer.client import _load_daemon_cfg + from ai_peer.constants import DEFAULT_HOST, LAN_HOST + if PID_FILE.exists(): try: pid = int(PID_FILE.read_text().strip()) - # os.kill + SIGTERM is not portable; use psutil-free approach - import signal as _sig - sig = getattr(_sig, "SIGTERM", None) - if sig is not None: - os.kill(pid, sig) - else: - # Windows fallback - subprocess.call(["taskkill", "/PID", str(pid), "/F"], - stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL) + # Try HTTP shutdown first (graceful, cross-platform) + try: + cfg_host, cfg_port = _load_daemon_cfg() + check_host = DEFAULT_HOST if cfg_host == LAN_HOST else cfg_host + req = urllib.request.Request( + f"http://{check_host}:{cfg_port}/api/shutdown", method="POST" + ) + urllib.request.urlopen(req, timeout=3) + except Exception: + # HTTP failed — fall back to signal/force kill + import signal as _sig + sig = getattr(_sig, "SIGTERM", None) + if sig is not None: + try: + os.kill(pid, sig) + except (ProcessLookupError, OSError): + pass + else: + subprocess.call(["taskkill", "/PID", str(pid), "/F"], + stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL) except (ProcessLookupError, ValueError, OSError): pass PID_FILE.unlink(missing_ok=True) diff --git a/skills/ai-peer/tests/test_db.py b/skills/ai-peer/tests/test_db.py index 66cc10e..eee2e9a 100644 --- a/skills/ai-peer/tests/test_db.py +++ b/skills/ai-peer/tests/test_db.py @@ -9,7 +9,9 @@ @pytest.fixture def db(): with tempfile.TemporaryDirectory() as tmp: - yield PeerDB(Path(tmp) / "test.db") + database = PeerDB(Path(tmp) / "test.db") + yield database + database.close() class TestRooms: