#!/usr/bin/env python3 """ giip_agent.py -- GIIP logical-machine agent client (giip skill "giip-agent", v1.0.0) Standard-library only (urllib.request, platform). No third-party dependencies, so it runs on hosts where installing packages is uncertain or disallowed. Safety contract (see SKILL.md for the authoritative copy): - Base URL: https://giipfaw.azurewebsites.net/api/giipApiSk2 - Auth: a single per-csn SK, read from the GIIP_API_KEY environment variable at run time and never written to disk, logs, or exception messages. It is sent as a `token` form field (NOT an HTTP header -- this endpoint is a different, older dispatcher than the one used by the giip-issue skill). - This endpoint (giipApiSk2) ALWAYS returns HTTP 200 for a request it can parse at all. There is no HTTP-level error signal for a bad SK, a permission problem, or "no work available" -- every one of those is expressed inside the JSON body's `data[0].RstVal` field. Treat a non-200 HTTP status as a transport-level failure only; always parse the body to learn the real outcome. - `register`/`heartbeat` is `AgentAutoRegister`. The FIRST call for a given (hostname, csn) pair creates a new lssn; every subsequent call with the SAME hostname is a heartbeat/update against that same lssn -- it never creates a second lssn. The hostname MUST be built as `-` (e.g. "Lowy-DP01-claude-code") so that each distinct AI tool on a host gets its own lssn. This script builds that hostname itself from --tool-slug; it does not accept a bare hostname override, specifically to make the convention hard to violate by accident. - `poll` is `CQEQueueGet`. `RstVal` of `0` or `404` means "checked, nothing queued right now" -- this is a normal, successful poll with no work, not an error, and this script exits 0 for it. Only a network failure, an unparseable response, or an RstVal outside {200, 201, 0, 404} is treated as an error (exit code 3 or 7). Poll at a reasonable interval (60s is what real giipAgent installs use via cron); this script does not loop internally -- each invocation makes exactly one poll. A tight busy-loop poll is prohibited: it will not get you work any sooner (the queue is populated on giip's own schedule) and will draw attention as abuse. - Passing an `--lssn` that does not belong to your SK's csn does NOT error on `poll` -- the underlying query silently finds no matching queue row and returns RstVal 404 ("no queue"), identical to the normal empty-queue case. There is no way to distinguish "wrong lssn" from "no work yet" from the response alone; if you are unsure an lssn is correct, re-run `register` (a heartbeat) first and confirm the returned lssn matches what you intended to poll. - `report` is `KVSPut` with kFactor="cqeresult". Unlike register/poll, a `report` against an lssn your SK's key group (cgsn) does not own returns a clean error (`RstVal` 411, "Server not found in tLSvr") rather than a silent no-op -- report this to the caller as a failure (exit code 5), do not retry it as a heartbeat or anything else. - No auto-retry: on a failed or ambiguous call (timeout, unclear response), never automatically resend it. Report a clear exit code and error message with no secret material in it. Exit codes: 0 success (including poll's normal "no work queued" outcome) 1 usage / argument error 2 GIIP_API_KEY environment variable is not set 3 HTTP / network error (includes timeout) talking to the API 4 register: unexpected RstVal (not 200/201) from AgentAutoRegister 5 report: KVSPut rejected the write (e.g. RstVal 411, lssn not owned by this SK's key group) 6 JSON parse error in the API response 7 poll: unexpected RstVal (not 200/201/0/404) from CQEQueueGet """ import argparse import json import os import platform import sys import urllib.error import urllib.parse import urllib.request API_URL = "https://giipfaw.azurewebsites.net/api/giipApiSk2" TIMEOUT_SECONDS = 30 RECOMMENDED_POLL_INTERVAL_SECONDS = 60 def _get_api_key(): """Read the SK from the environment. Never log or echo the value.""" key = os.environ.get("GIIP_API_KEY") if not key: _emit_and_exit( success=False, exit_code=2, error="GIIP_API_KEY environment variable is not set. " "Set it for this process only; do not persist it to a file.", ) return key def _emit_and_exit(success, exit_code, error=None, **fields): """Print the standard JSON result shape to stdout and exit.""" payload = {"success": success} payload.update(fields) if error is not None: payload["error"] = error print(json.dumps(payload, ensure_ascii=False)) sys.exit(exit_code) def _call(sk, text, jsondata_obj): """ Issue one POST to giipApiSk2. Returns the parsed JSON body (a dict). Never includes sk in any exception message. giipApiSk2 always responds HTTP 200 for any request it can route at all -- the real outcome is in the JSON body, not the HTTP status. """ jsondata = json.dumps(jsondata_obj, ensure_ascii=False) body = urllib.parse.urlencode({"text": text, "token": sk, "jsondata": jsondata}).encode( "utf-8" ) req = urllib.request.Request( API_URL, data=body, headers={"Content-Type": "application/x-www-form-urlencoded"}, method="POST", ) try: with urllib.request.urlopen(req, timeout=TIMEOUT_SECONDS) as resp: raw = resp.read().decode("utf-8", errors="replace") except urllib.error.HTTPError as e: raw = e.read().decode("utf-8", errors="replace") if e.fp else "" except urllib.error.URLError as e: _emit_and_exit( success=False, exit_code=3, error="Network error contacting GIIP API: {}".format(e.reason), ) except TimeoutError: _emit_and_exit( success=False, exit_code=3, error="Timed out after {}s contacting GIIP API. Do not " "auto-retry.".format(TIMEOUT_SECONDS), ) try: parsed = json.loads(raw) if raw else {} except (ValueError, json.JSONDecodeError): _emit_and_exit( success=False, exit_code=6, error="Could not parse GIIP API response as JSON.", ) if "error" in parsed and "data" not in parsed: _emit_and_exit( success=False, exit_code=3, error="GIIP API reported an error: {}".format(parsed.get("error")), ) return parsed def _first_row(parsed): """Extract the first row of a giipApiSk2 `data` array response, or {}.""" data = parsed.get("data") or [] if not data: return {} return data[0] def _build_hostname(tool_slug, hostname_prefix): prefix = hostname_prefix or platform.node() or "unknown-host" return "{}-{}".format(prefix, tool_slug) def _default_system_info(): """Best-effort system info via stdlib `platform` only. Every field is optional server-side -- a missing/None value is fine.""" return { "os": "{} {}".format(platform.system(), platform.release()).strip(), "cpu": platform.processor() or platform.machine() or None, "agent_version": "giip-agent-skill/1.0.0", } def cmd_register(args): sk = _get_api_key() hostname = _build_hostname(args.tool_slug, args.hostname_prefix) info = _default_system_info() jsondata = { "hostname": hostname, "os": args.os or info["os"], "cpu": args.cpu or info["cpu"], "cpu_cores": args.cpu_cores, "memory_gb": args.memory_gb, "disk_gb": args.disk_gb, "agent_version": args.agent_version or info["agent_version"], "ipv4_global": args.ipv4_global, "ipv4_local": args.ipv4_local, } parsed = _call(sk, "AgentAutoRegister hostname jsondata", jsondata) row = _first_row(parsed) rst_val = str(row.get("RstVal", "")) if rst_val not in ("200", "201"): _emit_and_exit( success=False, exit_code=4, hostname=hostname, error="AgentAutoRegister returned RstVal={!r} (expected 200 or " "201): {}".format(rst_val, row.get("RstMsg") or row.get("Proc_MSG")), ) lssn = row.get("lssn") action = row.get("action", "unknown") _emit_and_exit( success=True, exit_code=0, hostname=hostname, lssn=lssn, action=action, ) def cmd_poll(args): sk = _get_api_key() hostname = _build_hostname(args.tool_slug, args.hostname_prefix) jsondata = { "lssn": args.lssn, "hostname": hostname, "os": args.os or _default_system_info()["os"], "op": "op", } parsed = _call(sk, "CQEQueueGet lssn hostname os op", jsondata) row = _first_row(parsed) rst_val = str(row.get("RstVal", "")) if rst_val in ("0", "404"): _emit_and_exit(success=True, exit_code=0, lssn=args.lssn, has_work=False) if rst_val not in ("200", "201"): _emit_and_exit( success=False, exit_code=7, lssn=args.lssn, error="CQEQueueGet returned RstVal={!r} (expected 200, 201, 0, " "or 404): {}".format(rst_val, row.get("ProcName") or row.get("RstMsg")), ) _emit_and_exit( success=True, exit_code=0, lssn=args.lssn, has_work=True, mslsn=row.get("mslsn"), mssn=row.get("mssn"), script_type=row.get("script_type"), ms_body=row.get("ms_body"), ) def cmd_report(args): sk = _get_api_key() stdout_text = "" stderr_text = "" if args.stdout_file: with open(args.stdout_file, "r", encoding="utf-8", errors="replace") as f: stdout_text = f.read() if args.stderr_file: with open(args.stderr_file, "r", encoding="utf-8", errors="replace") as f: stderr_text = f.read() kvalue = { "mslsn": args.mslsn, "mssn": args.mssn, "lssn": args.lssn, "status": args.status, "exit_code": args.exit_code, "stdout": stdout_text, "stderr": stderr_text, } jsondata = { "kType": "lssn", "kKey": str(args.lssn), "kFactor": "cqeresult", "kValue": kvalue, } parsed = _call(sk, "KVSPut kType kKey kFactor", jsondata) row = _first_row(parsed) rst_val = str(row.get("RstVal", "")) if rst_val != "200": _emit_and_exit( success=False, exit_code=5, lssn=args.lssn, error="KVSPut rejected the result write, RstVal={!r}: {}".format( rst_val, row.get("RstMsg") ), ) _emit_and_exit(success=True, exit_code=0, lssn=args.lssn) def build_parser(): p = argparse.ArgumentParser( prog="giip_agent.py", description="GIIP logical-machine agent client (register/heartbeat, " "poll the CQE queue, report a result). One call per invocation -- " "this script never loops or daemonizes itself.", ) sub = p.add_subparsers(dest="command", required=True) def add_hostname_args(sp): sp.add_argument( "--tool-slug", type=str, required=True, help="Short identifier for this AI tool, e.g. 'claude-code', " "'codex', 'antigravity'. The registered hostname is always " "'-' -- never a bare hostname.", ) sp.add_argument( "--hostname-prefix", type=str, default=None, help="Physical/VM hostname to prefix. Defaults to this " "process's platform.node() if omitted.", ) p_reg = sub.add_parser( "register", help="Register this (host, tool) as a logical machine, or send a " "heartbeat if it is already registered (AgentAutoRegister).", ) add_hostname_args(p_reg) p_reg.add_argument("--os", type=str, default=None) p_reg.add_argument("--cpu", type=str, default=None) p_reg.add_argument("--cpu-cores", type=int, default=None) p_reg.add_argument("--memory-gb", type=int, default=None) p_reg.add_argument("--disk-gb", type=int, default=None) p_reg.add_argument("--agent-version", type=str, default=None) p_reg.add_argument("--ipv4-global", type=str, default=None) p_reg.add_argument("--ipv4-local", type=str, default=None) p_reg.set_defaults(func=cmd_register) p_hb = sub.add_parser( "heartbeat", help="Identical to 'register' -- a repeat call against an existing " "hostname updates that lssn instead of creating a new one.", ) add_hostname_args(p_hb) p_hb.add_argument("--os", type=str, default=None) p_hb.add_argument("--cpu", type=str, default=None) p_hb.add_argument("--cpu-cores", type=int, default=None) p_hb.add_argument("--memory-gb", type=int, default=None) p_hb.add_argument("--disk-gb", type=int, default=None) p_hb.add_argument("--agent-version", type=str, default=None) p_hb.add_argument("--ipv4-global", type=str, default=None) p_hb.add_argument("--ipv4-local", type=str, default=None) p_hb.set_defaults(func=cmd_register) p_poll = sub.add_parser( "poll", help="Check this lssn's queue exactly once (CQEQueueGet). " "RstVal 0/404 ('no work') is reported as success with " "has_work=false, not as an error.", ) p_poll.add_argument("--lssn", type=int, required=True) add_hostname_args(p_poll) p_poll.add_argument("--os", type=str, default=None) p_poll.set_defaults(func=cmd_poll) p_report = sub.add_parser( "report", help="Report the result of an executed queue item (KVSPut, " "kFactor=cqeresult). Only call this with an mslsn/mssn that a " "prior 'poll' actually returned.", ) p_report.add_argument("--lssn", type=int, required=True) p_report.add_argument("--mslsn", type=int, required=True) p_report.add_argument("--mssn", type=int, required=True) p_report.add_argument("--status", type=str, required=True) p_report.add_argument("--exit-code", type=int, required=True) p_report.add_argument("--stdout-file", type=str, default=None) p_report.add_argument("--stderr-file", type=str, default=None) p_report.set_defaults(func=cmd_report) return p def main(): parser = build_parser() args = parser.parse_args() args.func(args) if __name__ == "__main__": main()