#!/usr/bin/env python3 from __future__ import annotations import argparse import datetime as dt import hashlib import json import os import sys from pathlib import Path from typing import Any VERSION = "2026.05.27-codee-remote-host-foundation" ALLOWLIST = { "doctor", "heartbeat", "scan", "portal", "support", "test lane", "preview dry-run", } BLOCKED_ACTIONS = { "raw shell", "secret export", "destructive cleanup", "silent screen streaming", "owner credential access", } def now() -> str: return dt.datetime.now(dt.timezone.utc).isoformat() def write_json(path: Path, payload: dict[str, Any], *, force: bool = True) -> bool: path.parent.mkdir(parents=True, exist_ok=True) if path.exists() and not force: return False path.write_text(json.dumps(payload, indent=2, sort_keys=True) + "\n", encoding="utf-8") return True def read_json(path: Path) -> dict[str, Any]: if not path.exists(): return {} try: parsed = json.loads(path.read_text(encoding="utf-8")) except Exception: return {} return parsed if isinstance(parsed, dict) else {} def append_event(root: Path, event: str, payload: dict[str, Any] | None = None) -> Path: path = root / ".codee" / "remote-host" / "events.jsonl" path.parent.mkdir(parents=True, exist_ok=True) record = { "schema": "codee.remote-host.event.v1", "created_at": now(), "event": event, "host_version": VERSION, "payload": payload or {}, } with path.open("a", encoding="utf-8") as handle: handle.write(json.dumps(record, sort_keys=True) + "\n") return path def device_id(root: Path, device_name: str) -> str: seed = f"{root.resolve()}:{device_name}" return "cdev_" + hashlib.sha256(seed.encode("utf-8")).hexdigest()[:16] def command_install(args: argparse.Namespace) -> int: root = Path(args.path).expanduser().resolve() device_name = str(args.device_name or (os.uname().nodename if hasattr(os, "uname") else "Codee Desktop")).strip() payload = { "ok": True, "schema": "codee.remote-host.v1", "version": VERSION, "created_at": now(), "workspace": str(root), "device_name": device_name, "device_id": device_id(root, device_name), "allowlisted_commands": sorted(ALLOWLIST), "blocked_actions": sorted(BLOCKED_ACTIONS), "screen_streaming": "disabled_until_security_gate_passes", "input_control": "disabled_until_explicit_desktop_consent", "queue_path": ".codee/remote-host/command-queue.json", "state_path": ".codee/remote-host/host-state.json", } if args.write: write_json(root / ".codee" / "remote-host" / "host-state.json", payload) write_json(root / ".codee" / "remote-host" / "command-queue.json", {"schema": "codee.remote-command-queue.v1", "commands": []}) append_event(root, "RemoteHostInstalled", {"device_id": payload["device_id"], "device_name": device_name}) if args.json: print(json.dumps(payload, indent=2)) else: print("Codee Remote Host") print(f"version={VERSION}") print(f"device_id={payload['device_id']}") print(f"screen_streaming={payload['screen_streaming']}") if args.write: print(f"state={root / '.codee' / 'remote-host' / 'host-state.json'}") return 0 def command_status(args: argparse.Namespace) -> int: root = Path(args.path).expanduser().resolve() state = read_json(root / ".codee" / "remote-host" / "host-state.json") queue = read_json(root / ".codee" / "remote-host" / "command-queue.json") commands = queue.get("commands") if isinstance(queue.get("commands"), list) else [] payload = { "ok": bool(state), "schema": "codee.remote-host-status.v1", "created_at": now(), "host_version": VERSION, "state_exists": bool(state), "device_id": state.get("device_id", ""), "pending_commands": len(commands), "allowlisted_commands": sorted(ALLOWLIST), "screen_streaming": state.get("screen_streaming", "not_installed"), } if args.json: print(json.dumps(payload, indent=2)) else: print(f"Codee Remote Host status: ok={payload['ok']}") print(f"device_id={payload['device_id']}") print(f"pending_commands={payload['pending_commands']}") print(f"screen_streaming={payload['screen_streaming']}") return 0 if payload["ok"] else 1 def command_run_once(args: argparse.Namespace) -> int: root = Path(args.path).expanduser().resolve() queue_path = root / ".codee" / "remote-host" / "command-queue.json" queue = read_json(queue_path) commands = queue.get("commands") if isinstance(queue.get("commands"), list) else [] processed: list[dict[str, Any]] = [] remaining: list[dict[str, Any]] = [] for item in commands: command = str(item.get("command") if isinstance(item, dict) else item).strip() allowed = command in ALLOWLIST result = { "command": command, "allowed": allowed, "status": "accepted-dry-run" if allowed else "rejected", "note": "Foundation host records queue results. Actual command execution stays backend-gated.", "processed_at": now(), } processed.append(result) append_event(root, "RemoteHostCommandProcessed", result) if not allowed: remaining.append(item if isinstance(item, dict) else {"command": command}) write_json(queue_path, {"schema": "codee.remote-command-queue.v1", "commands": remaining, "processed": processed[-25:]}) if args.json: print(json.dumps({"ok": True, "processed": processed, "remaining": remaining}, indent=2)) else: print(f"Codee Remote Host run-once: processed={len(processed)} remaining={len(remaining)}") return 0 def command_audit(args: argparse.Namespace) -> int: root = Path(args.path).expanduser().resolve() event_path = root / ".codee" / "remote-host" / "events.jsonl" events = [] if event_path.exists(): for line in event_path.read_text(encoding="utf-8", errors="replace").splitlines()[-100:]: try: events.append(json.loads(line)) except Exception: continue payload = {"ok": True, "schema": "codee.remote-host-audit.v1", "created_at": now(), "event_count": len(events), "latest_events": events[-10:]} write_json(root / ".codee" / "remote-host" / "latest-audit.json", payload) if args.json: print(json.dumps(payload, indent=2)) else: print(f"Codee Remote Host audit: events={len(events)}") print(f"report={root / '.codee' / 'remote-host' / 'latest-audit.json'}") return 0 def build_parser() -> argparse.ArgumentParser: parser = argparse.ArgumentParser(description="Codee Remote Host foundation. No silent screen streaming or public owner credentials.") parser.add_argument("--version", action="version", version=VERSION) sub = parser.add_subparsers(dest="command", required=True) install = sub.add_parser("install", help="Install local host state and queue files.") install.add_argument("path", nargs="?", default=".") install.add_argument("--device-name", default="") install.add_argument("--write", action="store_true") install.add_argument("--json", action="store_true") install.set_defaults(func=command_install) status = sub.add_parser("status", help="Show local host status.") status.add_argument("path", nargs="?", default=".") status.add_argument("--json", action="store_true") status.set_defaults(func=command_status) run_once = sub.add_parser("run-once", help="Process one local command queue pass in dry-run mode.") run_once.add_argument("path", nargs="?", default=".") run_once.add_argument("--json", action="store_true") run_once.set_defaults(func=command_run_once) audit = sub.add_parser("audit", help="Summarize host events.") audit.add_argument("path", nargs="?", default=".") audit.add_argument("--json", action="store_true") audit.set_defaults(func=command_audit) return parser def main(argv: list[str] | None = None) -> int: parser = build_parser() args = parser.parse_args(argv) return int(args.func(args)) if __name__ == "__main__": raise SystemExit(main())