from __future__ import annotations import argparse import base64 import concurrent.futures import json import platform import os import pathlib import shutil import sqlite3 import stat import subprocess import sys import tempfile import threading import time import urllib.error import urllib.parse import urllib.request import zipfile CONNECTOR_STATE_PATH = pathlib.Path( os.environ.get("SISAGA_ABCD_CONNECTOR_STATE_PATH") or pathlib.Path.home() / ".sisaga" / "abcd-connector.json" ) CONNECTOR_EXECUTION_SCHEMA = "sisaga.independent_ab.connector_execution.v1" CONNECTOR_ARCHIVE_MAX_BYTES = 22_000_000 CONNECTOR_ARCHIVE_EXPANDED_MAX_BYTES = 100_000_000 CONNECTOR_ARCHIVE_EXCLUDED_DIR_NAMES = {".git", ".next", ".npm-cache", "node_modules"} CONNECTOR_COMPLETION_MAX_BYTES = 40_000_000 CONNECTOR_STDOUT_MAX_BYTES = 4_000_000 CONNECTOR_LAST_MESSAGE_MAX_BYTES = 256_000 CONNECTOR_CODEX_TOKEN_EVIDENCE_SCHEMA = "sisaga.independent_ab.codex_token_events.v1" CODEX_CONFIG_OVERRIDES = { 'approval_policy="never"', 'sandbox_mode="danger-full-access"', "sandbox_workspace_write.network_access=true", } CODEX_REASONING_EFFORTS = {"low", "medium", "high", "xhigh"} # claude CLI (>= 2.1.198) --effort levels; mirrors the cockpit whitelist. CLAUDE_REASONING_EFFORTS = {"low", "medium", "high", "xhigh", "max"} def normalized_server(value: str) -> str: result = str(value or "").strip().rstrip("/") parsed = urllib.parse.urlsplit(result) hostname = str(parsed.hostname or "").lower() if parsed.scheme == "https" and hostname: return result if parsed.scheme == "http" and hostname in {"127.0.0.1", "localhost", "::1"}: return result raise ValueError("ABCD connector server must use HTTPS; plain HTTP is allowed only for loopback") def read_connector_states() -> dict: try: data = json.loads(CONNECTOR_STATE_PATH.read_text(encoding="utf-8")) except (OSError, ValueError): data = {} servers = data.get("servers") if isinstance(data.get("servers"), dict) else {} return {"schema": "sisaga.abcd.connector.local_state.v1", "servers": servers} def load_connector_state(server: str = "") -> dict: states = read_connector_states()["servers"] key = normalized_server(server) if str(server or "").strip() else "" if key: row = states.get(key) return row if isinstance(row, dict) else {} rows = [row for row in states.values() if isinstance(row, dict)] return rows[-1] if rows else {} def save_connector_state(server: str, connector_id: str, token: str, name: str = "") -> dict: key = normalized_server(server) if not key or not connector_id or not token: raise ValueError("server, connector_id, and token are required") data = read_connector_states() row = { "server": key, "connector_id": str(connector_id), "token": str(token), "name": str(name or ""), "saved_at_epoch": int(time.time()), } data["servers"][key] = row CONNECTOR_STATE_PATH.parent.mkdir(parents=True, exist_ok=True) try: os.chmod(CONNECTOR_STATE_PATH.parent, stat.S_IRUSR | stat.S_IWUSR | stat.S_IXUSR) except OSError: pass temp_path = CONNECTOR_STATE_PATH.with_suffix(".tmp") temp_path.write_text(json.dumps(data, ensure_ascii=False, indent=2) + "\n", encoding="utf-8") try: os.chmod(temp_path, stat.S_IRUSR | stat.S_IWUSR) except OSError: pass temp_path.replace(CONNECTOR_STATE_PATH) return row def resolve_saved_connection(server: str, token: str, name: str = "") -> tuple[str, str, str]: saved = load_connector_state(server) resolved_server = normalized_server(server or saved.get("server")) resolved_token = str(token or saved.get("token") or "").strip() resolved_name = str(name or saved.get("name") or "").strip() if not resolved_server or not resolved_token: raise RuntimeError("connector is not paired; run the pair command shown by the ABCD page") return resolved_server, resolved_token, resolved_name def post_json(url: str, payload: dict, token: str = "", timeout: int = 30) -> dict: data = json.dumps(payload, ensure_ascii=False).encode("utf-8") request = urllib.request.Request(url, data=data, method="POST") request.add_header("Content-Type", "application/json; charset=utf-8") if token: request.add_header("Authorization", f"Bearer {token}") try: with urllib.request.urlopen(request, timeout=timeout) as response: return json.loads(response.read().decode("utf-8")) except urllib.error.HTTPError as error: body = error.read().decode("utf-8", errors="replace") raise RuntimeError(f"HTTP {error.code}: {body}") from error def get_json(url: str, token: str = "") -> dict: request = urllib.request.Request(url, method="GET") if token: request.add_header("Authorization", f"Bearer {token}") try: with urllib.request.urlopen(request, timeout=35) as response: return json.loads(response.read().decode("utf-8")) except urllib.error.HTTPError as error: if error.code == 404: return {"ok": False, "empty": True, "error": "jobs endpoint unavailable"} body = error.read().decode("utf-8", errors="replace") raise RuntimeError(f"HTTP {error.code}: {body}") from error def cli_subprocess_command(parts: list[str]) -> list[str]: if not parts: return [] suffix = pathlib.Path(str(parts[0])).suffix.lower() if os.name == "nt" and suffix in {".cmd", ".bat"}: return [os.environ.get("COMSPEC") or "cmd.exe", "/d", "/s", "/c", "call", *parts] if os.name == "nt" and suffix == ".ps1": return ["powershell.exe", "-NoProfile", "-ExecutionPolicy", "Bypass", "-File", *parts] return parts def probe_cli_capability(command: str, args: list[str]) -> dict: path = shutil.which(command) status_command = "claude auth status" if command == "claude" else "codex login status" if not path: return { "status": "cli_missing", "reason": f"{command} CLI was not found in PATH", "action": "Install the CLI or add it to PATH, then restart Connector", } try: completed = subprocess.run( cli_subprocess_command([path, *args]), stdin=subprocess.DEVNULL, stdout=subprocess.PIPE, stderr=subprocess.PIPE, text=True, encoding="utf-8", errors="replace", timeout=12, check=False, ) except subprocess.TimeoutExpired: return { "status": "probe_timeout", "reason": f"{command} login status check timed out", "action": f"Run {status_command} manually and restart Connector", } except Exception as error: return { "status": "probe_failed", "reason": f"{command} login status check failed ({type(error).__name__})", "action": f"Run {status_command} manually and restart Connector", } output = "\n".join([str(completed.stdout or ""), str(completed.stderr or "")]).strip() lowered = output.lower() logged_in = None try: parsed = json.loads(str(completed.stdout or "{}")) if isinstance(parsed, dict) and isinstance(parsed.get("loggedIn"), bool): logged_in = bool(parsed["loggedIn"]) except (TypeError, ValueError): pass if logged_in is False or "not logged in" in lowered or "loggedin\": false" in lowered: login_command = "claude auth login" if command == "claude" else "codex login" return { "status": "not_authenticated", "reason": f"{command} CLI is installed but not logged in", "action": login_command, } if completed.returncode == 0: return {"status": "ready", "reason": "", "action": ""} return { "status": "probe_failed", "reason": f"{status_command} exited with code {int(completed.returncode)}", "action": f"Run {status_command} manually, fix the reported error, then restart Connector", } def cli_available(command: str, args: list[str]) -> bool: return probe_cli_capability(command, args).get("status") == "ready" def detect_capabilities_with_diagnostics() -> tuple[list[str], list[str], dict[str, dict]]: providers: list[str] = [] models: list[str] = [] diagnostics = { "codex": probe_cli_capability("codex", ["login", "status"]), "claude": probe_cli_capability("claude", ["auth", "status"]), } if diagnostics["codex"].get("status") == "ready": providers.append("codex") models.extend(["gpt-5.5", "gpt-5.6-sol", "gpt-5.6-terra", "gpt-5.6-luna"]) if diagnostics["claude"].get("status") == "ready": providers.append("claude") models.extend(["claude-opus-4-8", "claude-sonnet-5", "claude-fable-5"]) return providers, models, diagnostics def detect_capabilities() -> tuple[list[str], list[str]]: providers, models, _diagnostics = detect_capabilities_with_diagnostics() return providers, models def machine_hint() -> str: return f"{platform.node()} ยท {platform.system()} {platform.release()}" def pair(args: argparse.Namespace) -> int: server = normalized_server(args.server) previous = load_connector_state(server) previous_connector_id = str(previous.get("connector_id") or "").strip() previous_token = str(previous.get("token") or "").strip() providers, models, diagnostics = detect_capabilities_with_diagnostics() payload = { "pair_code": args.code, "display_name": args.name or machine_hint(), "machine_hint": machine_hint(), "providers": providers, "models": models, "capability_diagnostics": diagnostics, } if previous_connector_id: payload["previous_connector_id"] = previous_connector_id result = post_json( server + "/api/independent-ab/connectors/pair/claim", payload, token=previous_token, ) token = result.get("token") connector_id = result.get("connector_id") if not token or not connector_id: raise RuntimeError(f"pairing failed: {result}") save_connector_state(server, str(connector_id), str(token), args.name or machine_hint()) print(json.dumps({"paired": True, "connector_id": connector_id, "providers": providers, "models": models}, ensure_ascii=False)) if args.watch: args.server = server args.token = str(token) return run_loop(args) return 0 def heartbeat_loop(server: str, token: str, name: str = "") -> None: url = server.rstrip("/") + "/api/independent-ab/connectors/heartbeat" while True: providers, models, diagnostics = detect_capabilities_with_diagnostics() payload = { "display_name": name or machine_hint(), "machine_hint": machine_hint(), "providers": providers, "models": models, "capability_diagnostics": diagnostics, } result = post_json(url, payload, token=token) print(json.dumps({"heartbeat": result.get("ok") is True, "providers": providers, "models": models}, ensure_ascii=False), flush=True) time.sleep(30) def safe_extract_zip( archive_b64: str, target_dir: str, max_bytes: int = CONNECTOR_ARCHIVE_MAX_BYTES, max_expanded_bytes: int = CONNECTOR_ARCHIVE_EXPANDED_MAX_BYTES, ) -> None: raw = base64.b64decode(archive_b64.encode("ascii"), validate=True) if len(raw) > max_bytes: raise RuntimeError("job archive too large") with tempfile.NamedTemporaryFile(suffix=".zip", delete=False) as handle: handle.write(raw) archive_path = handle.name try: with zipfile.ZipFile(archive_path) as archive: total = 0 root = os.path.abspath(target_dir) for info in archive.infolist(): unix_mode = (int(info.external_attr) >> 16) & 0xFFFF if stat.S_ISLNK(unix_mode): raise RuntimeError(f"archive symlink is not allowed: {info.filename}") total += int(info.file_size or 0) if total > max_expanded_bytes: raise RuntimeError("job archive expanded size too large") dest = os.path.abspath(os.path.join(target_dir, info.filename)) if dest != root and not dest.startswith(root + os.sep): raise RuntimeError(f"unsafe archive path: {info.filename}") archive.extractall(target_dir) finally: try: os.unlink(archive_path) except OSError: pass def zip_dir_b64( root: str, max_bytes: int = CONNECTOR_ARCHIVE_MAX_BYTES, max_expanded_bytes: int = CONNECTOR_ARCHIVE_EXPANDED_MAX_BYTES, ) -> str: archive_path = os.path.join(tempfile.gettempdir(), f"sisaga-abcd-result-{int(time.time() * 1000)}.zip") try: total = 0 with zipfile.ZipFile(archive_path, "w", compression=zipfile.ZIP_DEFLATED) as archive: for current, dirs, files in os.walk(root): dirs[:] = [ item for item in dirs if item not in CONNECTOR_ARCHIVE_EXCLUDED_DIR_NAMES and not os.path.islink(os.path.join(current, item)) ] for name in files: path = os.path.join(current, name) if os.path.islink(path): continue total += os.path.getsize(path) if total > max_expanded_bytes: raise RuntimeError("result archive expanded size too large") archive.write(path, os.path.relpath(path, root)) with open(archive_path, "rb") as handle: raw = handle.read() if len(raw) > max_bytes: raise RuntimeError("result archive too large") return base64.b64encode(raw).decode("ascii") finally: try: os.unlink(archive_path) except OSError: pass def cleanup_connector_temp_dir(path: str, attempts: int = 4) -> bool: target = os.path.abspath(str(path or "")) temp_root = os.path.abspath(tempfile.gettempdir()) if os.path.dirname(target) != temp_root or not os.path.basename(target).startswith("sisaga-abcd-job-"): return False for attempt in range(max(1, attempts)): try: shutil.rmtree(target) return True except FileNotFoundError: return True except OSError: if attempt + 1 < max(1, attempts): time.sleep(attempt + 1) return False def build_job_command( *, cli: str, provider: str, model: str, workdir: str, last_message_path: str, policy: dict, ) -> list[str]: if str(policy.get("schema") or "") != CONNECTOR_EXECUTION_SCHEMA: raise RuntimeError("missing or unsupported connector execution policy") if str(policy.get("provider") or "") != provider: raise RuntimeError("connector execution policy provider mismatch") if provider == "codex": overrides = [str(item) for item in policy.get("config_overrides", [])] if set(overrides) != CODEX_CONFIG_OVERRIDES: raise RuntimeError("connector execution policy config overrides mismatch") reasoning_effort = str(policy.get("reasoning_effort") or "").strip().lower() if reasoning_effort and reasoning_effort not in CODEX_REASONING_EFFORTS: raise RuntimeError("connector execution policy reasoning effort is invalid") if not all(bool(policy.get(key)) for key in ("ignore_rules", "clean_codex_home", "disable_memories", "dangerous_runtime")): raise RuntimeError("connector execution policy is weaker than the ABCD isolation contract") return [ cli, "exec", "--skip-git-repo-check", "--cd", workdir, "--add-dir", workdir, *( ("--model", model) if model else () ), "--dangerously-bypass-approvals-and-sandbox", "--ignore-rules", "--disable", "memories", *( ("--disable", "plugins") if policy.get("disable_plugins") else () ), *( ("--disable", "apps") if policy.get("disable_apps") else () ), *( ("--disable", "fast_mode") if policy.get("standard_speed") else () ), *( ("-c", f'model_reasoning_effort="{reasoning_effort}"') if reasoning_effort else () ), *[part for override in overrides for part in ("-c", override)], "--json", "-o", last_message_path, "-", ] if provider == "claude": session_id = str(policy.get("session_id") or "").strip() if not session_id: raise RuntimeError("claude connector execution policy is missing session_id") reasoning_effort = str(policy.get("reasoning_effort") or "").strip().lower() if reasoning_effort and reasoning_effort not in CLAUDE_REASONING_EFFORTS: raise RuntimeError("connector execution policy reasoning effort is invalid") runtime_guard_args = [str(item) for item in policy.get("runtime_guard_args", [])] # Guard lanes already carry --effort inside runtime_guard_args (the # cockpit bakes the user override in); append it here only for lanes # without one so the flag never appears twice. effort_in_guard = "--effort" in runtime_guard_args return [ cli, "-p", "--output-format", "stream-json", "--verbose", "--add-dir", workdir, "--permission-mode", "bypassPermissions", *runtime_guard_args, "--session-id", session_id, *( ("--model", model) if model else () ), *( ("--effort", reasoning_effort) if reasoning_effort and not effort_in_guard else () ), ] raise RuntimeError("unsupported provider") def prepare_sterile_codex_home(target: str) -> None: target_path = pathlib.Path(target) target_path.mkdir(parents=True, exist_ok=True) source_home = pathlib.Path.home() / ".codex" for name in ("auth.json", "installation_id", "cap_sid"): source = source_home / name if source.is_file() and not source.is_symlink(): shutil.copy2(source, target_path / name) (target_path / "config.toml").write_text( 'approval_policy = "never"\n' 'sandbox_mode = "danger-full-access"\n\n' '[features]\n' 'memories = false\n', encoding="utf-8", ) def connector_result_ok(provider: str, returncode: int, stdout_text: str) -> dict: events: list[dict] = [] for raw_line in str(stdout_text or "").splitlines(): try: item = json.loads(raw_line) except (TypeError, ValueError): continue if isinstance(item, dict): events.append(item) if int(returncode) != 0: return {"ok": False, "error": f"{provider} CLI exited with code {returncode}"} if provider == "codex": if any(str(item.get("type") or "") == "turn.failed" or item.get("is_error") for item in events): return {"ok": False, "error": "Codex emitted a failure terminal event"} if any(str(item.get("type") or "") == "turn.completed" for item in events): return {"ok": True, "error": ""} return {"ok": False, "error": "Codex exited without turn.completed"} results = [item for item in events if str(item.get("type") or "") == "result"] if not results: return {"ok": False, "error": "Claude exited without a result event"} final = results[-1] if final.get("is_error") or not str(final.get("session_id") or "").strip(): return {"ok": False, "error": str(final.get("result") or "Claude result is_error=true")} return {"ok": True, "error": ""} def codex_thread_id_from_stdout(stdout_text: str) -> str: for raw_line in str(stdout_text or "").splitlines(): try: event = json.loads(raw_line) except (TypeError, ValueError): continue if isinstance(event, dict) and str(event.get("type") or "") == "thread.started": thread_id = str(event.get("thread_id") or "").strip() if thread_id: return thread_id return "" def codex_rollout_path(codex_home: str, thread_id: str) -> pathlib.Path | None: home = pathlib.Path(codex_home) state_db = home / "state_5.sqlite" if state_db.is_file(): try: uri = f"file:{state_db.as_posix()}?mode=ro" connection = sqlite3.connect(uri, uri=True, timeout=1.0) try: row = connection.execute( "select rollout_path from threads where id=? limit 1", (thread_id,), ).fetchone() finally: connection.close() if row and str(row[0] or "").strip(): candidate = pathlib.Path(str(row[0]).strip()) if not candidate.is_absolute(): candidate = home / candidate if candidate.is_file(): return candidate except sqlite3.Error: pass sessions = home / "sessions" if not sessions.is_dir(): return None candidates = sorted( sessions.rglob("*.jsonl"), key=lambda path: path.stat().st_mtime if path.is_file() else 0, reverse=True, ) for candidate in candidates: try: with candidate.open("r", encoding="utf-8", errors="replace") as handle: for index, raw_line in enumerate(handle): if index >= 64: break try: item = json.loads(raw_line) except (TypeError, ValueError): continue payload = item.get("payload") if isinstance(item, dict) else None if ( isinstance(payload, dict) and str(item.get("type") or "") == "session_meta" and str(payload.get("id") or "") == thread_id ): return candidate except OSError: continue return None def read_codex_rollout_token_evidence( codex_home: str, stdout_text: str, *, job_id: str, attempt_id: str, ) -> dict: thread_id = codex_thread_id_from_stdout(stdout_text) evidence = { "schema": CONNECTOR_CODEX_TOKEN_EVIDENCE_SCHEMA, "exact": False, "source": "connector_codex_rollout_token_count", "job_id": str(job_id or ""), "attempt_id": str(attempt_id or ""), "connector_thread_id": thread_id, "token_events": [], "token_event_count": 0, "turn_token_total": 0, } if not thread_id: evidence["error"] = "Codex thread.started is missing" return evidence rollout_path = codex_rollout_path(codex_home, thread_id) if rollout_path is None: evidence["error"] = "Codex rollout for the current connector job was not found" return evidence token_events: list[dict] = [] previous_total = 0 try: with rollout_path.open("r", encoding="utf-8", errors="replace") as handle: for raw_line in handle: try: item = json.loads(raw_line) except (TypeError, ValueError): continue if not isinstance(item, dict) or str(item.get("type") or "") != "event_msg": continue payload = item.get("payload") if isinstance(item.get("payload"), dict) else {} if str(payload.get("type") or "") != "token_count": continue info = payload.get("info") if isinstance(payload.get("info"), dict) else {} last_usage = info.get("last_token_usage") if isinstance(info.get("last_token_usage"), dict) else {} total_usage = info.get("total_token_usage") if isinstance(info.get("total_token_usage"), dict) else {} last_tokens = last_usage.get("total_tokens") total_tokens = total_usage.get("total_tokens") if not isinstance(last_tokens, int) or isinstance(last_tokens, bool) or last_tokens <= 0: continue if not isinstance(total_tokens, int) or isinstance(total_tokens, bool) or total_tokens <= 0: total_tokens = previous_total + last_tokens if total_tokens == previous_total and token_events: continue if total_tokens < previous_total: evidence["error"] = "Codex rollout token totals are not monotonic" return evidence usage = { key: value for key, value in last_usage.items() if key in { "input_tokens", "cached_input_tokens", "output_tokens", "reasoning_output_tokens", "total_tokens", } and isinstance(value, int) and not isinstance(value, bool) } previous_total = total_tokens token_events.append({ "kind": "token_count", "timestamp": str(item.get("timestamp") or ""), "source": "connector_codex_rollout_token_count", "sequence": len(token_events) + 1, "last_tokens": last_tokens, "total_tokens": total_tokens, "usage": usage, }) except OSError as error: evidence["error"] = f"Codex rollout could not be read: {error}" return evidence if not token_events: evidence["error"] = "Codex rollout contains no token_count events" return evidence evidence.update({ "exact": True, "rollout_file": rollout_path.name, "token_events": token_events, "token_event_count": len(token_events), "turn_token_total": token_events[-1]["total_tokens"], "error": "", }) return evidence def codex_terminal_token_evidence( stdout_text: str, *, job_id: str, attempt_id: str, ) -> dict: thread_id = codex_thread_id_from_stdout(stdout_text) evidence = { "schema": CONNECTOR_CODEX_TOKEN_EVIDENCE_SCHEMA, "exact": False, "source": "connector_codex_stdout_turn_completed", "job_id": str(job_id or ""), "attempt_id": str(attempt_id or ""), "connector_thread_id": thread_id, "token_events": [], "token_event_count": 0, "turn_token_total": 0, } if not thread_id: evidence["error"] = "Codex thread.started is missing" return evidence completed: dict | None = None for raw_line in stdout_text.splitlines(): try: item = json.loads(raw_line) except (TypeError, ValueError): continue if isinstance(item, dict) and str(item.get("type") or "") == "turn.completed": completed = item usage_raw = completed.get("usage") if isinstance(completed, dict) else None if not isinstance(usage_raw, dict): evidence["error"] = "Codex turn.completed usage is missing" return evidence input_tokens = usage_raw.get("input_tokens") output_tokens = usage_raw.get("output_tokens") if ( not isinstance(input_tokens, int) or isinstance(input_tokens, bool) or input_tokens < 0 or not isinstance(output_tokens, int) or isinstance(output_tokens, bool) or output_tokens < 0 ): evidence["error"] = "Codex turn.completed usage is invalid" return evidence total_tokens = input_tokens + output_tokens if total_tokens <= 0: evidence["error"] = "Codex turn.completed usage is empty" return evidence usage = { "input_tokens": input_tokens, "output_tokens": output_tokens, "total_tokens": total_tokens, } cached_input_tokens = usage_raw.get("cached_input_tokens") if isinstance(cached_input_tokens, int) and not isinstance(cached_input_tokens, bool): usage["cached_input_tokens"] = cached_input_tokens evidence.update({ "exact": True, "token_events": [{ "kind": "token_count", "timestamp": str(completed.get("timestamp") or ""), "source": "connector_codex_stdout_turn_completed", "sequence": 1, "last_tokens": total_tokens, "total_tokens": total_tokens, "usage": usage, }], "token_event_count": 1, "turn_token_total": total_tokens, "error": "", }) return evidence def read_codex_token_evidence( codex_home: str, stdout_text: str, *, job_id: str, attempt_id: str, rollout_retry_attempts: int = 1, rollout_retry_delay: float = 0.0, ) -> dict: attempts = max(1, int(rollout_retry_attempts or 1)) rollout_evidence: dict = {} for attempt in range(attempts): rollout_evidence = read_codex_rollout_token_evidence( codex_home, stdout_text, job_id=job_id, attempt_id=attempt_id, ) if rollout_evidence.get("exact"): return rollout_evidence if attempt + 1 < attempts and rollout_retry_delay > 0: time.sleep(rollout_retry_delay) terminal_evidence = codex_terminal_token_evidence( stdout_text, job_id=job_id, attempt_id=attempt_id, ) return terminal_evidence if terminal_evidence.get("exact") else rollout_evidence def truncate_jsonl(value: str, max_bytes: int) -> str: raw = str(value or "").encode("utf-8") if len(raw) <= max_bytes: return str(value or "") head_budget = max(512, min(max_bytes // 4, 512_000)) tail_budget = max(512, max_bytes - head_budget - 160) head = raw[:head_budget].decode("utf-8", errors="ignore") tail = raw[-tail_budget:].decode("utf-8", errors="ignore") if "\n" in head: head = head.rsplit("\n", 1)[0] if "\n" in tail: tail = tail.split("\n", 1)[-1] marker = json.dumps({"type": "sisaga.connector.output_truncated"}, ensure_ascii=False) return head.rstrip() + "\n" + marker + "\n" + tail.lstrip() def truncate_text_tail(value: str, max_bytes: int) -> str: raw = str(value or "").encode("utf-8") if len(raw) <= max_bytes: return str(value or "") return raw[-max_bytes:].decode("utf-8", errors="ignore") def fit_completion_payload(payload: dict, max_bytes: int = CONNECTOR_COMPLETION_MAX_BYTES) -> dict: fitted = dict(payload) stdout_limit = min(CONNECTOR_STDOUT_MAX_BYTES, max(2_048, max_bytes // 8)) last_limit = min(CONNECTOR_LAST_MESSAGE_MAX_BYTES, max(1_024, max_bytes // 32)) original_stdout = str(fitted.get("stdout_jsonl") or "") original_last = str(fitted.get("last_message") or "") fitted["stdout_jsonl"] = truncate_jsonl(original_stdout, stdout_limit) fitted["last_message"] = truncate_text_tail(original_last, last_limit) while len(json.dumps(fitted, ensure_ascii=False).encode("utf-8")) > max_bytes and stdout_limit > 2_048: stdout_limit = max(2_048, stdout_limit // 2) fitted["stdout_jsonl"] = truncate_jsonl(original_stdout, stdout_limit) while len(json.dumps(fitted, ensure_ascii=False).encode("utf-8")) > max_bytes and last_limit > 1_024: last_limit = max(1_024, last_limit // 2) fitted["last_message"] = truncate_text_tail(original_last, last_limit) if len(json.dumps(fitted, ensure_ascii=False).encode("utf-8")) > max_bytes: raise RuntimeError("connector completion payload exceeds the safe request budget") return fitted def connector_prompt_for_workdir(prompt: str, server_workdir: str) -> str: mapped = str(prompt or "") source = str(server_workdir or "").strip() if source: aliases = {source, source.replace("\\", "/"), source.replace("/", "\\")} for alias in sorted((value for value in aliases if value), key=len, reverse=True): mapped = mapped.replace(alias, ".") transport_note = ( "SISAGA ABCD Connector runtime path mapping: the process current working directory is the " "assigned product directory. Write every product and evidence file under the current working " "directory only; any server-side absolute lane path in the request maps to this directory." ) return f"{transport_note}\n\n{mapped}" def run_job(job: dict) -> dict: provider = str(job.get("provider") or "").strip().lower() if provider not in {"codex", "claude"}: raise RuntimeError("unsupported provider") model = str(job.get("model") or "").strip() prompt = connector_prompt_for_workdir( str(job.get("prompt") or ""), str(job.get("lane_workdir") or ""), ) if not prompt.strip(): raise RuntimeError("empty job prompt") command_name = "codex" if provider == "codex" else "claude" cli = shutil.which(command_name) if not cli: raise RuntimeError(f"{command_name} cli not found") temp_dir = tempfile.mkdtemp(prefix="sisaga-abcd-job-") codex_home = "" try: workdir = os.path.join(temp_dir, "work") os.makedirs(workdir, exist_ok=True) if job.get("workdir_archive_b64"): safe_extract_zip(str(job["workdir_archive_b64"]), workdir) stdout_path = os.path.join(temp_dir, "stdout.jsonl") last_message_path = os.path.join(temp_dir, "last-message.md") policy = job.get("execution_policy") if isinstance(job.get("execution_policy"), dict) else {} env = os.environ.copy() if provider == "codex": codex_home = os.path.join(temp_dir, "codex-home") prepare_sterile_codex_home(codex_home) for key in list(env): if key.startswith("CODEX_"): env.pop(key, None) env["CODEX_HOME"] = codex_home command = build_job_command( cli=cli, provider=provider, model=model, workdir=workdir, last_message_path=last_message_path, policy=policy, ) command = cli_subprocess_command(command) with open(stdout_path, "wb") as out: proc = subprocess.Popen(command, cwd=workdir, stdin=subprocess.PIPE, stdout=out, stderr=subprocess.STDOUT, env=env) assert proc.stdin is not None proc.stdin.write(prompt.encode("utf-8")) proc.stdin.close() proc.wait() stdout_text = open(stdout_path, "r", encoding="utf-8", errors="replace").read() last_message = "" if os.path.exists(last_message_path): last_message = open(last_message_path, "r", encoding="utf-8", errors="replace").read() terminal = connector_result_ok(provider, int(proc.returncode or 0), stdout_text) token_evidence = ( read_codex_token_evidence( codex_home, stdout_text, job_id=str(job.get("job_id") or ""), attempt_id=str(job.get("attempt_id") or ""), rollout_retry_attempts=8, rollout_retry_delay=0.25, ) if provider == "codex" else {} ) if provider == "claude" and not last_message: for raw_line in reversed(stdout_text.splitlines()): try: item = json.loads(raw_line) except (TypeError, ValueError): continue if isinstance(item, dict) and str(item.get("type") or "") == "result": last_message = str(item.get("result") or "") break return { "ok": bool(terminal["ok"]), "job_id": job.get("job_id"), "attempt_id": job.get("attempt_id"), "returncode": proc.returncode, "stdout_jsonl": stdout_text, "last_message": last_message, "token_evidence": token_evidence, "result_archive_b64": zip_dir_b64(workdir), "error": str(terminal.get("error") or ""), } finally: cleanup_connector_temp_dir(temp_dir) def active_job_heartbeat_loop( server: str, token: str, name: str, job: dict, stop: threading.Event, ) -> None: heartbeat_url = server + "/api/independent-ab/connectors/heartbeat" while not stop.wait(20): try: providers, models, diagnostics = detect_capabilities_with_diagnostics() post_json(heartbeat_url, { "display_name": name or machine_hint(), "machine_hint": machine_hint(), "providers": providers, "models": models, "capability_diagnostics": diagnostics, "active_job_id": job.get("job_id"), "attempt_id": job.get("attempt_id"), }, token=token) except Exception: continue def post_completion_with_retry(server: str, token: str, result: dict, attempts: int = 3) -> dict: payload = fit_completion_payload(result) complete_url = server + "/api/independent-ab/connectors/jobs/complete" last_error: Exception | None = None for attempt in range(max(1, attempts)): try: return post_json(complete_url, payload, token=token, timeout=180) except Exception as error: # noqa: BLE001 last_error = error if attempt + 1 < attempts: time.sleep(2 + attempt * 3) raise RuntimeError(f"connector completion failed after {attempts} attempts: {last_error}") def execute_connector_job(server: str, token: str, name: str, job: dict) -> dict: lease_stop = threading.Event() lease_thread = threading.Thread( target=active_job_heartbeat_loop, args=(server, token, name, job, lease_stop), name=f"sisaga-abcd-job-lease-{job.get('job_id')}", daemon=True, ) lease_thread.start() try: try: result = run_job(job) except Exception as error: # noqa: BLE001 result = { "ok": False, "job_id": job.get("job_id"), "attempt_id": job.get("attempt_id"), "error": str(error), } return post_completion_with_retry(server, token, result) finally: lease_stop.set() lease_thread.join(timeout=5) def connector_parallel_workers(args: argparse.Namespace) -> int: try: requested = int(getattr(args, "parallel", 4) or 0) except (TypeError, ValueError): requested = 4 return max(1, min(8, requested)) def run_loop(args: argparse.Namespace) -> int: server, token, name = resolve_saved_connection(args.server, args.token, args.name) heartbeat_url = server + "/api/independent-ab/connectors/heartbeat" jobs_url = server + "/api/independent-ab/connectors/jobs/next" workers = connector_parallel_workers(args) last_heartbeat = 0.0 with concurrent.futures.ThreadPoolExecutor(max_workers=workers, thread_name_prefix="sisaga-abcd-lane") as pool: active: set[concurrent.futures.Future] = set() while True: for future in list(active): if not future.done(): continue active.remove(future) try: completed = future.result() print(json.dumps({"job_complete": True, **completed}, ensure_ascii=False), flush=True) except Exception as error: # noqa: BLE001 print(json.dumps({"job_complete": False, "error": str(error)}, ensure_ascii=False), flush=True) now = time.monotonic() if now - last_heartbeat >= 25: try: providers, models, diagnostics = detect_capabilities_with_diagnostics() post_json(heartbeat_url, { "display_name": name or machine_hint(), "machine_hint": machine_hint(), "providers": providers, "models": models, "capability_diagnostics": diagnostics, }, token=token) last_heartbeat = now except Exception as error: # noqa: BLE001 print(json.dumps({"heartbeat": False, "error": str(error)}, ensure_ascii=False), flush=True) claimed = False while len(active) < workers: try: job_result = get_json(jobs_url, token=token) except Exception as error: # noqa: BLE001 print(json.dumps({"job_poll": False, "error": str(error)}, ensure_ascii=False), flush=True) break job = job_result.get("job") if isinstance(job_result.get("job"), dict) else None if not job: break active.add(pool.submit(execute_connector_job, server, token, name, job)) claimed = True if not claimed: time.sleep(1 if active else 5) def heartbeat(args: argparse.Namespace) -> int: server, token, name = resolve_saved_connection(args.server, args.token, args.name) heartbeat_loop(server, token, name) return 0 def main(argv: list[str] | None = None) -> int: parser = argparse.ArgumentParser(description="SISAGA ABCD local CLI connector") sub = parser.add_subparsers(dest="cmd", required=True) pair_parser = sub.add_parser("pair", help="Claim a web pairing code and start heartbeats") pair_parser.add_argument("--server", required=True, help="ABCD server URL, for example https://code.sisaga.xyz") pair_parser.add_argument("--code", required=True, help="Pairing code shown by the ABCD page") pair_parser.add_argument("--name", default="", help="Display name for this execution endpoint") pair_parser.add_argument("--watch", action="store_true", help="Keep sending heartbeats after pairing") pair_parser.add_argument("--parallel", type=int, default=4, help="Maximum concurrent ABCD lanes (default: 4)") pair_parser.set_defaults(func=pair) hb_parser = sub.add_parser("heartbeat", help="Send continuous heartbeats with an existing token") hb_parser.add_argument("--server", default="") hb_parser.add_argument("--token", default="") hb_parser.add_argument("--name", default="") hb_parser.set_defaults(func=heartbeat) run_parser = sub.add_parser("run", help="Keep this machine online and execute assigned ABCD jobs") run_parser.add_argument("--server", default="") run_parser.add_argument("--token", default="") run_parser.add_argument("--name", default="") run_parser.add_argument("--parallel", type=int, default=4) run_parser.set_defaults(func=run_loop) args = parser.parse_args(argv) return int(args.func(args) or 0) if __name__ == "__main__": raise SystemExit(main())